Implement Phase 4 integrations and display telemetry

Add MQTT playback control, durable Home Assistant incident notifications, and display heartbeat telemetry on top of the verified Phase 3B playback authority. No playback logic is duplicated: every MQTT command and every observed condition goes through the existing playback service, and all integration work happens after the authoritative SQLite commit. MQTT (server/src/mqtt) subscribes the exact required command topics under a configurable prefix that defaults to announcements. Durable commands use QoS 1; next and previous stay at QoS 0 because they are not idempotent and carry no command ID. Retained authoritative state is published only on the separate availability, mode/state, and playback/state topics, never onto a command topic, and only after a mutation commits. A registered last will reports an ungraceful disappearance, a clean shutdown retracts it, and a broker outage degrades status while playback, HTTP, and the display protocol continue untouched. Notifications (server/src/notifications) use the Phase 1 incident and outbox tables. A failing condition opens one incident and queues exactly one notification; continued failure only updates its observation time; recovery queues exactly one recovery; a later failure is a new episode. Delivery retries on a documented capped schedule with a stable notification ID, survives restart, and never blocks playback. With no webhook URL configured, incidents are still recorded but no delivery work is queued. Conditions come from storage health transitions, committed playback state, display heartbeat staleness, and newly failed uploads. Displays (server/src/displays) ingest validated, bounded display:heartbeat telemetry keyed by persistent client ID rather than socket ID. Telemetry is read-only and never touches playback or playlist revisions. Offline is decided by heartbeat staleness, not by a closed socket, and a startup grace period keeps an API restart from manufacturing offline alerts. Adds GET /api/displays and meaningful-change display:status events. Health and status now report redacted storage, MQTT, notification, playback, and display state. External integrations degrade status and never affect readiness, so no integration outage can cause a restart loop. No MQTT password, S3 credential, webhook URL, or webhook secret is exposed. Uses the existing Phase 1 schema unchanged; no migration was required. Co-Authored-By: Claude Opus 5 <[email protected]>
2026-08-16 22:52:59 -05:00
parent 993986c25e
commit ae70bf1dd8
22 changed files with 3691 additions and 24 deletions
@@ -0,0 +1,430 @@
+const test = require("node:test");
+const assert = require("node:assert/strict");
+const { loadConfig } = require("../src/config");
+const { createRuntime } = require("../src/runtime");
+
+const silentLogger = { info() {}, warn() {}, error() {} };
+
+function testConfig(overrides = {}) {
+  return loadConfig({
+    NODE_ENV: "test",
+    HOST: "127.0.0.1",
+    PORT: "0",
+    DATABASE_PATH: ":memory:",
+    APP_TIMEZONE: "America/Chicago",
+    SHUTDOWN_TIMEOUT_MS: "1000",
+    S3_ENDPOINT: "http://seaweedfs:8333",
+    S3_BUCKET: "announcements-test",
+    MQTT_URL: "mqtt://broker:1883",
+    MQTT_PASSWORD: "mqtt-password-that-must-not-leak",
+    HA_WEBHOOK_URL: "http://home-assistant:8123/api/webhook/private-id",
+    ...overrides,
+  });
+}
+
+function testStorage(health = { status: "healthy", checkedAt: null, lastSuccessAt: null }) {
+  return {
+    async start() {},
+    stop() {},
+    health: () => health,
+  };
+}
+
+function createFakeBroker({ failPublish = false } = {}) {
+  const listeners = new Map();
+  const client = {
+    published: [],
+    subscriptions: [],
+    on(event, handler) {
+      listeners.set(event, [...(listeners.get(event) || []), handler]);
+      return client;
+    },
+    emit(event, ...args) {
+      for (const handler of listeners.get(event) || []) handler(...args);
+    },
+    publish(topic, payload, options, callback) {
+      if (failPublish) throw new Error("broker is gone");
+      client.published.push({ topic, payload: String(payload) });
+      callback?.(null);
+    },
+    subscribe(topic, options, callback) {
+      client.subscriptions.push(topic);
+      callback?.(null);
+    },
+    unsubscribe() {},
+    end(force, options, callback) {
+      callback?.();
+    },
+    deliver(topic, payload) {
+      client.emit("message", topic, Buffer.from(payload));
+    },
+  };
+  return client;
+}
+
+async function jsonRequest(baseUrl, route, { body, ...init } = {}) {
+  const response = await fetch(`${baseUrl}${route}`, {
+    ...init,
+    ...(body === undefined
+      ? {}
+      : {
+          headers: { "content-type": "application/json", ...init.headers },
+          body: JSON.stringify(body),
+        }),
+  });
+  return { response, body: await response.json() };
+}
+
+function splitEnginePackets(payload) {
+  return payload.split("\x1e").filter(Boolean);
+}
+
+async function connectSocket(baseUrl) {
+  let requestNumber = 0;
+  const endpoint = (sid) => {
+    const parameters = new URLSearchParams({
+      EIO: "4",
+      transport: "polling",
+      t: String(requestNumber++),
+    });
+    if (sid) parameters.set("sid", sid);
+    return `${baseUrl}/socket.io/?${parameters}`;
+  };
+  const opening = splitEnginePackets(await (await fetch(endpoint())).text());
+  const { sid } = JSON.parse(opening.find((packet) => packet.startsWith("0")).slice(1));
+  await (
+    await fetch(endpoint(sid), {
+      method: "POST",
+      headers: { "content-type": "text/plain;charset=UTF-8" },
+      body: "40",
+    })
+  ).text();
+  await (await fetch(endpoint(sid))).text();
+  return {
+    async send(event, payload) {
+      const response = await fetch(endpoint(sid), {
+        method: "POST",
+        headers: { "content-type": "text/plain;charset=UTF-8" },
+        body: `42${JSON.stringify([event, payload])}`,
+      });
+      await response.text();
+    },
+    async poll() {
+      return (await fetch(endpoint(sid), { signal: AbortSignal.timeout(3_000) })).text();
+    },
+    async close() {
+      await (
+        await fetch(endpoint(sid), {
+          method: "POST",
+          headers: { "content-type": "text/plain;charset=UTF-8" },
+          body: "41",
+        })
+      ).text();
+    },
+  };
+}
+
+function socketEvents(payload) {
+  return splitEnginePackets(payload)
+    .filter((packet) => packet.startsWith("42"))
+    .map((packet) => JSON.parse(packet.slice(2)));
+}
+
+async function startRuntime(t, { broker, config, storage, notificationFetch } = {}) {
+  const runtime = createRuntime({
+    config: config || testConfig(),
+    logger: silentLogger,
+    storage: storage || testStorage(),
+    ...(broker === null ? {} : { mqttConnect: () => broker }),
+    ...(notificationFetch ? { notificationFetch } : {}),
+  });
+  await runtime.start();
+  t.after(() => runtime.stop("test cleanup"));
+  broker?.emit("connect");
+  return { runtime, baseUrl: `http://127.0.0.1:${runtime.address.port}` };
+}
+
+async function seedPlaylist(runtime, baseUrl) {
+  const timestamp = Date.now();
+  runtime.database
+    .prepare(
+      `INSERT INTO assets (
+        id, source_sha256, display_name, original_filename, status,
+        display_object_key, display_sha256, display_content_type,
+        display_byte_size, display_width, display_height, created_at, updated_at
+      ) VALUES (?, ?, ?, ?, 'ready', 'private/display/a.webp', ?, 'image/webp', 10, 1920, 1080, ?, ?)`,
+    )
+    .run("asset-one", "a".repeat(64), "One", "one.png", "b".repeat(64), timestamp, timestamp);
+  runtime.database
+    .prepare(
+      `INSERT INTO assets (
+        id, source_sha256, display_name, original_filename, status,
+        display_object_key, display_sha256, display_content_type,
+        display_byte_size, display_width, display_height, created_at, updated_at
+      ) VALUES (?, ?, ?, ?, 'ready', 'private/display/b.webp', ?, 'image/webp', 10, 1920, 1080, ?, ?)`,
+    )
+    .run("asset-two", "c".repeat(64), "Two", "two.png", "d".repeat(64), timestamp, timestamp);
+  const playlist = await jsonRequest(baseUrl, "/api/playlists", {
+    method: "POST",
+    body: { name: "Integration" },
+  });
+  for (const assetId of ["asset-one", "asset-two"]) {
+    await jsonRequest(baseUrl, `/api/playlists/${playlist.body.id}/entries`, {
+      method: "POST",
+      body: { assetId, durationOverrideMs: 60_000 },
+    });
+  }
+  await jsonRequest(baseUrl, `/api/playlists/${playlist.body.id}/activate`, {
+    method: "POST",
+  });
+  return playlist.body;
+}
+
+test("an MQTT command and its REST equivalent reach the same playback authority", async (t) => {
+  const broker = createFakeBroker();
+  const { runtime, baseUrl } = await startRuntime(t, { broker });
+  await seedPlaylist(runtime, baseUrl);
+
+  const start = (await jsonRequest(baseUrl, "/api/playback")).body;
+  assert.equal(start.mode, "on");
+
+  broker.deliver("announcements/mode", "off");
+  const afterMqtt = (await jsonRequest(baseUrl, "/api/playback")).body;
+  assert.equal(afterMqtt.mode, "off", "MQTT committed through the playback service");
+  assert.equal(afterMqtt.playbackRevision, start.playbackRevision + 1);
+
+  await jsonRequest(baseUrl, "/api/playback/mode", { method: "PUT", body: { mode: "on" } });
+  const afterRest = (await jsonRequest(baseUrl, "/api/playback")).body;
+  assert.equal(afterRest.mode, "on");
+
+  // Both transports move the same cursor with identical semantics.
+  const beforeNext = (await jsonRequest(baseUrl, "/api/playback")).body;
+  broker.deliver("announcements/next", "");
+  const mqttNext = (await jsonRequest(baseUrl, "/api/playback")).body;
+  assert.notEqual(mqttNext.currentEntryId, beforeNext.currentEntryId);
+  const restNext = (await jsonRequest(baseUrl, "/api/playback/next", { method: "POST" })).body;
+  assert.equal(restNext.currentEntryId, beforeNext.currentEntryId, "next wrapped back");
+
+  broker.deliver("announcements/pause", "");
+  assert.equal((await jsonRequest(baseUrl, "/api/playback")).body.transportState, "paused");
+  broker.deliver("announcements/resume", "");
+  assert.equal((await jsonRequest(baseUrl, "/api/playback")).body.transportState, "playing");
+  broker.deliver("announcements/static", "asset-one");
+  assert.equal(
+    (await jsonRequest(baseUrl, "/api/playback")).body.transportState,
+    "playing",
+    "a non-UUID static payload is rejected before it reaches playback",
+  );
+});
+
+test("MQTT publishes retained authoritative state only after a committed change", async (t) => {
+  const broker = createFakeBroker();
+  const { runtime, baseUrl } = await startRuntime(t, { broker });
+  await seedPlaylist(runtime, baseUrl);
+
+  assert.equal(
+    broker.published.find((message) => message.topic === "announcements/availability").payload,
+    "online",
+  );
+  broker.published.length = 0;
+
+  await jsonRequest(baseUrl, "/api/playback/mode", { method: "PUT", body: { mode: "off" } });
+  assert.equal(
+    broker.published.filter((message) => message.topic === "announcements/mode/state").at(-1)
+      .payload,
+    "off",
+  );
+
+  // A read and a no-op command commit nothing, so nothing is republished.
+  broker.published.length = 0;
+  await jsonRequest(baseUrl, "/api/playback");
+  await jsonRequest(baseUrl, "/api/playback/mode", { method: "PUT", body: { mode: "off" } });
+  assert.equal(broker.published.length, 0);
+
+  const commandTopics = new Set(runtime.mqttService.commandTopics);
+  assert.ok(
+    broker.published.every((message) => !commandTopics.has(message.topic)),
+    "state is never echoed onto a command topic",
+  );
+});
+
+test("a broker outage degrades status while playback and HTTP continue", async (t) => {
+  const broker = createFakeBroker();
+  const { runtime, baseUrl } = await startRuntime(t, { broker });
+  await seedPlaylist(runtime, baseUrl);
+
+  broker.emit("error", new Error("connect ECONNREFUSED"));
+  broker.emit("close");
+
+  const health = await jsonRequest(baseUrl, "/api/health");
+  assert.equal(health.response.status, 200, "MQTT is never readiness-critical");
+  assert.equal(health.body.services.mqtt.status, "disconnected");
+  assert.equal(health.body.status, "degraded");
+  assert.equal((await jsonRequest(baseUrl, "/health/ready")).response.status, 200);
+
+  // Playback keeps working through REST while the broker is unreachable.
+  const advanced = await jsonRequest(baseUrl, "/api/playback/next", { method: "POST" });
+  assert.equal(advanced.response.status, 200);
+  assert.ok(advanced.body.currentEntryId);
+  assert.equal(runtime.playbackService.getSnapshot().transportState, "playing");
+});
+
+test("an MQTT publish failure after commit never rolls back playback", async (t) => {
+  const broker = createFakeBroker({ failPublish: true });
+  const { runtime, baseUrl } = await startRuntime(t, { broker });
+  await seedPlaylist(runtime, baseUrl);
+
+  const before = (await jsonRequest(baseUrl, "/api/playback")).body;
+  const changed = await jsonRequest(baseUrl, "/api/playback/mode", {
+    method: "PUT",
+    body: { mode: "off" },
+  });
+
+  assert.equal(changed.response.status, 200);
+  assert.equal(changed.body.mode, "off");
+  assert.equal(changed.body.playbackRevision, before.playbackRevision + 1);
+  assert.equal(
+    runtime.database
+      .prepare("SELECT mode FROM playback_state WHERE output_id = 'global'")
+      .get().mode,
+    "off",
+    "the SQLite commit is authoritative regardless of MQTT",
+  );
+});
+
+test("an unreachable Home Assistant never blocks playback or the API", async (t) => {
+  const broker = createFakeBroker();
+  const { runtime, baseUrl } = await startRuntime(t, {
+    broker,
+    notificationFetch: () => Promise.reject(new Error("ECONNREFUSED")),
+    storage: testStorage({
+      status: "degraded",
+      code: "STORAGE_UNAVAILABLE",
+      checkedAt: null,
+      lastSuccessAt: null,
+    }),
+  });
+  await seedPlaylist(runtime, baseUrl);
+
+  runtime.notifications.observeFailure("storage", {
+    event: "storage.unavailable",
+    severity: "critical",
+    message: "Announcement storage is unavailable.",
+    context: {},
+  });
+  await runtime.notifications.deliverDue();
+
+  assert.equal(
+    runtime.database
+      .prepare("SELECT COUNT(*) AS count FROM notification_outbox WHERE delivered_at IS NULL")
+      .get().count,
+    1,
+    "the failed delivery stays pending for retry",
+  );
+  const playback = await jsonRequest(baseUrl, "/api/playback/next", { method: "POST" });
+  assert.equal(playback.response.status, 200);
+  assert.equal((await jsonRequest(baseUrl, "/health/ready")).response.status, 200);
+});
+
+test("display heartbeats flow over Socket.IO into the status API without touching playback", async (t) => {
+  const broker = createFakeBroker();
+  const { runtime, baseUrl } = await startRuntime(t, { broker });
+  await seedPlaylist(runtime, baseUrl);
+  const socket = await connectSocket(baseUrl);
+  t.after(() => socket.close().catch(() => {}));
+
+  const before = (await jsonRequest(baseUrl, "/api/playback")).body;
+  const playlistBefore = (await jsonRequest(baseUrl, "/api/playlists")).body;
+
+  await socket.send("display:heartbeat", {
+    protocolVersion: 1,
+    clientId: "lobby-display",
+    outputId: "global",
+    activePlaylistId: before.activePlaylistId,
+    playlistRevision: 1,
+    playbackRevision: before.playbackRevision,
+    currentEntryId: before.currentEntryId,
+    currentAssetId: "asset-one",
+    cachedAssetCount: 2,
+    playableAssetCount: 2,
+    mostRecentError: null,
+  });
+  await socket.send("display:heartbeat", { protocolVersion: 99, clientId: "bogus" });
+  await socket.send("display:heartbeat", "not-an-object");
+  await new Promise((resolve) => setTimeout(resolve, 100));
+
+  const displays = await jsonRequest(baseUrl, "/api/displays");
+  assert.equal(displays.response.status, 200);
+  assert.equal(displays.body.displays.length, 1, "malformed telemetry created no display");
+  const display = displays.body.displays[0];
+  assert.equal(display.clientId, "lobby-display");
+  assert.equal(display.status, "online");
+  assert.equal(display.cachedAssetCount, 2);
+  assert.equal(display.mostRecentError, null);
+
+  assert.equal(
+    (await jsonRequest(baseUrl, "/api/playback")).body.playbackRevision,
+    before.playbackRevision,
+    "telemetry never increments the playback revision",
+  );
+  assert.deepEqual(
+    (await jsonRequest(baseUrl, "/api/playlists")).body,
+    playlistBefore,
+    "telemetry never increments a playlist revision",
+  );
+
+  const health = await jsonRequest(baseUrl, "/api/health");
+  assert.equal(health.body.services.displays.known, 1);
+  assert.equal(health.body.services.displays.online, 1);
+});
+
+test("a first display heartbeat is announced once as a display:status event", async (t) => {
+  const broker = createFakeBroker();
+  const { runtime, baseUrl } = await startRuntime(t, { broker });
+  await seedPlaylist(runtime, baseUrl);
+  const socket = await connectSocket(baseUrl);
+  t.after(() => socket.close().catch(() => {}));
+
+  const beat = {
+    protocolVersion: 1,
+    clientId: "narthex-display",
+    outputId: "global",
+    cachedAssetCount: 1,
+    playableAssetCount: 1,
+  };
+  await socket.send("display:heartbeat", beat);
+  await socket.send("display:heartbeat", beat);
+  await socket.send("display:heartbeat", beat);
+  await new Promise((resolve) => setTimeout(resolve, 100));
+
+  const events = socketEvents(await socket.poll()).filter(
+    ([name]) => name === "display:status",
+  );
+  assert.equal(events.length, 1, "steady telemetry does not create realtime churn");
+  assert.equal(events[0][1].reason, "first_seen");
+  assert.equal(events[0][1].clientId, "narthex-display");
+  assert.doesNotMatch(JSON.stringify(events[0][1]), /socket|sid/i);
+});
+
+test("detailed status reports every integration without leaking any secret", async (t) => {
+  const broker = createFakeBroker();
+  const { baseUrl } = await startRuntime(t, { broker });
+
+  const health = await jsonRequest(baseUrl, "/api/health");
+  const status = await jsonRequest(baseUrl, "/api/status");
+  assert.equal(health.body.services.mqtt.status, "connected");
+  assert.equal(health.body.services.mqtt.configured, true);
+  assert.equal(health.body.services.mqtt.topicPrefix, "announcements");
+  assert.equal(health.body.services.notifications.configured, true);
+  assert.equal(health.body.services.notifications.status, "ok");
+  assert.equal(health.body.services.displays.status, "none_seen");
+  assert.equal(health.body.services.playback.status, "ok");
+  assert.equal(status.response.status, 200);
+
+  const serialized = JSON.stringify(health.body) + JSON.stringify(status.body);
+  assert.doesNotMatch(serialized, /mqtt-password-that-must-not-leak/);
+  assert.doesNotMatch(serialized, /home-assistant:8123/);
+  assert.doesNotMatch(serialized, /private-id/);
+  assert.doesNotMatch(serialized, /webhook.?url/i);
+  assert.doesNotMatch(serialized, /broker:1883/);
+});