feat(gateway): DataSink.publish_reading multi-name fan-out

ingestor/io/data_sink.py

@@ -0,0 +1,39 @@
+import ingestor.metrics as metrics
+
+
+class DataSink:
+    """Gateway egress wrapper: fans a reading out to every configured name/collection.
+
+    `postgres` is unused for now — task 12 adds `update_write_status` on top of it.
+    """
+
+    def __init__(self, data_manager, postgres, metadata: dict | None = None) -> None:
+        self.data_manager = data_manager
+        self.postgres = postgres
+        self.metadata = metadata
+
+    async def emit_metric(self, **kwargs) -> None:
+        # ponytail: delegate to data_manager, it's already a SientiaMonitoring
+        await self.data_manager.emit_metric(**kwargs)
+
+    async def publish_reading(self, tag: str, value, timestamp, names: list[dict]) -> None:
+        """Publish one MongoDB document per name entry and record its tag value metric."""
+        for name in names:
+            doc = {
+                'tag': tag,
+                'name': name['tag_name'],
+                'model_id': name['model_id'],
+                'timestamp': timestamp,
+                'value': value,
+            }
+            await self.data_manager.publish(name['collection'], doc)
+            await self.emit_metric(
+                metric_object=metrics.OPC_TAG_VALUE,
+                method='set',
+                value=value,
+                tags={
+                    'pod_id': self.data_manager.pod_id,
+                    'tag_name': name['tag_name'],
+                    'model_id': name['model_id'],
+                },
+            )

tests/unit/io/test_data_sink_publish.py

@@ -0,0 +1,69 @@
+from unittest.mock import AsyncMock
+
+import pytest
+
+import ingestor.metrics as metrics
+from ingestor.io.data_sink import DataSink
+
+
+def make_sink(data_manager=None):
+    dm = data_manager or AsyncMock()
+    sink = DataSink(data_manager=dm, postgres=AsyncMock())
+    sink.emit_metric = AsyncMock()
+    return sink, dm
+
+
+@pytest.mark.asyncio
+async def test_publish_reading_fans_out_per_name():
+    sink, dm = make_sink()
+    names = [
+        {'collection': 'raw_x', 'tag_name': 'Counter', 'model_id': '1', 'model_name': 'm1'},
+        {'collection': 'raw_y', 'tag_name': 'Contador', 'model_id': '2', 'model_name': 'm2'},
+    ]
+
+    await sink.publish_reading(
+        tag='ns=2;i=2', value=42, timestamp='2026-07-13T00:00:00Z', names=names
+    )
+
+    assert dm.publish.await_count == 2
+
+
+@pytest.mark.asyncio
+async def test_publish_reading_builds_doc_and_routes_collection():
+    sink, dm = make_sink()
+    names = [{'collection': 'raw_x', 'tag_name': 'Counter', 'model_id': '1', 'model_name': 'm1'}]
+
+    await sink.publish_reading(
+        tag='ns=2;i=2', value=42, timestamp='2026-07-13T00:00:00Z', names=names
+    )
+
+    collection, doc = dm.publish.await_args_list[0].args
+    assert collection == 'raw_x'
+    assert doc == {
+        'tag': 'ns=2;i=2',
+        'name': 'Counter',
+        'model_id': '1',
+        'timestamp': '2026-07-13T00:00:00Z',
+        'value': 42,
+    }
+
+
+@pytest.mark.asyncio
+async def test_publish_reading_emits_tag_value_metric_per_name():
+    sink, dm = make_sink()
+    names = [
+        {'collection': 'raw_x', 'tag_name': 'Counter', 'model_id': '1', 'model_name': 'm1'},
+        {'collection': 'raw_y', 'tag_name': 'Contador', 'model_id': '2', 'model_name': 'm2'},
+    ]
+
+    await sink.publish_reading(
+        tag='ns=2;i=2', value=42, timestamp='2026-07-13T00:00:00Z', names=names
+    )
+
+    assert sink.emit_metric.await_count == 2
+    sink.emit_metric.assert_any_await(
+        metric_object=metrics.OPC_TAG_VALUE,
+        method='set',
+        value=42,
+        tags={'pod_id': dm.pod_id, 'tag_name': 'Counter', 'model_id': '1'},
+    )