feat(gateway): NatsWriteConsumer (Core NATS, queue group) (C2)

ingestor/adapters/nats_consumer.py

@@ -0,0 +1,53 @@
+"""NATS input adapter for OPC write commands (contract C2).
+
+Thin adapter: decode the message JSON and hand the dict to WriteEngine,
+which is already raise-safe against malformed commands. No metrics of its
+own — the engine records write metrics; a malformed-JSON drop is logged
+here. No reconnect logic — nats-py handles reconnection internally.
+"""
+
+import json
+
+import nats
+from sientia_do.observability.logger import Logger
+
+from ingestor.engines.write import WriteEngine
+
+
+class NatsWriteConsumer:
+    def __init__(
+        self,
+        nats_url: str,
+        write_engine: WriteEngine,
+        logger: Logger,
+        subject: str = 'opc.write',
+        queue: str = 'gateway-writers',
+    ):
+        self.nats_url = nats_url
+        self.write_engine = write_engine
+        self.logger = logger
+        self.subject = subject
+        self.queue = queue
+        self.nc = None
+        self.sub = None
+
+    async def start(self) -> None:
+        self.nc = await nats.connect(servers=self.nats_url)
+        self.sub = await self.nc.subscribe(self.subject, queue=self.queue, cb=self._handle)
+        self.logger.info(f'NatsWriteConsumer started: subject={self.subject} queue={self.queue}')
+
+    async def _handle(self, msg) -> None:
+        try:
+            payload = json.loads(msg.data)
+        except (ValueError, UnicodeDecodeError) as e:
+            preview = msg.data[:200]
+            self.logger.error(f'NatsWriteConsumer dropped malformed JSON ({e}): {preview!r}')
+            return
+        await self.write_engine.write(payload)
+
+    async def stop(self) -> None:
+        if self.nc is None:
+            return
+        if self.sub is not None:
+            await self.sub.unsubscribe()
+        await self.nc.drain()

tests/unit/adapters/test_nats_consumer.py

@@ -0,0 +1,76 @@
+import json
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from ingestor.adapters.nats_consumer import NatsWriteConsumer
+
+
+def make_consumer(**kwargs):
+    return NatsWriteConsumer(
+        nats_url='nats://localhost:4222',
+        write_engine=AsyncMock(),
+        logger=MagicMock(),
+        **kwargs,
+    )
+
+
+@pytest.mark.asyncio
+async def test_handle_valid_json_dispatches_to_write_engine():
+    consumer = make_consumer()
+    payload = {'server_id': 'srv1', 'addr': 'ns=2;i=5', 'value': 1.0, 'data_type': 'float'}
+    msg = MagicMock(data=json.dumps(payload).encode())
+
+    await consumer._handle(msg)
+
+    consumer.write_engine.write.assert_awaited_once_with(payload)
+
+
+@pytest.mark.asyncio
+async def test_handle_invalid_json_drops_message_and_logs_error():
+    consumer = make_consumer()
+    msg = MagicMock(data=b'not-json')
+
+    await consumer._handle(msg)
+
+    consumer.write_engine.write.assert_not_awaited()
+    consumer.logger.error.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_start_subscribes_with_subject_and_queue():
+    consumer = make_consumer(subject='opc.write', queue='gateway-writers')
+    mock_nc = AsyncMock()
+    mock_nc.subscribe = AsyncMock()
+
+    with patch('ingestor.adapters.nats_consumer.nats') as mock_nats:
+        mock_nats.connect = AsyncMock(return_value=mock_nc)
+        await consumer.start()
+
+    mock_nats.connect.assert_awaited_once_with(servers='nats://localhost:4222')
+    mock_nc.subscribe.assert_awaited_once_with(
+        'opc.write', queue='gateway-writers', cb=consumer._handle
+    )
+
+
+@pytest.mark.asyncio
+async def test_stop_before_start_does_not_raise():
+    consumer = make_consumer()
+
+    await consumer.stop()  # no raise
+
+
+@pytest.mark.asyncio
+async def test_stop_after_start_drains_and_unsubscribes():
+    consumer = make_consumer()
+    mock_nc = AsyncMock()
+    mock_sub = AsyncMock()
+    mock_nc.subscribe = AsyncMock(return_value=mock_sub)
+
+    with patch('ingestor.adapters.nats_consumer.nats') as mock_nats:
+        mock_nats.connect = AsyncMock(return_value=mock_nc)
+        await consumer.start()
+        await consumer.stop()
+
+    mock_sub.unsubscribe.assert_awaited_once()
+    mock_nc.drain.assert_awaited_once()