feat(gateway): DataSink.update_write_status Postgres by timestamp+model_id (C3); drop unused metadata param

ingestor/io/data_sink.py

@@ -1,16 +1,18 @@
+import re
+
+from sqlalchemy import text
+
 import ingestor.metrics as metrics
 
+_IDENTIFIER_RE = re.compile(r'^[A-Za-z0-9_]+$')
+
 
 class DataSink:
-    """Gateway egress wrapper: fans a reading out to every configured name/collection.
+    """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:
+    def __init__(self, data_manager, postgres) -> 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
@@ -37,3 +39,30 @@ class DataSink:
                     'model_id': name['model_id'],
                 },
             )
+
+    async def update_write_status(self, cmd: dict, result) -> None:
+        """UPDATE the per-model prediction row's write_status (C3), by timestamp+model_id.
+
+        schema/table_name are per-model SQL identifiers, not values — they can't be
+        bound params, so they're checked against an allowlist before interpolation.
+        status/timestamp/model_id ARE bound params, never f-stringed.
+        """
+        schema = self._validate_identifier(cmd['schema'])
+        table = self._validate_identifier(cmd['table_name'])
+        # schema/table are allowlist-validated identifiers above; status/timestamp/model_id
+        # are bound params below, never interpolated — not a real injection vector.
+        stmt = text(
+            f'UPDATE {schema}.{table} SET write_status = :status '  # noqa: S608
+            'WHERE timestamp = :timestamp AND model_id = :model_id'
+        )
+        params = {'status': result.status, 'timestamp': cmd['timestamp'], 'model_id': cmd['model_id']}
+        # ponytail: sync engine call inside async def — same pattern sientia_do's own
+        # Postgres activities use (e.g. export_data_to_postgres); no async pg driver here.
+        with self.postgres.engine.begin() as connection:
+            connection.execute(stmt, params)
+
+    @staticmethod
+    def _validate_identifier(name: str) -> str:
+        if not _IDENTIFIER_RE.match(name):
+            raise ValueError(f'invalid SQL identifier: {name!r}')
+        return name