src/domain/interfaces/services/message_broker.py

"""
domain/interfaces/services/message_broker.py
─────────────────────────────────────────────
MessageBroker — abstract contract for async pub/sub messaging.

This interface deliberately does NOT mention RabbitMQ, Kafka, or Redis.
Swapping the broker is done by creating a new implementation class —
zero changes required in Domain or Application layers.
"""
from __future__ import annotations

from abc import ABC, abstractmethod
from typing import Any, Callable, Coroutine


# Type alias: an async callback that receives a message payload dict
MessageHandler = Callable[[dict[str, Any]], Coroutine[Any, Any, None]]


class MessageBroker(ABC):
    """
    Async pub/sub messaging contract.

    Lifecycle:
        await broker.connect()
        await broker.publish(...)
        await broker.consume(...)   # runs until cancelled
        await broker.disconnect()
    """

    @abstractmethod
    async def connect(self) -> None:
        """
        Establish a connection to the message broker.

        Should be idempotent — calling connect() on an already-connected
        broker should be a no-op (or re-use the existing connection).
        """
        ...

    @abstractmethod
    async def disconnect(self) -> None:
        """
        Gracefully close the broker connection and release resources.
        """
        ...

    @abstractmethod
    async def publish(
        self,
        queue_name: str,
        message: dict[str, Any],
        *,
        persistent: bool = True,
    ) -> None:
        """
        Publish a message to the specified queue.

        Args:
            queue_name:  Name of the destination queue/topic.
            message:     Payload to send — must be JSON-serialisable.
            persistent:  If ``True``, the broker should persist the message
                         to disk (survives broker restart). Default is True.
        """
        ...

    @abstractmethod
    async def consume(
        self,
        queue_name: str,
        handler: MessageHandler,
        *,
        auto_ack: bool = False,
    ) -> None:
        """
        Start consuming messages from the specified queue.

        This method blocks (runs indefinitely) until the task is cancelled.
        For each message received, ``handler(payload_dict)`` is awaited.

        Args:
            queue_name: Name of the source queue/topic.
            handler:    Async callable invoked with the deserialized message payload.
            auto_ack:   If ``True``, acknowledge messages automatically before
                        processing (at-most-once). If ``False`` (default),
                        ack only after handler completes successfully (at-least-once).
        """
        ...

    @abstractmethod
    async def is_connected(self) -> bool:
        """Return ``True`` if the broker connection is active."""
        ...