feat(mq): relay messages from microservices to clients via request.client.notify

server/__init__.py

@@ -118,6 +118,7 @@
 
 from .asyncio_extensions import map_suppress, synchronizedmethod
 from .broadcast_service import BroadcastService
+from .client_message_queue_service import ClientMessageQueueService
 from .config import TRACE, config
 from .configuration_service import ConfigurationService
 from .core import Service, create_services
@@ -144,6 +145,7 @@
 
 __all__ = (
     "BroadcastService",
+    "ClientMessageQueueService",
     "ConfigurationService",
     "GameConnection",
     "GameService",

server/client_message_queue_service.py

@@ -0,0 +1,122 @@
+"""
+Forward RabbitMQ messages from trusted microservices to connected clients.
+
+# Wire contract
+Publishers post to the `MQ_EXCHANGE_NAME` topic exchange with routing key
+`request.client.notify`. Addressing lives in AMQP message headers:
+
+- `user-id` (int, optional): forward the body to the player with this id, if
+  connected to this lobby instance. If not connected, the message is logged
+  and acked.
+- `channel` (str, optional, reserved): future per-channel pub/sub. Currently
+  recognised but not yet implemented.
+- If neither header is set, the message is broadcast to every authenticated
+  client connected to this instance.
+
+The message body is a UTF-8 JSON object and is forwarded to the client
+verbatim. The lobby server does not validate or rewrap it; producers are
+trusted because the broker is reachable only from internal services.
+"""
+
+import json
+import logging
+from typing import TYPE_CHECKING, Any, ClassVar, Optional
+
+from aio_pika.abc import AbstractIncomingMessage, AbstractQueue
+
+from .config import config
+from .core import Service
+from .decorators import with_logger
+from .message_queue_service import MessageQueueService
+from .player_service import PlayerService
+
+if TYPE_CHECKING:
+    from server import ServerInstance
+
+
+CLIENT_NOTIFY_ROUTING_KEY = "request.client.notify"
+
+
+@with_logger
+class ClientMessageQueueService(Service):
+    """Consume `request.client.notify` messages and forward to local clients."""
+
+    _logger: ClassVar[logging.Logger]
+
+    def __init__(
+        self,
+        server: "ServerInstance",
+        message_queue_service: MessageQueueService,
+        player_service: PlayerService,
+    ):
+        """Wire dependencies; consumer is started in `initialize`."""
+        self.server = server
+        self.message_queue_service = message_queue_service
+        self.player_service = player_service
+        self._queue: Optional[AbstractQueue] = None
+        self._consumer_tag: Optional[str] = None
+
+    async def initialize(self) -> None:
+        result = await self.message_queue_service.declare_queue_and_consume(
+            exchange_name=config.MQ_EXCHANGE_NAME,
+            routing_key=CLIENT_NOTIFY_ROUTING_KEY,
+            callback=self._on_message,
+        )
+        if result is not None:
+            self._queue, self._consumer_tag = result
+
+    async def shutdown(self) -> None:
+        if self._queue is not None and self._consumer_tag is not None:
+            await self._queue.cancel(self._consumer_tag)
+        self._queue = None
+        self._consumer_tag = None
+
+    async def _on_message(self, message: AbstractIncomingMessage) -> None:
+        async with message.process(requeue=False):
+            try:
+                payload = json.loads(message.body)
+            except (ValueError, UnicodeDecodeError):
+                self._logger.warning(
+                    "Dropping client-notify message with non-JSON body"
+                )
+                return
+
+            if not isinstance(payload, dict):
+                self._logger.warning(
+                    "Dropping client-notify message: payload is not a JSON object"
+                )
+                return
+
+            headers = message.headers or {}
+            user_id = headers.get("user-id")
+            channel = headers.get("channel")
+
+            if user_id is not None:
+                self._dispatch_to_user(user_id, payload)
+            elif channel is not None:
+                self._logger.info(
+                    "client-notify channel %r received but channel routing is "
+                    "not yet implemented; dropping",
+                    channel,
+                )
+            else:
+                self.server.write_broadcast(payload)
+
+    def _dispatch_to_user(self, user_id: Any, payload: dict) -> None:
+        try:
+            player_id = int(user_id)
+        except (TypeError, ValueError):
+            self._logger.warning(
+                "Dropping client-notify message: invalid user-id %r", user_id
+            )
+            return
+
+        player = self.player_service[player_id]
+        if player is None:
+            self._logger.info(
+                "client-notify for user %s ignored: not connected here",
+                player_id,
+            )
+            return
+
+        player.write_message(payload)

server/message_queue_service.py

@@ -5,11 +5,17 @@
 import asyncio
 import json
 import logging
-from typing import ClassVar, Iterable, Optional
+from typing import Awaitable, Callable, ClassVar, Iterable, Optional
 
 import aio_pika
 from aio_pika import DeliveryMode, ExchangeType
-from aio_pika.abc import AbstractChannel, AbstractConnection, AbstractExchange
+from aio_pika.abc import (
+    AbstractChannel,
+    AbstractConnection,
+    AbstractExchange,
+    AbstractIncomingMessage,
+    AbstractQueue
+)
 from aio_pika.exceptions import ProbableAuthenticationError
 
 from .asyncio_extensions import synchronizedmethod
@@ -187,6 +193,49 @@ async def publish_many(
                     routing
                 )
 
+    async def declare_queue_and_consume(
+        self,
+        exchange_name: str,
+        routing_key: str,
+        callback: Callable[[AbstractIncomingMessage], Awaitable[None]],
+        queue_name: str = "",
+        exclusive: bool = True,
+        auto_delete: bool = True,
+        durable: bool = False,
+    ) -> Optional[tuple[AbstractQueue, str]]:
+        """
+        Declare a queue, bind it to an exchange with the given routing key, and
+        start consuming. Returns `(queue, consumer_tag)` so the caller can
+        cancel on shutdown. Returns None if the broker connection is not ready.
+        """
+        await self.initialize()
+        if not self._is_ready:
+            self._logger.warning(
+                "Not connected to RabbitMQ, unable to declare consumer queue."
+            )
+            return None
+
+        assert self._channel is not None
+
+        exchange = self._exchanges.get(exchange_name)
+        if exchange is None:
+            raise KeyError(f"Unknown exchange {exchange_name}.")
+
+        queue = await self._channel.declare_queue(
+            queue_name,
+            exclusive=exclusive,
+            auto_delete=auto_delete,
+            durable=durable,
+        )
+        await queue.bind(exchange, routing_key=routing_key)
+        consumer_tag = await queue.consume(callback)
+
+        self._logger.debug(
+            "Consuming from queue %r bound to %s/%s",
+            queue.name, exchange_name, routing_key,
+        )
+        return queue, consumer_tag
+
     @synchronizedmethod("initialization_lock")
     async def reconnect(self) -> None:
         self._is_ready = False