space_worker/worker.py

import logging
import time
from urllib.error import URLError

from .messages import ResponseBuilder
from .supabase import RealtimeConnection, SupabaseBroadcaster, extract_broadcast


logger = logging.getLogger("celery-space.worker")


class RealtimeWorker:
    def __init__(self, settings):
        self.settings = settings
        self.responses = ResponseBuilder(settings)
        self.broadcaster = SupabaseBroadcaster(settings)

    def run_forever(self):
        while True:
            try:
                logger.info("connecting supabase realtime url=%s", self.settings.supabase_url)
                with RealtimeConnection(self.settings) as connection:
                    self.consume(connection)
            except RuntimeError as exc:
                logger.error("%s; retrying in %s seconds", exc, self.settings.reconnect_seconds)
                time.sleep(self.settings.reconnect_seconds)
            except Exception:
                logger.exception(
                    "supabase realtime worker disconnected; retrying in %s seconds",
                    self.settings.reconnect_seconds,
                )
                time.sleep(self.settings.reconnect_seconds)

    def consume(self, connection):
        for message in connection.iter_messages():
            broadcast = extract_broadcast(message)
            if not broadcast:
                continue

            event, payload = broadcast
            if not self.settings.accepts_request_event(event):
                logger.info("ignored realtime broadcast event=%s accepted=%s", event, self.settings.request_events)
                continue
            if not isinstance(payload, dict):
                logger.warning("ignored non-object realtime payload event=%s", event)
                continue

            logger.info("received realtime broadcast event=%s payload_keys=%s", event, sorted(payload.keys()))
            self.handle_payload(payload)

    def handle_payload(self, payload):
        try:
            response = self.responses.build_success(payload)
        except Exception as exc:
            response = self.responses.build_error(payload, exc)

        reply_channel = self.responses.reply_channel_for(payload)
        try:
            self.broadcaster.send(
                reply_channel,
                self.settings.response_event,
                response,
                private=self.settings.response_private,
            )
        except URLError as exc:
            logger.warning("broadcast failed reply_channel=%s error=%s", reply_channel, exc)
            return

        logger.info(
            "processed realtime message user_id=%s reply_channel=%s correlation_id=%s",
            response.get("user_id"),
            reply_channel,
            response.get("correlation_id"),
        )