DSMR hardening: restart-free config + decouple from telegram id

Config hot-reload (no restart):
- MqttManager.unsubscribe(); apply_dsmr_subscription(settings) re-applies the
  DSMR subscription (subscribe/unsubscribe/topic-change, fresh snapshot so the
  sample interval also takes effect).
- Called from lifespan AND PUT /api/config, so toggling DSMR ingest in the UI
  takes effect immediately without restarting the app.

Decouple ingest from the DSMR telegram id (overflows / resets to zero):
- Migration 20260624_12: drop UNIQUE(source_id), make recorded_at UNIQUE.
- Idempotency now keys on recorded_at (telegram timestamp); the table's own
  autoincrement PK is the stable identity. source_id kept only as a reference.
- Regression test: colliding telegram ids no longer drop new data.

app/services/dsmr_ingest.py

@@ -47,6 +47,50 @@ if TYPE_CHECKING:
 logger = logging.getLogger(__name__)
 
 
+# Tracks the DSMR topic currently subscribed via the MQTT manager, so a config
+# change can unsubscribe the old topic before subscribing the new one.
+_current_dsmr_topic: str | None = None
+
+
+def apply_dsmr_subscription(settings: "Settings") -> None:
+    """(Re)apply the DSMR MQTT subscription to match *settings* — restart-free.
+
+    Call this at startup and after every config save.  It makes the live MQTT
+    subscription reflect the current ``dsmr_ingest_enabled`` / ``dsmr_mqtt_topic``
+    / ``dsmr_sample_interval_s`` settings without an app restart:
+
+    - **Disabled** → unsubscribe any active DSMR subscription.
+    - **Enabled** → (re)subscribe to ``dsmr_mqtt_topic`` with a handler bound to
+      a *fresh* settings snapshot, so a changed sample interval also takes effect.
+    - **Topic changed** → unsubscribe the old topic before subscribing the new one.
+
+    Idempotent and safe to call when MQTT is not connected (the subscription is
+    queued in the manager and established on the next connect).
+    """
+    # Imported here (not at module top) to avoid a circular import at app start.
+    from app.integrations.mqtt import mqtt_manager
+
+    global _current_dsmr_topic
+
+    if not settings.dsmr_ingest_enabled:
+        if _current_dsmr_topic is not None:
+            mqtt_manager.unsubscribe(_current_dsmr_topic)
+            logger.info("DSMR ingest disabled — unsubscribed from topic=%s.", _current_dsmr_topic)
+            _current_dsmr_topic = None
+        return
+
+    topic = settings.dsmr_mqtt_topic
+    if _current_dsmr_topic is not None and _current_dsmr_topic != topic:
+        mqtt_manager.unsubscribe(_current_dsmr_topic)
+
+    # Re-subscribe (overwrites any existing handler for this topic) with a fresh
+    # settings snapshot so dsmr_sample_interval_s changes take effect too.
+    snapshot = settings
+    mqtt_manager.subscribe(topic, lambda payload: handle_message(payload, snapshot))
+    _current_dsmr_topic = topic
+    logger.info("DSMR ingest enabled — subscribed to topic=%s.", topic)
+
+
 def handle_message(payload_bytes: bytes, settings: "Settings") -> None:
     """Parse one DSMR MQTT payload and persist it if it passes the sample filter.
 
@@ -107,7 +151,8 @@ def _handle_message_inner(payload_bytes: bytes, settings: "Settings") -> None:
         # This telegram is between sample points; discard silently.
         return
 
-    # --- 4. Extract source_id (telegram's own id field, for idempotency) ---
+    # --- 4. Extract source_id (telegram's own id) — stored only as a reference,
+    #        NOT used for uniqueness/idempotency (it overflows and gets reset). ---
     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.
@@ -118,19 +163,22 @@ def _handle_message_inner(payload_bytes: bytes, settings: "Settings") -> None:
         source_id = None
 
     # --- 5. Persist to database ---
+    # Idempotency is keyed on recorded_at (the telegram timestamp), which is
+    # telegram-id-independent: a single P1 meter emits one telegram per second,
+    # and down-sampling keeps at most one per interval-aligned second.  The
+    # UNIQUE(recorded_at) constraint is the backstop for the IntegrityError race.
     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)
+        existing = session.scalar(
+            select(DsmrReading).where(DsmrReading.recorded_at == ts_utc)
+        )
+        if existing is not None:
+            logger.debug(
+                "dsmr_ingest: recorded_at=%s already in DB, skipping.",
+                ts_utc.isoformat(),
             )
-            if existing is not None:
-                logger.debug(
-                    "dsmr_ingest: source_id=%d already in DB, skipping.", source_id
-                )
-                return
+            return
 
         reading = DsmrReading(
             recorded_at=ts_utc,
@@ -145,12 +193,11 @@ def _handle_message_inner(payload_bytes: bytes, settings: "Settings") -> None:
             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.
+        # Race / duplicate: another insert beat us to the same recorded_at.
         session.rollback()
         logger.debug(
-            "dsmr_ingest: IntegrityError for source_id=%s (duplicate, skipped).",
-            source_id,
+            "dsmr_ingest: IntegrityError for recorded_at=%s (duplicate, skipped).",
+            ts_utc.isoformat(),
         )
     except Exception:
         session.rollback()