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,126 @@
+"""
+Consumes messages published by trusted microservices and forwards them to
+connected clients.
+
+# Wire contract
+Publishers post to the `MQ_EXCHANGE_NAME` topic exchange with routing key
+`client.push`. 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, 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_PUSH_ROUTING_KEY = "client.push"
+
+
+@with_logger
+class ClientMessageQueueService(Service):
+    """
+    Consumes `client.push` messages from RabbitMQ and forwards them to
+    connected clients on this lobby instance.
+    """
+
+    _logger: ClassVar[logging.Logger]
+
+    def __init__(
+        self,
+        server: "ServerInstance",
+        message_queue_service: MessageQueueService,
+        player_service: PlayerService,
+    ):
+        self.server = server
+        self.message_queue_service = message_queue_service
+        self.player_service = player_service
+        self._queue: Optional[AbstractQueue] = None
+
+    async def initialize(self) -> None:
+        self._queue = await self.message_queue_service.declare_queue_and_consume(
+            exchange_name=config.MQ_EXCHANGE_NAME,
+            routing_key=CLIENT_PUSH_ROUTING_KEY,
+            callback=self._on_message,
+        )
+
+    async def shutdown(self) -> None:
+        if self._queue is not None:
+            try:
+                await self._queue.cancel(self._queue.name)
+            except Exception:
+                self._logger.debug(
+                    "Error cancelling client-push consumer", exc_info=True
+                )
+            self._queue = 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-push message with non-JSON body"
+                )
+                return
+
+            if not isinstance(payload, dict):
+                self._logger.warning(
+                    "Dropping client-push 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-push 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, payload: dict) -> None:
+        try:
+            player_id = int(user_id)
+        except (TypeError, ValueError):
+            self._logger.warning(
+                "Dropping client-push message: invalid user-id %r", user_id
+            )
+            return
+
+        player = self.player_service[player_id]
+        if player is None:
+            self._logger.info(
+                "client-push 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[AbstractQueue]:
+        """
+        Declare a queue, bind it to an exchange with the given routing key, and
+        start consuming. Returns the queue 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)
+        await queue.consume(callback)
+
+        self._logger.debug(
+            "Consuming from queue %r bound to %s/%s",
+            queue.name, exchange_name, routing_key,
+        )
+        return queue
+
     @synchronizedmethod("initialization_lock")
     async def reconnect(self) -> None:
         self._is_ready = False

tests/unit_tests/test_client_message_queue_service.py

@@ -0,0 +1,166 @@
+import json
+from unittest import mock
+
+import pytest
+
+from server import ServerInstance
+from server.client_message_queue_service import (
+    CLIENT_PUSH_ROUTING_KEY,
+    ClientMessageQueueService
+)
+from server.config import config
+
+
+def make_incoming_message(body: bytes, headers: dict | None = None):
+    """
+    Build a stand-in for aio_pika's IncomingMessage with the bits the service
+    actually touches.
+    """
+    message = mock.Mock()
+    message.body = body
+    message.headers = headers
+
+    process_cm = mock.MagicMock()
+    process_cm.__aenter__ = mock.AsyncMock(return_value=None)
+    process_cm.__aexit__ = mock.AsyncMock(return_value=False)
+    message.process = mock.Mock(return_value=process_cm)
+    return message
+
+
+@pytest.fixture
+def server_instance():
+    return mock.create_autospec(ServerInstance)
+
+
+@pytest.fixture
+def fake_player_service():
+    """
+    Minimal stand-in for PlayerService supporting __getitem__/__setitem__.
+    """
+    class _FakePlayerService:
+        def __init__(self):
+            self._players = {}
+
+        def __getitem__(self, player_id):
+            return self._players.get(player_id)
+
+        def __setitem__(self, player_id, player):
+            self._players[player_id] = player
+
+    return _FakePlayerService()
+
+
+@pytest.fixture
+async def client_message_queue_service(server_instance, fake_player_service):
+    mq_service = mock.Mock()
+    mq_service.declare_queue_and_consume = mock.AsyncMock(return_value=None)
+    service = ClientMessageQueueService(
+        server=server_instance,
+        message_queue_service=mq_service,
+        player_service=fake_player_service,
+    )
+    await service.initialize()
+    yield service
+    await service.shutdown()
+
+
+async def test_initialize_declares_consumer(client_message_queue_service):
+    mq = client_message_queue_service.message_queue_service
+    mq.declare_queue_and_consume.assert_awaited_once()
+    kwargs = mq.declare_queue_and_consume.await_args.kwargs
+    assert kwargs["exchange_name"] == config.MQ_EXCHANGE_NAME
+    assert kwargs["routing_key"] == CLIENT_PUSH_ROUTING_KEY
+    assert kwargs["callback"] == client_message_queue_service._on_message
+
+
+async def test_dispatch_to_connected_user(
+    client_message_queue_service, fake_player_service
+):
+    player = mock.Mock()
+    player.write_message = mock.Mock()
+    fake_player_service[42] = player
+
+    payload = {"command": "notice", "text": "hi"}
+    msg = make_incoming_message(json.dumps(payload).encode(), {"user-id": 42})
+
+    await client_message_queue_service._on_message(msg)
+
+    player.write_message.assert_called_once_with(payload)
+    client_message_queue_service.server.write_broadcast.assert_not_called()
+
+
+async def test_dispatch_to_disconnected_user_is_dropped(
+    client_message_queue_service, caplog
+):
+    payload = {"command": "notice"}
+    msg = make_incoming_message(json.dumps(payload).encode(), {"user-id": 999})
+
+    await client_message_queue_service._on_message(msg)
+
+    client_message_queue_service.server.write_broadcast.assert_not_called()
+    assert any("not connected here" in m for m in caplog.messages)
+
+
+async def test_broadcast_when_no_user_id_header(client_message_queue_service):
+    payload = {"command": "announcement", "text": "hello world"}
+    msg = make_incoming_message(json.dumps(payload).encode(), headers=None)
+
+    await client_message_queue_service._on_message(msg)
+
+    client_message_queue_service.server.write_broadcast.assert_called_once_with(payload)
+
+
+async def test_broadcast_when_headers_empty(client_message_queue_service):
+    payload = {"command": "announcement"}
+    msg = make_incoming_message(json.dumps(payload).encode(), headers={})
+
+    await client_message_queue_service._on_message(msg)
+
+    client_message_queue_service.server.write_broadcast.assert_called_once_with(payload)
+
+
+async def test_malformed_json_body_is_dropped(
+    client_message_queue_service, caplog
+):
+    msg = make_incoming_message(b"not json at all", headers=None)
+
+    await client_message_queue_service._on_message(msg)
+
+    client_message_queue_service.server.write_broadcast.assert_not_called()
+    assert any("non-JSON body" in m for m in caplog.messages)
+
+
+async def test_non_object_json_body_is_dropped(client_message_queue_service):
+    msg = make_incoming_message(b"[1, 2, 3]", headers=None)
+
+    await client_message_queue_service._on_message(msg)
+
+    client_message_queue_service.server.write_broadcast.assert_not_called()
+
+
+async def test_invalid_user_id_header_is_dropped(
+    client_message_queue_service, caplog
+):
+    msg = make_incoming_message(
+        json.dumps({"command": "x"}).encode(),
+        headers={"user-id": "not-an-int"},
+    )
+
+    await client_message_queue_service._on_message(msg)
+
+    client_message_queue_service.server.write_broadcast.assert_not_called()
+    assert any("invalid user-id" in m for m in caplog.messages)
+
+
+async def test_channel_header_currently_drops(
+    client_message_queue_service, caplog
+):
+    msg = make_incoming_message(
+        json.dumps({"command": "x"}).encode(),
+        headers={"channel": "matchmaker"},
+    )
+
+    await client_message_queue_service._on_message(msg)
+
+    client_message_queue_service.server.write_broadcast.assert_not_called()
+    assert any("channel routing is not yet implemented" in m for m in caplog.messages)