ServerKit/backend/app/queue_bus/service.py

"""High-level service for the ServerKit Queue Bus."""
import logging

from app.queue_bus.sqlalchemy_broker import SQLAlchemyBroker, QueueBusError
from app.utils.slug import unique_slug

logger = logging.getLogger(__name__)


class QueueBusService:
    """Thin wrapper around the configured broker.

    The service is the single entry point used by API routes, internal
    consumers, and the plugin SDK. Today it always uses SQLAlchemyBroker;
    in the future it can select a broker from config.
    """

    _broker = None

    @classmethod
    def broker(cls):
        if cls._broker is None:
            cls._broker = SQLAlchemyBroker()
        return cls._broker

    @classmethod
    def reset_broker(cls):
        """Reset the broker singleton. Useful for tests."""
        cls._broker = None

    # ------------------------------------------------------------------
    # Groups
    # ------------------------------------------------------------------
    @classmethod
    def create_group(cls, slug=None, name=None, description=None, owner_type='system', owner_id=None, config=None):
        base = slug or name
        if not base:
            raise QueueBusError('Either slug or name is required', 400)
        generated_slug = unique_slug(base, lambda s: cls.broker().get_group(s) is not None)
        return cls.broker().create_group(
            slug=generated_slug,
            name=name or generated_slug,
            description=description,
            owner_type=owner_type,
            owner_id=owner_id,
            config=config,
        )

    @classmethod
    def update_group(cls, slug, name=None, description=None, config=None):
        return cls.broker().update_group(slug, name, description, config)

    @classmethod
    def delete_group(cls, slug):
        return cls.broker().delete_group(slug)

    @classmethod
    def get_group(cls, slug):
        return cls.broker().get_group(slug)

    @classmethod
    def list_groups(cls, owner_type=None, owner_id=None, limit=100, offset=0):
        return cls.broker().list_groups(owner_type, owner_id, limit, offset)

    # ------------------------------------------------------------------
    # Queues
    # ------------------------------------------------------------------
    @classmethod
    def create_queue(cls, group_slug, slug=None, name=None, description=None, config=None):
        base = slug or name
        if not base:
            raise QueueBusError('Either slug or name is required', 400)
        generated_slug = unique_slug(base, lambda s: cls.broker().get_queue(group_slug, s) is not None)
        return cls.broker().create_queue(
            group_slug=group_slug,
            slug=generated_slug,
            name=name or generated_slug,
            description=description,
            config=config,
        )

    @classmethod
    def update_queue(cls, group_slug, slug, name=None, description=None, config=None):
        return cls.broker().update_queue(group_slug, slug, name, description, config)

    @classmethod
    def delete_queue(cls, group_slug, queue_slug):
        return cls.broker().delete_queue(group_slug, queue_slug)

    @classmethod
    def get_queue(cls, group_slug, queue_slug):
        return cls.broker().get_queue(group_slug, queue_slug)

    @classmethod
    def list_queues(cls, group_slug, limit=100, offset=0):
        return cls.broker().list_queues(group_slug, limit, offset)

    # ------------------------------------------------------------------
    # Messages
    # ------------------------------------------------------------------
    @classmethod
    def send(cls, group_slug, queue_slug, payload, priority=0, delay_ms=0, max_attempts=None):
        return cls.broker().send(
            group_slug=group_slug,
            queue_slug=queue_slug,
            payload=payload,
            priority=priority,
            delay_ms=delay_ms,
            max_attempts=max_attempts,
        )

    @classmethod
    def receive(cls, group_slug, queue_slug, visibility_timeout_ms=30000, max_messages=1):
        return cls.broker().receive(
            group_slug=group_slug,
            queue_slug=queue_slug,
            visibility_timeout_ms=visibility_timeout_ms,
            max_messages=max_messages,
        )

    @classmethod
    def recover_in_flight(cls):
        return cls.broker().recover_in_flight()

    @classmethod
    def complete(cls, group_slug, queue_slug, message_id):
        return cls.broker().complete(group_slug, queue_slug, message_id)

    @classmethod
    def fail(cls, group_slug, queue_slug, message_id, error_message=None, requeue=False):
        return cls.broker().fail(
            group_slug=group_slug,
            queue_slug=queue_slug,
            message_id=message_id,
            error_message=error_message,
            requeue=requeue,
        )

    @classmethod
    def requeue(cls, group_slug, queue_slug, message_id):
        return cls.broker().requeue(group_slug, queue_slug, message_id)

    @classmethod
    def delete_message(cls, group_slug, queue_slug, message_id):
        return cls.broker().delete_message(group_slug, queue_slug, message_id)

    @classmethod
    def get_message(cls, group_slug, queue_slug, message_id):
        return cls.broker().get_message(group_slug, queue_slug, message_id)

    @classmethod
    def list_messages(cls, group_slug, queue_slug, status=None, limit=100, offset=0):
        return cls.broker().list_messages(
            group_slug=group_slug,
            queue_slug=queue_slug,
            status=status,
            limit=limit,
            offset=offset,
        )

    @classmethod
    def get_stats(cls, group_slug=None, queue_slug=None):
        return cls.broker().get_stats(group_slug, queue_slug)

    @classmethod
    def ensure_queue(cls, group_slug, queue_slug, config=None):
        """Idempotently ensure a group and queue exist. Used by SDK/consumers."""
        group = cls.get_group(group_slug)
        if not group:
            cls.create_group(group_slug, name=group_slug.replace('-', ' ').title())
        queue = cls.get_queue(group_slug, queue_slug)
        if not queue:
            return cls.create_queue(group_slug, queue_slug, name=queue_slug.replace('-', ' ').title(), config=config)
        return queue