🚧 Add broker status page

backend/app/dynsec.py

@@ -61,6 +61,9 @@ class DynsecClient:
         self._lock = threading.Lock()
         self._responses: dict[str, queue.Queue[dict[str, Any]]] = {}
         self._connected = threading.Event()
+        # Optional tap for $SYS telemetry (app.metrics registers itself
+        # at startup). Receives (topic, payload) for every $SYS message.
+        self._event_sink = None
 
     # -- connection lifecycle ------------------------------------------------
     def connect(self) -> None:
@@ -118,12 +121,25 @@ class DynsecClient:
                 return
             self._teardown_locked()
 
+    def set_event_sink(self, sink) -> None:
+        """Register a callable receiving every $SYS message (topic, payload).
+
+        Used by app.metrics for the broker dashboard (spec sections 33,
+        55). The sink must never raise: telemetry must not be able to
+        disturb the management channel.
+        """
+        self._event_sink = sink
+
     # -- paho callbacks -------------------------------------------------------
     def _on_connect(self, client, userdata, flags, reason_code, properties) -> None:
         if reason_code == 0:
             # FIXED response topic — all commands reply here; requests are
             # matched by the correlationData echo inside the payload.
             client.subscribe(RESPONSE_TOPIC, qos=1)
+            # $SYS telemetry for the dashboard (spec sections 33, 55).
+            # qos 0: telemetry is periodically republished, losing one
+            # is harmless.
+            client.subscribe("$SYS/#", qos=0)
             self._connected.set()
         else:
             logger.error("dynsec client connect failed: %s", reason_code)
@@ -149,6 +165,14 @@ class DynsecClient:
 
     def _on_message(self, client, userdata, msg) -> None:
         if msg.topic != RESPONSE_TOPIC:
+            # Not a dynsec reply: forward $SYS telemetry to the sink, if
+            # one is registered. Swallow sink errors — telemetry must
+            # never disturb the management channel.
+            if msg.topic.startswith("$SYS/") and self._event_sink is not None:
+                try:
+                    self._event_sink(msg.topic, msg.payload)
+                except Exception:  # noqa: BLE001
+                    logger.warning("$SYS event sink raised; message dropped")
             return
         try:
             payload = json.loads(msg.payload.decode("utf-8"))

backend/app/main.py

@@ -66,6 +66,21 @@ async def lifespan(app: FastAPI):
     try:
         dynsec_client.connect()
         logger.info("dynamic security channel connected")
+        from app import metrics
+
+        dynsec_client.set_event_sink(metrics.handle_sys_message)
+        # Best effort: ensure the management identity can read $SYS —
+        # the 2.1 auto-generated admin role set does not guarantee it
+        # (see the compose healthcheck note). sys-observe is one of the
+        # plugin's default roles (read-only $SYS/#); addClientRole is
+        # idempotent broker-side.
+        try:
+            dynsec_client.add_client_role(
+                settings.mosquitto_dynsec_username, "sys-observe", priority=1
+            )
+            logger.info("sys-observe role granted to the management identity")
+        except Exception as exc:  # noqa: BLE001
+            logger.warning("could not grant sys-observe: %s", exc)
     except Exception as exc:  # noqa: BLE001 - startup should not crash the API
         logger.warning("dynsec channel not ready yet: %s", exc)
     yield

backend/app/metrics.py

@@ -0,0 +1,123 @@
+"""$SYS broker telemetry (spec sections 33, 55).
+
+The dynsec channel's MQTT connection subscribes to $SYS/# and forwards
+every message here (see DynsecClient.set_event_sink). Values are parsed
+ONCE at ingest; snapshot() returns the latest known state. The broker
+publishes $SYS on the configured sys_interval (10 s in the bootstrap
+config), so a fresh backend needs a few seconds before the first
+snapshot is populated — every field is therefore Optional and the
+`available` flag says whether ANY $SYS message has been seen.
+"""
+
+from __future__ import annotations
+
+import threading
+from datetime import UTC, datetime
+from typing import Any, Callable
+
+SYS_PREFIX = "$SYS/broker/"
+
+_lock = threading.Lock()
+_values: dict[str, Any] = {}
+_last_update: datetime | None = None
+
+
+def _as_int(payload: str) -> object:
+    try:
+        return int(payload)
+    except ValueError:
+        return payload
+
+
+def _as_float(payload: str) -> object:
+    try:
+        return float(payload)
+    except ValueError:
+        return payload
+
+
+# $SYS topic -> (snapshot field, parser). Field names mirror
+# BrokerStatusOut (schemas.py). NOTE: "retained messages/count"
+# genuinely contains a space in the Mosquitto topic name.
+_TOPIC_FIELDS: dict[str, tuple[str, Callable[[str], object]]] = {
+    f"{SYS_PREFIX}version": ("version", str),
+    f"{SYS_PREFIX}uptime": ("uptime", str),
+    f"{SYS_PREFIX}timestamp": ("build_timestamp", str),
+    # clients/disconnected = persistent clients currently offline,
+    # i.e. the "stored sessions" of the dashboard (spec section 33).
+    f"{SYS_PREFIX}clients/connected": ("clients_connected", _as_int),
+    f"{SYS_PREFIX}clients/disconnected": ("clients_disconnected", _as_int),
+    f"{SYS_PREFIX}clients/maximum": ("clients_maximum", _as_int),
+    f"{SYS_PREFIX}clients/total": ("clients_total", _as_int),
+    f"{SYS_PREFIX}clients/expired": ("clients_expired", _as_int),
+    f"{SYS_PREFIX}messages/received": ("messages_received", _as_int),
+    f"{SYS_PREFIX}messages/sent": ("messages_sent", _as_int),
+    f"{SYS_PREFIX}messages/stored": ("messages_stored", _as_int),
+    f"{SYS_PREFIX}messages/dropped": ("messages_dropped", _as_int),
+    f"{SYS_PREFIX}publish/messages/received": ("publish_received", _as_int),
+    f"{SYS_PREFIX}publish/messages/sent": ("publish_sent", _as_int),
+    f"{SYS_PREFIX}publish/messages/dropped": ("publish_dropped", _as_int),
+    f"{SYS_PREFIX}bytes/received": ("bytes_received", _as_int),
+    f"{SYS_PREFIX}bytes/sent": ("bytes_sent", _as_int),
+    f"{SYS_PREFIX}subscriptions/count": ("subscriptions_count", _as_int),
+    f"{SYS_PREFIX}retained messages/count": ("retained_count", _as_int),
+    # load/* are per-second averages over the last minute ("x.xx").
+    f"{SYS_PREFIX}load/messages/received/1min": (
+        "load_messages_received_1min",
+        _as_float,
+    ),
+    f"{SYS_PREFIX}load/messages/sent/1min": (
+        "load_messages_sent_1min",
+        _as_float,
+    ),
+    f"{SYS_PREFIX}load/publish/received/1min": (
+        "load_publish_received_1min",
+        _as_float,
+    ),
+    f"{SYS_PREFIX}load/publish/sent/1min": (
+        "load_publish_sent_1min",
+        _as_float,
+    ),
+    f"{SYS_PREFIX}load/bytes/received/1min": (
+        "load_bytes_received_1min",
+        _as_float,
+    ),
+    f"{SYS_PREFIX}load/bytes/sent/1min": (
+        "load_bytes_sent_1min",
+        _as_float,
+    ),
+}
+
+_FIELDS = [field for field, _ in _TOPIC_FIELDS.values()]
+
+
+def handle_sys_message(topic: str, payload: bytes) -> None:
+    """Ingest one $SYS message (wired as the dynsec client's event sink)."""
+    entry = _TOPIC_FIELDS.get(topic)
+    if entry is None:
+        return  # unknown $SYS topic — deliberately ignored
+    field, parser = entry
+    text = payload.decode("utf-8", errors="replace").strip()
+    global _last_update
+    with _lock:
+        _values[field] = parser(text)
+        _last_update = datetime.now(UTC)
+
+
+def reset() -> None:
+    """Clear all telemetry (used by tests)."""
+    global _last_update
+    with _lock:
+        _values.clear()
+        _last_update = None
+
+
+def snapshot() -> dict[str, Any]:
+    """Latest known $SYS state; every field None until first publish."""
+    with _lock:
+        values = dict(_values)
+        last_update = _last_update
+    snap: dict[str, Any] = {field: values.get(field) for field in _FIELDS}
+    snap["available"] = last_update is not None
+    snap["last_update"] = last_update
+    return snap

backend/app/routers/broker.py

@@ -33,6 +33,7 @@ from app.models import (
 from app.schemas import (
     BrokerSettingsIn,
     BrokerSettingsOut,
+    BrokerStatusOut,
     ConfigPreviewOut,
     DeployRequest,
     ListenerIn,
@@ -316,6 +317,19 @@ def _publish_certificates(db: Session) -> None:
                 )
 
 
+@router.get("/status", response_model=BrokerStatusOut)
+def broker_status(_user: AnyReader):
+    """Live $SYS telemetry for the dashboard (spec sections 33, 55).
+
+    Read-only: every management role may view (spec section 9). Fields
+    are None until the first $SYS publish arrives — the broker publishes
+    on sys_interval (10 s), so a fresh backend needs a few seconds.
+    """
+    from app import metrics
+
+    return BrokerStatusOut(**metrics.snapshot())
+
+
 @router.get("/config/preview", response_model=ConfigPreviewOut)
 def preview_config(
     db: Annotated[Session, Depends(get_db)], _user: AnyReader

backend/app/schemas.py

@@ -230,6 +230,43 @@ class ConfigPreviewOut(BaseModel):
     warnings: list[str]
 
 
+class BrokerStatusOut(BaseModel):
+    """$SYS snapshot for the dashboard (spec sections 33, 55).
+
+    Every metric is None until the first $SYS publish arrives (the
+    broker publishes on sys_interval, 10 s in the bootstrap config);
+    `available` says whether ANY $SYS message has been ingested.
+    """
+
+    available: bool
+    last_update: datetime | None
+    version: str | None = None
+    uptime: str | None = None
+    build_timestamp: str | None = None
+    clients_connected: int | None = None
+    clients_disconnected: int | None = None
+    clients_maximum: int | None = None
+    clients_total: int | None = None
+    clients_expired: int | None = None
+    messages_received: int | None = None
+    messages_sent: int | None = None
+    messages_stored: int | None = None
+    messages_dropped: int | None = None
+    publish_received: int | None = None
+    publish_sent: int | None = None
+    publish_dropped: int | None = None
+    bytes_received: int | None = None
+    bytes_sent: int | None = None
+    subscriptions_count: int | None = None
+    retained_count: int | None = None
+    load_messages_received_1min: float | None = None
+    load_messages_sent_1min: float | None = None
+    load_publish_received_1min: float | None = None
+    load_publish_sent_1min: float | None = None
+    load_bytes_received_1min: float | None = None
+    load_bytes_sent_1min: float | None = None
+
+
 class DeployRequest(BaseModel):
     comment: str | None = None
 

backend/tests/test_api.py

@@ -194,6 +194,51 @@ def test_system_endpoints_rbac(auditor_client):
     )
 
 
+def test_broker_status_endpoint(admin_client, auditor_client):
+    """GET /broker/status returns the $SYS snapshot (spec 33/55).
+
+    No $SYS messages are ingested in tests (no real broker), so the
+    endpoint must report available: False with all metrics None; after
+    ingesting fake $SYS messages the snapshot reflects them. Read-only:
+    the auditor role may view (spec section 9).
+    """
+    from app import metrics
+
+    metrics.reset()
+
+    empty = admin_client.get("/api/v1/broker/status")
+    assert empty.status_code == 200
+    body = empty.json()
+    assert body["available"] is False
+    assert body["version"] is None
+    assert body["clients_connected"] is None
+
+    # Ingest fake $SYS telemetry and verify the snapshot reflects it.
+    metrics.handle_sys_message("$SYS/broker/version", b"2.1.2")
+    metrics.handle_sys_message("$SYS/broker/clients/connected", b"4")
+    metrics.handle_sys_message("$SYS/broker/retained messages/count", b"7")
+    metrics.handle_sys_message("$SYS/broker/load/messages/received/1min", b"1.25")
+
+    body = admin_client.get("/api/v1/broker/status").json()
+    assert body["available"] is True
+    assert body["version"] == "2.1.2"
+    assert body["clients_connected"] == 4
+    assert body["retained_count"] == 7  # topic contains a literal space
+    assert body["load_messages_received_1min"] == 1.25
+    assert body["last_update"] is not None
+
+    # Unknown $SYS topics are ignored without breaking the snapshot.
+    metrics.handle_sys_message("$SYS/broker/something/unknown", b"x")
+    body = admin_client.get("/api/v1/broker/status").json()
+    assert body["available"] is True
+
+    # Read-only for every management role (spec section 9: auditor has
+    # read access to configuration/status).
+    assert auditor_client.get("/api/v1/broker/status").status_code == 200
+
+    metrics.reset()
+
+
 def test_role_acl_lifecycle(admin_client):
     admin_client.post("/api/v1/mqtt/roles", json={"rolename": "telemetry"})