M6-T06: extend MqttManager with subscribe + DSMR ingest

- mqtt.py: subscription registry, on_message dispatch (swallows handler
  errors), re-subscribe on (re)connect; publish/connect/disconnect/reconnect
  unchanged.
- app/services/dsmr_ingest.py: handle_message stores the full telegram frame
  (no allowlist; gas/phases/null preserved), 10s downsample by timestamp.second,
  source_id idempotency (select + IntegrityError rollback). Net-thread safe.
- main.py registers the DSMR subscription only when dsmr_ingest_enabled.

app/integrations/mqtt.py

@@ -18,12 +18,18 @@ Key design decisions
 - **Reconnect**: ``reconnect(settings)`` tears down the old client and
   establishes a fresh connection with the new settings.  Callers (e.g. PUT
   /api/config) call this after saving MQTT-related config values.
+- **Subscribe**: ``subscribe(topic, handler)`` registers a topic → handler
+  mapping.  Subscriptions are re-established automatically on reconnect.
+  The ``on_message`` callback dispatches incoming payloads to the registered
+  handler; handler exceptions are caught and logged so they never crash the
+  paho loop thread or the network connection.
 """
 
 from __future__ import annotations
 
 import logging
 import threading
+from collections.abc import Callable
 from typing import TYPE_CHECKING
 
 import paho.mqtt.client as mqtt
@@ -68,6 +74,9 @@ class MqttManager:
         self._client: mqtt.Client | None = None
         self._lock = threading.Lock()
         self._connected = False
+        # topic → handler registry; persists across reconnects so subscriptions
+        # are automatically re-established when the client reconnects.
+        self._subscriptions: dict[str, Callable[[bytes], None]] = {}
 
     # ------------------------------------------------------------------
     # Public properties
@@ -142,6 +151,30 @@ class MqttManager:
         except Exception:
             logger.exception("MQTT publish error (topic=%s).", topic)
 
+    def subscribe(self, topic: str, handler: Callable[[bytes], None]) -> None:
+        """Register *handler* to be called when a message arrives on *topic*.
+
+        If the client is already connected, the subscription is sent to the
+        broker immediately.  Otherwise it is queued and will be established
+        the next time ``_on_connect`` fires (including after a reconnect).
+
+        Handler exceptions are swallowed by the ``on_message`` dispatcher so
+        that a buggy handler can never crash the paho loop thread.
+        """
+        self._subscriptions[topic] = handler
+        # Grab the client reference outside the lock to avoid holding the lock
+        # while calling back into paho (which may itself acquire internal locks).
+        with self._lock:
+            client = self._client
+            connected = self._connected
+
+        if client is not None and connected:
+            try:
+                client.subscribe(topic)
+                logger.debug("MQTT subscribed to topic=%s (immediate).", topic)
+            except Exception:
+                logger.exception("MQTT subscribe error (topic=%s).", topic)
+
     # ------------------------------------------------------------------
     # Internal helpers — must be called with self._lock held
     # ------------------------------------------------------------------
@@ -168,6 +201,15 @@ class MqttManager:
             else:
                 logger.info("MQTT connected (reason_code=%s).", reason_code)
                 self._connected = True
+                # Re-establish all registered subscriptions after (re-)connect.
+                for sub_topic in self._subscriptions:
+                    try:
+                        _client.subscribe(sub_topic)
+                        logger.debug("MQTT re-subscribed to topic=%s.", sub_topic)
+                    except Exception:
+                        logger.exception(
+                            "MQTT re-subscribe failed (topic=%s).", sub_topic
+                        )
 
         # VERSION2 on_disconnect signature:
         # (client, userdata, disconnect_flags, reason_code, properties)
@@ -184,8 +226,29 @@ class MqttManager:
             else:
                 logger.info("MQTT disconnected cleanly.")
 
+        # VERSION2 on_message signature: (client, userdata, message)
+        def _on_message(
+            _client: mqtt.Client,
+            _userdata: object,
+            message: mqtt.MQTTMessage,
+        ) -> None:
+            handler = self._subscriptions.get(message.topic)
+            if handler is None:
+                logger.debug(
+                    "MQTT on_message: no handler for topic=%s.", message.topic
+                )
+                return
+            try:
+                handler(message.payload)
+            except Exception:
+                logger.exception(
+                    "MQTT message handler raised for topic=%s (swallowed).",
+                    message.topic,
+                )
+
         client.on_connect = _on_connect
         client.on_disconnect = _on_disconnect
+        client.on_message = _on_message
 
         # TLS
         if settings.mqtt_tls_enabled:

app/main.py

@@ -28,6 +28,7 @@ from app.config import get_settings
 from app.integrations.mqtt import mqtt_manager
 from app.services.auth import AuthBootstrapError, initialize_auth_schema
 from app.services.config_page import build_runtime_settings, seed_missing_config_from_bootstrap, sync_app_hostname_from_bootstrap
+from app.services.dsmr_ingest import handle_message as dsmr_handle_message
 from app.services.public_ip import check_public_ipv4_and_notify
 from app.services.modbus_poll import poll_all_enabled_devices, BASE_POLL_TICK_SECONDS
 from app.services.ha_discovery import publish_discovery, publish_states
@@ -201,6 +202,22 @@ async def lifespan(_: FastAPI):
         _startup_session.close()
     mqtt_manager.connect(_startup_runtime_settings)
 
+    # DSMR ingest: subscribe to the configured MQTT topic when enabled.
+    # The handler captures a snapshot of the current runtime settings (for the
+    # dsmr_sample_interval_s value).  Runtime setting changes take effect after
+    # a server restart or an explicit reconnect — consistent with other MQTT
+    # configuration in this project.
+    if _startup_runtime_settings.dsmr_ingest_enabled:
+        _dsmr_topic = _startup_runtime_settings.dsmr_mqtt_topic
+        _dsmr_settings_snapshot = _startup_runtime_settings
+        mqtt_manager.subscribe(
+            _dsmr_topic,
+            lambda payload: dsmr_handle_message(payload, _dsmr_settings_snapshot),
+        )
+        logger.info("DSMR ingest enabled — subscribed to topic=%s.", _dsmr_topic)
+    else:
+        logger.debug("DSMR ingest disabled — not subscribing to DSMR topic.")
+
     yield
 
     # MQTT: clean shutdown before the process exits.

app/services/dsmr_ingest.py

@@ -0,0 +1,159 @@
+"""DSMR telegram ingest service.
+
+Subscribes to the DSMR Reader MQTT topic (``dsmr/json``) and persists
+down-sampled DSMR telegram frames to the ``dsmr_reading`` table.
+
+Design decisions
+----------------
+- **Whole-frame storage**: the entire parsed telegram dict is stored as a JSON
+  blob in ``DsmrReading.payload``; no field allow-list is applied.  This lets
+  future commodities (gas, heating, three-phase) be accommodated without a
+  table-schema change.
+- **10-second down-sampling** (configurable via ``dsmr_sample_interval_s``):
+  only telegrams whose ``timestamp`` second falls on an exact multiple of the
+  interval are persisted.  This reduces write volume from ~60 rows/min to ~6
+  rows/min while guaranteeing that every 15-minute boundary (second=00) is
+  captured.
+- **Idempotency**: the telegram's own ``id`` field is stored as ``source_id``
+  with a UNIQUE constraint.  A second delivery of the same telegram (e.g. after
+  a broker reconnect) is silently skipped.
+- **Network-thread safety**: ``handle_message`` is called from paho's background
+  loop thread.  It opens and closes its own short-lived SQLAlchemy session and
+  swallows all exceptions so that a buggy payload or transient DB error never
+  crashes the paho loop or drops the MQTT connection.
+- **Numeric values kept as strings**: the DSMR Reader emits all numeric readings
+  as JSON strings (e.g. ``"20915.154"``).  They are stored verbatim; conversion
+  to ``Decimal`` is deferred to the billing engine (T07) where precision matters.
+- **Null phases**: some telegrams omit certain phase readings (``null`` in JSON);
+  these are stored as-is without special handling.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from datetime import datetime, timezone
+from typing import TYPE_CHECKING
+
+import sqlalchemy.exc
+from sqlalchemy import select
+
+from app.db import get_session_local
+from app.models.energy import DsmrReading
+
+if TYPE_CHECKING:
+    from app.config import Settings
+
+logger = logging.getLogger(__name__)
+
+
+def handle_message(payload_bytes: bytes, settings: "Settings") -> None:
+    """Parse one DSMR MQTT payload and persist it if it passes the sample filter.
+
+    Called from the paho network thread; *must* swallow all exceptions so that
+    a bad payload or transient error does not crash the loop or drop the broker
+    connection.
+
+    Parameters
+    ----------
+    payload_bytes:
+        Raw bytes from the MQTT message.
+    settings:
+        Runtime settings snapshot (captured at subscription time).  Used for
+        ``dsmr_sample_interval_s``.
+    """
+    try:
+        _handle_message_inner(payload_bytes, settings)
+    except Exception:
+        logger.exception("dsmr_ingest.handle_message: unexpected error (swallowed).")
+
+
+def _handle_message_inner(payload_bytes: bytes, settings: "Settings") -> None:
+    """Inner implementation — may raise; caller wraps in try/except."""
+    # --- 1. Parse JSON ---
+    try:
+        data: dict = json.loads(payload_bytes)
+    except (json.JSONDecodeError, ValueError):
+        logger.debug("dsmr_ingest: invalid JSON payload (skipped).")
+        return
+
+    if not isinstance(data, dict):
+        logger.debug("dsmr_ingest: payload is not a JSON object (skipped).")
+        return
+
+    # --- 2. Parse timestamp ---
+    raw_ts = data.get("timestamp")
+    if raw_ts is None:
+        logger.debug("dsmr_ingest: missing 'timestamp' field (skipped).")
+        return
+
+    try:
+        # Python 3.11+ accepts the trailing 'Z' directly; for 3.10 compat we
+        # replace 'Z' with '+00:00' before parsing.
+        ts_str = raw_ts if not isinstance(raw_ts, str) else raw_ts.replace("Z", "+00:00")
+        ts_utc: datetime = datetime.fromisoformat(ts_str)
+        # Ensure it is timezone-aware UTC.
+        if ts_utc.tzinfo is None:
+            ts_utc = ts_utc.replace(tzinfo=timezone.utc)
+    except (ValueError, TypeError, AttributeError):
+        logger.debug(
+            "dsmr_ingest: cannot parse 'timestamp' value %r (skipped).", raw_ts
+        )
+        return
+
+    # --- 3. Down-sample: only persist if second falls on interval boundary ---
+    interval = settings.dsmr_sample_interval_s
+    if interval > 0 and (ts_utc.second % interval) != 0:
+        # This telegram is between sample points; discard silently.
+        return
+
+    # --- 4. Extract source_id (telegram's own id field, for idempotency) ---
+    source_id: int | None = data.get("id")
+    if source_id is not None and not isinstance(source_id, int):
+        # Unexpected type — treat as missing rather than raising.
+        logger.debug(
+            "dsmr_ingest: 'id' field has unexpected type %s (ignoring).",
+            type(source_id).__name__,
+        )
+        source_id = None
+
+    # --- 5. Persist to database ---
+    session_local = get_session_local()
+    session = session_local()
+    try:
+        # Idempotency check: if source_id is known and already exists, skip.
+        if source_id is not None:
+            existing = session.scalar(
+                select(DsmrReading).where(DsmrReading.source_id == source_id)
+            )
+            if existing is not None:
+                logger.debug(
+                    "dsmr_ingest: source_id=%d already in DB, skipping.", source_id
+                )
+                return
+
+        reading = DsmrReading(
+            recorded_at=ts_utc,
+            source_id=source_id,
+            payload=data,  # full frame, verbatim
+        )
+        session.add(reading)
+        session.commit()
+        logger.debug(
+            "dsmr_ingest: persisted reading recorded_at=%s source_id=%s.",
+            ts_utc.isoformat(),
+            source_id,
+        )
+    except sqlalchemy.exc.IntegrityError:
+        # Race condition: another handler inserted the same source_id between our
+        # check and our insert.  Roll back and ignore.
+        session.rollback()
+        logger.debug(
+            "dsmr_ingest: IntegrityError for source_id=%s (duplicate, skipped).",
+            source_id,
+        )
+    except Exception:
+        session.rollback()
+        logger.exception("dsmr_ingest: DB error (swallowed).")
+    finally:
+        session.close()