From ae70bf1dd8bee2eaede5af256cc6168c782a1a9f Mon Sep 17 00:00:00 2001 From: Media Date: Sun, 16 Aug 2026 22:52:59 -0500 Subject: [PATCH] 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 --- .env.example | 17 +- README.md | 188 +++++++++++ docker-compose.yaml | 5 + pnpm-lock.yaml | 246 ++++++++++++++ server/package.json | 1 + server/src/app.js | 16 + server/src/assets/service.js | 16 + server/src/config/index.js | 48 ++- server/src/displays/routes.js | 19 ++ server/src/displays/service.js | 327 +++++++++++++++++++ server/src/health/routes.js | 76 ++++- server/src/mqtt/index.js | 314 ++++++++++++++++++ server/src/notifications/conditions.js | 88 +++++ server/src/notifications/index.js | 357 ++++++++++++++++++++ server/src/runtime.js | 113 ++++++- server/src/storage/index.js | 29 +- server/test/displays.test.js | 376 +++++++++++++++++++++ server/test/integrations.test.js | 430 +++++++++++++++++++++++++ server/test/mqtt-broker.test.js | 250 ++++++++++++++ server/test/mqtt.test.js | 403 +++++++++++++++++++++++ server/test/notifications.test.js | 383 ++++++++++++++++++++++ server/test/storage-health.test.js | 13 +- 22 files changed, 3691 insertions(+), 24 deletions(-) create mode 100644 server/src/displays/routes.js create mode 100644 server/src/displays/service.js create mode 100644 server/src/mqtt/index.js create mode 100644 server/src/notifications/conditions.js create mode 100644 server/src/notifications/index.js create mode 100644 server/test/displays.test.js create mode 100644 server/test/integrations.test.js create mode 100644 server/test/mqtt-broker.test.js create mode 100644 server/test/mqtt.test.js create mode 100644 server/test/notifications.test.js diff --git a/.env.example b/.env.example index cb0ebaf..c694522 100644 --- a/.env.example +++ b/.env.example @@ -38,9 +38,24 @@ UPLOAD_MAX_FILES=20 IMAGE_MAX_INPUT_PIXELS=40000000 # UPLOAD_TEMP_DIRECTORY=/tmp/announcements-uploads -# Optional integrations. Leave blank to disable them during Phase 2. +# Optional integrations. Leave MQTT_URL blank to disable the MQTT integration +# and HA_WEBHOOK_URL blank to disable outgoing notifications entirely. MQTT_URL= MQTT_USERNAME= MQTT_PASSWORD= +# With the default prefix the command topics are announcements/mode, +# announcements/next, announcements/previous, announcements/pause, +# announcements/resume, announcements/reload, and announcements/static. +MQTT_TOPIC_PREFIX=announcements +MQTT_RECONNECT_PERIOD_MS=5000 +MQTT_CONNECT_TIMEOUT_MS=10000 + +# Outgoing Home Assistant webhook. The URL is a secret: it is never returned by +# any API response or status document. HA_WEBHOOK_URL= HA_WEBHOOK_SECRET= +HA_WEBHOOK_TIMEOUT_MS=10000 + +# A display is reported offline after this long without a valid heartbeat. +# The display client is expected to send one every 10-15 seconds. +DISPLAY_OFFLINE_THRESHOLD_MS=45000 diff --git a/README.md b/README.md index 443110a..035a003 100644 --- a/README.md +++ b/README.md @@ -181,6 +181,186 @@ exit static safely, return to normal rotation, and only then delete the asset. Phase 3B does not depend on the refusal, so the change belongs to the later phase that revisits asset lifecycle. +## MQTT integration + +The API connects to `MQTT_URL` when it is configured and stays disabled when it +is not. Commands are dispatched straight into the authoritative playback +service, so an MQTT command and its REST equivalent are the same operation with +the same semantics; no playback logic is duplicated in the integration. + +With the default `MQTT_TOPIC_PREFIX` of `announcements` the command topics are: + +| Topic | Payload | QoS | +| ------------------------- | ---------------------- | --- | +| `announcements/mode` | `on` or `off` | 1 | +| `announcements/pause` | ignored | 1 | +| `announcements/resume` | ignored | 1 | +| `announcements/reload` | ignored | 1 | +| `announcements/static` | LibraryAsset UUID | 1 | +| `announcements/next` | ignored | 0 | +| `announcements/previous` | ignored | 0 | + +Mode payloads are trimmed and case-insensitive. `off` is only the visual black +gate; it never pauses, restarts, or resets the playlist, cursor, static hold, or +timing. Static accepts only a stable LibraryAsset UUID: URLs, filenames, and S3 +object keys are rejected. The asset must exist, be `ready`, and be unexpired, +but it does not need to appear in any playlist. A rejected command is logged as +a structured warning, changes nothing, and publishes no state. + +`next` and `previous` stay at QoS 0 on purpose. They are not idempotent and +carry no command ID, so an at-least-once redelivery would silently skip a +slide. Commands are never retained, and the server never publishes onto a +command topic. + +The server publishes retained state on separate topics: + +| Topic | Payload | +| --------------------------------- | ---------------------------------- | +| `announcements/availability` | `online` / `offline` | +| `announcements/mode/state` | `on` / `off` | +| `announcements/playback/state` | small versioned JSON snapshot | + +Availability describes this API's MQTT integration, not a display. A registered +last will and testament lets the broker publish `offline` when the API +disappears without a clean disconnect; a planned shutdown retracts it directly. +The compact playback topic carries mode, transport, active playlist, current +entry and asset, static asset, and revision. It never carries the Socket.IO +manifest or any private storage detail. + +State is published only after a playback mutation has committed, and MQTT +publication happens after the SQLite commit. A publish failure degrades the +integration and is never allowed to roll playback back. A broker outage leaves +the API, HTTP, and playback untouched: the client reconnects on its own, then +resubscribes and republishes availability and authoritative state. + +## Home Assistant notifications + +Announcements sends outgoing generic webhook POSTs to `HA_WEBHOOK_URL`. Home +Assistant is only the receiver; Announcements never calls Companion directly and +needs no Home Assistant authentication beyond the configured webhook. The URL +and `HA_WEBHOOK_SECRET` are configuration, never API or status output. + +Notifications are durable and deduplicated through `notification_incidents` and +`notification_outbox`. A condition that starts failing opens one incident and +queues exactly one failure notification. While it keeps failing only its +observation time is updated, so a thirty-minute outage still sends one message. +Recovery closes the incident and queues exactly one recovery notification. If +the same condition fails again later that is a new episode with its own +failure/recovery pair. Open incidents and undelivered notifications survive an +API restart. + +Reported conditions are `storage.unavailable` / `storage.recovered`, driven by +the Phase 2 storage health transition rather than individual request failures; +`display.offline` / `display.recovered` per display client; and +`playback.no_playable_slides` / `playback.playable` from committed playback +state. A static override does not suppress the playback condition, because +holding one asset does not repair an empty playlist. A failed upload opens one +`asset.processing_failed` warning per asset; an individual failed upload has no +automatic recovery, so the operator resolves it by deleting or re-uploading. + +The payload is versioned: + +```json +{ + "version": 1, + "notificationId": "stable-delivery-id", + "incidentId": "stable-incident-id", + "event": "storage.unavailable", + "severity": "critical", + "state": "failure", + "message": "Announcement storage is unavailable.", + "timestamp": "2026-08-16T12:00:00.000Z", + "context": { "conditionKey": "storage", "episode": 1 } +} +``` + +Severities are `info`, `warning`, `critical`, and `recovery`. Home Assistant can +treat them all as ordinary notifications; the field exists so later Ekklesia or +HA handling can discriminate. + +Delivery is asynchronous and bounded. A 2xx marks the notification delivered. +Any other response and any transport failure leaves it pending and retries after +30 seconds, 1 minute, 5 minutes, 15 minutes, and 30 minutes, then every 30 +minutes. The notification ID is stable across retries so a receiver can +deduplicate, retries resume after a restart, and a failing webhook never blocks +playback, MQTT, storage, or heartbeats. When no webhook URL is configured, +incidents are still recorded but no delivery work is queued, and status reports +the integration as disabled. + +An example Home Assistant receiver: + +```yaml +automation: + - alias: Announcements Notifications + triggers: + - trigger: webhook + webhook_id: announcements + allowed_methods: + - POST + local_only: true + + actions: + - action: notify.mobile_app_YOUR_DEVICE + data: + title: "Announcements" + message: "{{ trigger.json.message }}" +``` + +Substitute any notify target; Companion is not required. Announcements cannot +send a webhook while the Announcements API itself is completely down. That is a +limitation of self-reporting, not a bug: if server-down alerting is wanted, an +external monitor such as Home Assistant should poll `/health/live`. + +## Display telemetry + +The future `/display` client reports read-only telemetry over the Socket.IO +event `display:heartbeat`, expected every 10-15 seconds: + +```json +{ + "protocolVersion": 1, + "clientId": "persistent-display-id", + "outputId": "global", + "activePlaylistId": "playlist-id", + "playlistRevision": 12, + "playbackRevision": 42, + "currentEntryId": "entry-id", + "currentAssetId": "asset-id", + "cachedAssetCount": 10, + "playableAssetCount": 10, + "mostRecentError": null +} +``` + +Every field is validated and bounded, `mostRecentError` is truncated, and +malformed telemetry is ignored without creating a display. Heartbeats can never +mutate playback and never increment a playback or playlist revision. Identity is +the persistent `clientId` stored in `display_clients`, never a Socket.IO +connection ID, so a browser reconnect remains the same display. + +A display is online while heartbeats are arriving and offline once +`DISPLAY_OFFLINE_THRESHOLD_MS` (45 seconds, about three missed heartbeats) has +passed since the last one. A closed socket alone never marks a display offline, +because reconnects and brief network hiccups are normal. Crossing the threshold +opens one `display.offline` incident that does not repeat while the display +stays away, and the next valid heartbeat recovers it exactly once. + +Display history is persisted, so a restart does not erase monitoring. The API +cannot know what happened to a display while the API itself was down, so it is +deliberately conservative: nothing is reported offline until a full threshold has +elapsed since startup. A healthy display that reconnects during that window is +never reported, a display that is still missing afterwards is reported once, and +an incident that was already open before the restart is not duplicated. + +`GET /api/displays` returns each known display with a status derived from +heartbeat freshness rather than mere database existence, and exposes no socket +internals. `display:status` is emitted only for meaningful changes — first seen, +offline, recovered, or a changed error — so steady telemetry creates no realtime +churn. `mostRecentError` is telemetry only; arbitrary browser strings never +become Home Assistant notifications. Structured display-side errors such as +decode failures arrive with the Phase 5 renderer, which will report them through +this same field and event. + ## Storage and health The API uses a small path-style S3 adapter configured by `S3_ENDPOINT`, @@ -193,3 +373,11 @@ Storage outages do not stop the API or fail process readiness. `/health/ready` reports storage as degraded while remaining HTTP 200 when SQLite and the API runtime are healthy. `/api/health` and `/api/status` expose cached, redacted storage state and recover after a successful probe or storage operation. + +`/health/live` reports only that the process is running. Readiness stays limited +to SQLite, migrations, and playback initialization. Every external integration +degrades the reported status instead of failing readiness, so a broker, webhook +receiver, or display problem can never put the API into a restart loop. +`/api/health` and `/api/status` report redacted storage, MQTT, notification, +playback, and display state. No MQTT password, S3 credential, webhook URL, or +webhook secret appears in any response. diff --git a/docker-compose.yaml b/docker-compose.yaml index d7f449a..6b2780e 100644 --- a/docker-compose.yaml +++ b/docker-compose.yaml @@ -53,8 +53,13 @@ services: MQTT_URL: "${MQTT_URL:-}" MQTT_USERNAME: "${MQTT_USERNAME:-}" MQTT_PASSWORD: "${MQTT_PASSWORD:-}" + MQTT_TOPIC_PREFIX: "${MQTT_TOPIC_PREFIX:-announcements}" + MQTT_RECONNECT_PERIOD_MS: "${MQTT_RECONNECT_PERIOD_MS:-5000}" + MQTT_CONNECT_TIMEOUT_MS: "${MQTT_CONNECT_TIMEOUT_MS:-10000}" HA_WEBHOOK_URL: "${HA_WEBHOOK_URL:-}" HA_WEBHOOK_SECRET: "${HA_WEBHOOK_SECRET:-}" + HA_WEBHOOK_TIMEOUT_MS: "${HA_WEBHOOK_TIMEOUT_MS:-10000}" + DISPLAY_OFFLINE_THRESHOLD_MS: "${DISPLAY_OFFLINE_THRESHOLD_MS:-45000}" volumes: - app-data:/data depends_on: diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 73526fa..0f3da87 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -62,6 +62,9 @@ importers: express: specifier: ^5.2.1 version: 5.2.1 + mqtt: + specifier: 5.15.2 + version: 5.15.2 multer: specifier: 2.0.2 version: 2.0.2 @@ -201,6 +204,10 @@ packages: engines: {node: '>=6.0.0'} hasBin: true + '@babel/runtime@7.29.7': + resolution: {integrity: sha512-Nq8OhGWiZIZGV6hLHoyAKLLcJihP/xFeBMGJoUrxTX2psI8dCifzLhZISFb+VWS3wFMRDmCGw5R+dOySCqPLhw==} + engines: {node: '>=6.9.0'} + '@babel/template@7.29.7': resolution: {integrity: sha512-puq+Gf35oI24FeN11LkoUQFqv9uwNeWpxXZi/Ji3rRIoKAzKnxRaZ+Gkj0vKS9ZCiTESfng1N9LyOyXvo+m+Gg==} engines: {node: '>=6.9.0'} @@ -599,6 +606,9 @@ packages: '@types/react@19.2.18': resolution: {integrity: sha512-AnzbBERsrLKtk2XSfTbYRLjQPdy116Sty4q+T+Bp3IC4l6jNBvreVPAHmpq9qhXQM7CXZPjLVmGMw9sy+hxQ3w==} + '@types/readable-stream@4.0.24': + resolution: {integrity: sha512-NRvUNC/JFGPJvqdAfEve8oginbM6V08u5NzLWpG8MwA2kTPOLnqk+wpwuPT+mp3aUsxyuT6m2gnrPuHYCruzEg==} + '@types/ws@8.18.1': resolution: {integrity: sha512-ThVF6DCVhA8kUGy+aazFQ4kXQ7E1Ty7A3ypFOe0IcJV8O/M511G99AW24irKrW56Wt44yG9+ij8FaqoBGkuBXg==} @@ -615,6 +625,10 @@ packages: babel-plugin-react-compiler: optional: true + abort-controller@3.0.0: + resolution: {integrity: sha512-h8lQ8tacZYnR3vNQTgibj+tODHI5/+l06Au2Pcriv/Gmet0eaj4TwWH41sO9wnHDiQsEj19q0drzdWdeAHtweg==} + engines: {node: '>=6.5'} + accepts@1.3.8: resolution: {integrity: sha512-PYAthTa2m2VKxuvSD3DPC/Gy+U+sOA1LAuT8mkmRuvw+NACSaeXEQ+NHcVF7rONl6qcaxV3Uuemwawk+7+SJLw==} engines: {node: '>= 0.6'} @@ -643,6 +657,9 @@ packages: resolution: {integrity: sha512-BLrgEcRTwX2o6gGxGOCNyMvGSp35YofuYzw9h1IMTRmKqttAZZVU67bdb9Pr2vUHA8+j3i2tJfjO6C6+4myGTA==} engines: {node: 18 || 20 || >=22} + base64-js@1.5.1: + resolution: {integrity: sha512-AKpaYlHn8t4SVbOHCy+b5+KKgvR4vrsD8vbvrbiQJps7fKDTkjkDry6ji0rUJjC0kzbNePLwzxq8iypo41qeWA==} + base64id@2.0.0: resolution: {integrity: sha512-lGe34o6EHj9y3Kts9R4ZYs/Gr+6N7MCaMlIFA3F1R2O5/m7K06AxfSeO5530PEERE6/WyEg3lsuyw4GHlPZHog==} engines: {node: ^4.5.0 || >= 5.9} @@ -656,6 +673,9 @@ packages: resolution: {integrity: sha512-jW6oufeDhXZaiX9Lw5A+oerVClx4iFrI6uDj1zu7SqUAjak9vbJvA0NEcKLNxHiQHb6kYCoFzzXYV0YOauhV3g==} engines: {node: '>=22'} + bl@6.1.6: + resolution: {integrity: sha512-jLsPgN/YSvPUg9UX0Kd73CXpm2Psg9FxMeCSXnk3WBO3CMT10JMwijubhGfHCnFu6TPn1ei3b975dxv7K2pWVg==} + body-parser@2.2.2: resolution: {integrity: sha512-oP5VkATKlNwcgvxi0vM0p/D3n2C3EReYVX+DNYs5TjZFn/oQt2j+4sVJtSMr18pdRr8wjTcBl6LoV+FUwzPmNA==} engines: {node: '>=18'} @@ -667,6 +687,9 @@ packages: resolution: {integrity: sha512-kLpxurY4Z4r9sgMsyG0Z9uzsBlgiU/EFKhj/h91/8yHu0edo7XuixOIH3VcJ8kkxs6/jPzoI6U9Vj3WqbMQ94g==} engines: {node: 18 || 20 || >=22} + broker-factory@3.1.15: + resolution: {integrity: sha512-ko+aWvgNuP49meGrdjUu7rC+Y+Wai3cCPxP3xWwHsHfehFjOh5ZQM2yC4gEB2UddeZ/YXhm0K1eG/L6fxym2Og==} + browserslist@4.28.8: resolution: {integrity: sha512-V2NpofLblG64mfOtSgDhOJESZEGogzDMBv/q+W6oc4LXWP/q75eOXoOaaOu1EOadB9U4Bwx/e0yzbvwKH8zalA==} engines: {node: ^6 || ^7 || ^8 || ^9 || ^10 || ^11 || ^12 || >=13.7} @@ -675,6 +698,9 @@ packages: buffer-from@1.1.2: resolution: {integrity: sha512-E+XQCRwSbaaiChtv6k6Dwgc+bx+Bs6vuKJHHl5kox/BaKbhiXzqQOwK4cO22yElGp2OCmjwVhT3HmxgyPGnJfQ==} + buffer@6.0.3: + resolution: {integrity: sha512-FTiCpNxtwiZZHEZbcbTIcZjERVICn9yq/pDFkTl95/AxzD1naBctN7YO68riM/gLSDY7sdrMby8hofADYuuqOA==} + busboy@1.6.0: resolution: {integrity: sha512-8SFQbg/0hQ9xy3UNTB0YEnsNBbWfhf7RtnzpL7TkBiTBRfrQ9Fxcnz7VJsleJpyp6rVLvXiuORqjlHi5q+PYuA==} engines: {node: '>=10.16.0'} @@ -694,6 +720,9 @@ packages: caniuse-lite@1.0.30001809: resolution: {integrity: sha512-xxWVywk6a6Arlk+hymeycyn/VgqEfLDxupvhH/xiY5SJ/18kmi9o6MiO320DCUzypORHLtvh0I4i04tUhCNHNQ==} + commist@3.2.0: + resolution: {integrity: sha512-4PIMoPniho+LqXmpS5d3NuGYncG6XWlkBSVGiWycL22dd42OYdUGil2CWuzklaJoNxyxUSpO4MKIBU94viWNAw==} + concat-stream@2.0.0: resolution: {integrity: sha512-MWufYdFw53ccGjCA+Ol7XJYpAlW6/prSMzuPOTRnJGcGzuhLn4Scrz7qf6o8bROZ514ltazcIFJZevcfbo0x7A==} engines: {'0': node >= 6.0} @@ -861,6 +890,14 @@ packages: resolution: {integrity: sha512-aIL5Fx7mawVa300al2BnEE4iNvo1qETxLrPI/o05L7z6go7fCw1J6EQmbK4FmJ2AS7kgVF/KEZWufBfdClMcPg==} engines: {node: '>= 0.6'} + event-target-shim@5.0.1: + resolution: {integrity: sha512-i/2XbnSz/uxRCU6+NdVJgKWDTM427+MqYbkQzD321DuCQJUqOuJKIA0IM2+W2xtYHdKOmZ4dR6fExsd4SXL+WQ==} + engines: {node: '>=6'} + + events@3.3.0: + resolution: {integrity: sha512-mQw+2fkQbALzQ7V0MY0IqdnXNOeTtP4r0lN9z7AAawCXgqea7bDii20AYrIBrFd/Hx0M2Ocz6S111CaFkUcb0Q==} + engines: {node: '>=0.8.x'} + express@5.2.1: resolution: {integrity: sha512-hIS4idWWai69NezIdRt2xFVofaF4j+6INOpJlVOLDO8zXGpUVEVzIYk12UUi2JzjEzWL3IOAxcTubgz9Po0yXw==} engines: {node: '>= 18'} @@ -874,6 +911,10 @@ packages: fast-levenshtein@2.0.6: resolution: {integrity: sha512-DCXu6Ifhqcks7TZKY3Hxp3y6qphY5SJZmrWMDrKcERSOXWQdMhU9Ig/PYrzyw/ul9jOIyh0N4M0tbC5hodg8dw==} + fast-unique-numbers@9.0.27: + resolution: {integrity: sha512-nDA9ADeINN8SA2u2wCtU+siWFTTDqQR37XvgPIDDmboWQeExz7X0mImxuaN+kJddliIqy2FpVRmnvRZ+j8i1/A==} + engines: {node: '>=18.2.0'} + fdir@6.5.0: resolution: {integrity: sha512-tIbYtZbucOs0BRGqPJkshJUYdL+SDH7dVM8gjy+ERp3WAUjLEFJE+02kanyHtwjWOnwrKYBiwAmM0p4kLJAnXg==} engines: {node: '>=12.0.0'} @@ -950,6 +991,9 @@ packages: resolution: {integrity: sha512-ej4AhfhfL2Q2zpMmLo7U1Uv9+PyhIZpgQLGT1F9miIGmiCJIoCgSmczFdrc97mWT4kVY72KA+WnnhJ5pghSvSg==} engines: {node: '>= 0.4'} + help-me@5.0.0: + resolution: {integrity: sha512-7xgomUX6ADmcYzFik0HzAxh/73YlKR9bmFzf51CZwR+b6YtzU2m0u49hQCqV6SvlqIqsaxovfwdvbnsw3b/zpg==} + hermes-estree@0.25.1: resolution: {integrity: sha512-0wUoCcLp+5Ev5pDW2OriHC2MJCbwLwuRx+gAqMTOkGKJJiBCLjtrvy4PWUGn6MIVefecRpzoOZ/UV6iGdOr+Cw==} @@ -964,6 +1008,9 @@ packages: resolution: {integrity: sha512-im9DjEDQ55s9fL4EYzOAv0yMqmMBSZp6G0VvFyTMPKWxiSBHUj9NW/qqLmXUwXrrM7AvqSlTCfvqRb0cM8yYqw==} engines: {node: '>=0.10.0'} + ieee754@1.2.1: + resolution: {integrity: sha512-dcyqhDvX1C46lXZcVqCpK+FtMRQVdIMN6/Df5js2zouUsqG7I6sFxitIC+7KYK29KdXOLHdu9zL4sFnoVQnqaA==} + ignore@5.3.2: resolution: {integrity: sha512-hsBTNUqQTDwkWtcdYI2i06Y/nUBEsNEDJKjWdigLvegy8kDuJAS8uRlpkkcQpyEXL0Z/pjDy5HBmMjRCJ2gq+g==} engines: {node: '>= 4'} @@ -975,6 +1022,10 @@ packages: inherits@2.0.4: resolution: {integrity: sha512-k/vGaX4/Yla3WzyMCvTQOXYeIHvqOKtnqBduzTHpzpQZzAskKMhZ2K+EnBiSM9zGSoIFeMpXKxa4dYeZIQqewQ==} + ip-address@10.5.0: + resolution: {integrity: sha512-R5SnVLJmgYYvf2F2ZgwSBnelz5G4q5AxIC277GDfUaNbrZKNANcBC7RHqYYePlszf4kBolVkJauG0ZjHHFh55g==} + engines: {node: '>= 12'} + ipaddr.js@1.9.1: resolution: {integrity: sha512-0KI/607xoxSToH7GjN1FfSbLoU0+btTicjsQSWQlh/hZykN8KpmMf7uYwPW3R+akZ6R/w18ZlXSHBYXiYUPO3g==} engines: {node: '>= 0.10'} @@ -993,6 +1044,9 @@ packages: isexe@2.0.0: resolution: {integrity: sha512-RHxMLp9lnKHGHRng9QFhRCMbYAcVpn69smSGcq3f36xjgVVWThj4qqLbTLlq7Ssj8B+fIQ1EuCEGI2lKsyQeIw==} + js-sdsl@4.3.0: + resolution: {integrity: sha512-mifzlm2+5nZ+lEcLJMoBK0/IH/bDg8XnJfd/Wq6IP+xoCjLZsTOnV2QpxlVbX9bMnkl5PdEjNtBJ9Cj1NjifhQ==} + js-tokens@4.0.0: resolution: {integrity: sha512-RdJUflcE3cUzKiMqQgsCu06FPu9UdIJO0beYbPhHN4k6apgJtifcoCtT9bcxOpYBtpD2kCM6Sbzg4CausW/PKQ==} @@ -1100,6 +1154,9 @@ packages: resolution: {integrity: sha512-iPZK6eYjbxRu3uB4/WZ3EsEIMJFMqAoopl3R+zuq0UjcAm/MO6KCweDgPfP3elTztoKP3KtnVHxTn2NHBSDVUw==} engines: {node: '>=10'} + lru-cache@10.4.3: + resolution: {integrity: sha512-JNAzZcXrCt42VGLuYz0zfAzDfAvJWW6AfYlDBQyDV5DClI2m5sAmK+OIO7s59XfsRsWHp02jAJrRadPRGTt6SQ==} + lru-cache@5.1.1: resolution: {integrity: sha512-KpNARQA3Iwv+jTA0utUVVbrh+Jlrr1Fv0e56GGzAFOXN7dk/FviaDW8LHmK52DlcH4WP2n6gI8vN1aesBFgo9w==} @@ -1146,6 +1203,14 @@ packages: resolution: {integrity: sha512-FP+p8RB8OWpF3YZBCrP5gtADmtXApB5AMLn+vdyA+PyxCjrCs00mjyUozssO33cwDeT3wNGdLxJ5M//YqtHAJw==} hasBin: true + mqtt-packet@9.0.2: + resolution: {integrity: sha512-MvIY0B8/qjq7bKxdN1eD+nrljoeaai+qjLJgfRn3TiMuz0pamsIWY2bFODPZMSNmabsLANXsLl4EMoWvlaTZWA==} + + mqtt@5.15.2: + resolution: {integrity: sha512-VWZU2CSUY3U3oN0PSBRDE5SNsFi4zqqNeQ/uv3pZWqY3CrBXD/dhd0ZLjlsk5YnebGlrapi4lRVNJPUNQ5aZ3w==} + engines: {node: '>=16.0.0'} + hasBin: true + ms@2.1.3: resolution: {integrity: sha512-6FlzubTLZG3J2a/NVCAleEhjzq5oxgHyaCU9yYXvcLsvoVaHJq/s5xXI6/XXP6tz7R9xAOtHnSO/tXtF3WRTlA==} @@ -1177,6 +1242,9 @@ packages: resolution: {integrity: sha512-D9UOmYG3UH1V+ENW56t5QXBwJw1YEY18ruVeus89Rw+SyIgjPkCO84bRzO3uNIYosJbNwiabWVn48o3uJLjxFQ==} engines: {node: '>=18'} + number-allocator@1.0.14: + resolution: {integrity: sha512-OrL44UTVAvkKdOdRQZIJpLkAdjXGTRda052sN4sO77bKEzYYqWKMBjQvrJFzqygI99gL6Z4u2xctPW1tB8ErvA==} + object-assign@4.1.1: resolution: {integrity: sha512-rJgTQnkUnH1sFw8yT6VSU3zD3sWmu6sZhIseY8VX+GRu3P6F7Fu+JNDoXfklElbLJSnc3FUQHVe4cU5hj+BcUg==} engines: {node: '>=0.10.0'} @@ -1234,6 +1302,13 @@ packages: resolution: {integrity: sha512-vkcDPrRZo1QZLbn5RLGPpg/WmIQ65qoWWhcGKf/b5eplkkarX0m9z8ppCat4mlOqUsWpyNuYgO3VRyrYHSzX5g==} engines: {node: '>= 0.8.0'} + process-nextick-args@2.0.1: + resolution: {integrity: sha512-3ouUOpQhtgrbOa17J7+uxOTpITYWaGP7/AhoR3+A+/1e9skrzelGi/dXzEYyvbxubEF6Wn2ypscTKiKJFFn1ag==} + + process@0.11.10: + resolution: {integrity: sha512-cdGef/drWFoydD1JsMzuFf8100nZl+GT+yacc2bEced5f9Rjk4z+WtFUTBu9PhOi9j/jfmBPu0mMEY4wIdAF8A==} + engines: {node: '>= 0.6.0'} + proxy-addr@2.0.7: resolution: {integrity: sha512-llQsMLSUDUPT44jdrU/O37qlnifitDP+ZwrmmZcoSKyLKvtZxpyV0n2/bD/N4tBAAZ/gJEdZU7KMraoK1+XYAg==} engines: {node: '>= 0.10'} @@ -1267,6 +1342,13 @@ packages: resolution: {integrity: sha512-9u/sniCrY3D5WdsERHzHE4G2YCXqoG5FTHUiCC4SIbr6XcLZBY05ya9EKjYek9O5xOAwjGq+1JdGBAS7Q9ScoA==} engines: {node: '>= 6'} + readable-stream@4.7.0: + resolution: {integrity: sha512-oIGGmcpTLwPga8Bn6/Z75SVaH1z5dUut2ibSyAMVhmUggWpmDn2dapB0n7f8nwaSiRtepAsfJyfXIO5DCVAODg==} + engines: {node: ^12.22.0 || ^14.17.0 || >=16.0.0} + + rfdc@1.4.1: + resolution: {integrity: sha512-q1b3N5QkRUWUl7iyylaaj3kOpIT0N2i9MqIEQXP73GVsN9cw3fdx8X63cEmWhJGi2PPCF23Ijp7ktmd39rawIA==} + rolldown@1.2.4: resolution: {integrity: sha512-rSr7irW0K7QRWzjdJXqZowkcRdDtjRduh43rBltnVKd0VFq839l1lJoDvGJb6gl7+4rTTCrPWu+YfujUL8Ug7w==} engines: {node: ^20.19.0 || >=22.12.0} @@ -1338,6 +1420,10 @@ packages: resolution: {integrity: sha512-ZX99e6tRweoUXqR+VBrslhda51Nh5MTQwou5tnUDgbtyM0dBgmhEDtWGP/xbKn6hqfPRHujUNwz5fy/wbbhnpw==} engines: {node: '>= 0.4'} + smart-buffer@4.2.0: + resolution: {integrity: sha512-94hK0Hh8rPqQl2xXc3HsaBoOXKV20MToPkcXvwbISWLEs+64sBq5kFgn2kJDHb1Pry9yrP0dxrCI9RRci7RXKg==} + engines: {node: '>= 6.0.0', npm: '>= 3.0.0'} + socket.io-adapter@2.5.6: resolution: {integrity: sha512-DkkO/dz7MGln0dHn5bmN3pPy+JmywNICWrJqVWiVOyvXjWQFIv9c2h24JrQLLFJ2aQVQf/Cvl1vblnd4r2apLQ==} @@ -1353,10 +1439,18 @@ packages: resolution: {integrity: sha512-2Dd78bqzzjE6KPkD5fHZmDAKRNe3J15q+YHDrIsy9WEkqttc7GY+kT9OBLSMaPbQaEd0x1BjcmtMtXkfpc+T5A==} engines: {node: '>=10.2.0'} + socks@2.8.9: + resolution: {integrity: sha512-LJhUYUvItdQ0LkJTmPeaEObWXAqFyfmP85x0tch/ez9cahmhlBBLbIqDFnvBnUJGagb0JbIQrkBs1wJ+yRYpEw==} + engines: {node: '>= 10.0.0', npm: '>= 3.0.0'} + source-map-js@1.2.1: resolution: {integrity: sha512-UXWMKhLOwVKb728IUtQPXxfYU+usdybtUrK/8uGE8CQMvrhOpwvzDBwj0QhSL7MQc7vIsISBG8VQ8+IDQxpfQA==} engines: {node: '>=0.10.0'} + split2@4.2.0: + resolution: {integrity: sha512-UcjcJOWknrNkF6PLX83qcHM6KHgVKNkV62Y8a5uYDVv9ydGQVwAHMKqHdJje1VTWpljG0WYpCDhrCdAOYH4TWg==} + engines: {node: '>= 10.x'} + statuses@2.0.2: resolution: {integrity: sha512-DvEy55V3DB7uknRo+4iOGT5fP1slR8wQohVdknigZPMpMstaKJQWhwiYBACJE3Ul2pTnATihhBYnRhZQHGBiRw==} engines: {node: '>= 0.8'} @@ -1469,6 +1563,18 @@ packages: resolution: {integrity: sha512-BN22B5eaMMI9UMtjrGd5g5eCYPpCPDUy0FJXbYsaT5zYxjFOckS53SQDE3pWkVoWpHXVb3BrYcEN4Twa55B5cA==} engines: {node: '>=0.10.0'} + worker-factory@7.0.50: + resolution: {integrity: sha512-hhwc0G+sFwM4qBuhJIUBn2p1Jf8v/FwmLUANBf/Q+Lt2uI8mfIZQhXaZQACodQD4R7Zp6cn/6702bIvNn2puJQ==} + + worker-timers-broker@8.0.18: + resolution: {integrity: sha512-FrjzDVX1wKfZN0gRbCFqv8VHuTncG4sbI/WGEg4tSSQeIsnwqg4YBYWMAHYLJtUDEYYmiK65UKFrVMVk2irDSg==} + + worker-timers-worker@9.0.15: + resolution: {integrity: sha512-KKUe7lZ/Aignr51H6hOUik8LwTnIgojH/1lwhli8A8qIEIyewogZTpNpMW5B6BF7nmwBOkUoTYgf1H3QShcjSA==} + + worker-timers@8.0.34: + resolution: {integrity: sha512-WXL+Dqsm0G6dnC66rQsvM3tPT2adhbqSirWUCZGglkALOr8ocS0KlpU0iuKbjflJOlu3ydZikMEVrYnHOM9suw==} + wrappy@1.0.2: resolution: {integrity: sha512-l4Sp/DRseor9wL6EvV2+TuQn63dMkPjZ/sp9XkghTEbV9KlPS1xUsZ3u7/IQO4wxtcFB4bgpQPRcR3QCvezPcQ==} @@ -1764,6 +1870,8 @@ snapshots: dependencies: '@babel/types': 7.29.8 + '@babel/runtime@7.29.7': {} + '@babel/template@7.29.7': dependencies: '@babel/code-frame': 7.29.7 @@ -2070,6 +2178,10 @@ snapshots: dependencies: csstype: 3.2.3 + '@types/readable-stream@4.0.24': + dependencies: + '@types/node': 25.8.0 + '@types/ws@8.18.1': dependencies: '@types/node': 25.8.0 @@ -2079,6 +2191,10 @@ snapshots: '@rolldown/pluginutils': 1.0.1 vite: 8.2.1(@types/node@25.8.0) + abort-controller@3.0.0: + dependencies: + event-target-shim: 5.0.1 + accepts@1.3.8: dependencies: mime-types: 2.1.35 @@ -2106,6 +2222,8 @@ snapshots: balanced-match@4.0.4: {} + base64-js@1.5.1: {} + base64id@2.0.0: {} baseline-browser-mapping@2.11.14: {} @@ -2114,6 +2232,13 @@ snapshots: dependencies: node-addon-api: 8.9.2 + bl@6.1.6: + dependencies: + '@types/readable-stream': 4.0.24 + buffer: 6.0.3 + inherits: 2.0.4 + readable-stream: 4.7.0 + body-parser@2.2.2: dependencies: bytes: 3.1.2 @@ -2134,6 +2259,13 @@ snapshots: dependencies: balanced-match: 4.0.4 + broker-factory@3.1.15: + dependencies: + '@babel/runtime': 7.29.7 + fast-unique-numbers: 9.0.27 + tslib: 2.8.1 + worker-factory: 7.0.50 + browserslist@4.28.8: dependencies: baseline-browser-mapping: 2.11.14 @@ -2144,6 +2276,11 @@ snapshots: buffer-from@1.1.2: {} + buffer@6.0.3: + dependencies: + base64-js: 1.5.1 + ieee754: 1.2.1 + busboy@1.6.0: dependencies: streamsearch: 1.1.0 @@ -2162,6 +2299,8 @@ snapshots: caniuse-lite@1.0.30001809: {} + commist@3.2.0: {} + concat-stream@2.0.0: dependencies: buffer-from: 1.1.2 @@ -2344,6 +2483,10 @@ snapshots: etag@1.8.1: {} + event-target-shim@5.0.1: {} + + events@3.3.0: {} + express@5.2.1: dependencies: accepts: 2.0.0 @@ -2383,6 +2526,11 @@ snapshots: fast-levenshtein@2.0.6: {} + fast-unique-numbers@9.0.27: + dependencies: + '@babel/runtime': 7.29.7 + tslib: 2.8.1 + fdir@6.5.0(picomatch@4.0.5): optionalDependencies: picomatch: 4.0.5 @@ -2457,6 +2605,8 @@ snapshots: dependencies: function-bind: 1.1.2 + help-me@5.0.0: {} + hermes-estree@0.25.1: {} hermes-parser@0.25.1: @@ -2475,12 +2625,16 @@ snapshots: dependencies: safer-buffer: 2.1.2 + ieee754@1.2.1: {} + ignore@5.3.2: {} imurmurhash@0.1.4: {} inherits@2.0.4: {} + ip-address@10.5.0: {} + ipaddr.js@1.9.1: {} is-extglob@2.1.1: {} @@ -2493,6 +2647,8 @@ snapshots: isexe@2.0.0: {} + js-sdsl@4.3.0: {} + js-tokens@4.0.0: {} jsesc@3.1.0: {} @@ -2567,6 +2723,8 @@ snapshots: dependencies: p-locate: 5.0.0 + lru-cache@10.4.3: {} + lru-cache@5.1.1: dependencies: yallist: 3.1.1 @@ -2601,6 +2759,37 @@ snapshots: dependencies: minimist: 1.2.8 + mqtt-packet@9.0.2: + dependencies: + bl: 6.1.6 + debug: 4.4.3 + process-nextick-args: 2.0.1 + transitivePeerDependencies: + - supports-color + + mqtt@5.15.2: + dependencies: + '@types/readable-stream': 4.0.24 + '@types/ws': 8.18.1 + commist: 3.2.0 + concat-stream: 2.0.0 + debug: 4.4.3 + help-me: 5.0.0 + lru-cache: 10.4.3 + minimist: 1.2.8 + mqtt-packet: 9.0.2 + number-allocator: 1.0.14 + readable-stream: 4.7.0 + rfdc: 1.4.1 + socks: 2.8.9 + split2: 4.2.0 + worker-timers: 8.0.34 + ws: 8.21.3 + transitivePeerDependencies: + - bufferutil + - supports-color + - utf-8-validate + ms@2.1.3: {} multer@2.0.2: @@ -2625,6 +2814,13 @@ snapshots: node-releases@2.0.53: {} + number-allocator@1.0.14: + dependencies: + debug: 4.4.3 + js-sdsl: 4.3.0 + transitivePeerDependencies: + - supports-color + object-assign@4.1.1: {} object-inspect@1.13.4: {} @@ -2674,6 +2870,10 @@ snapshots: prelude-ls@1.2.1: {} + process-nextick-args@2.0.1: {} + + process@0.11.10: {} + proxy-addr@2.0.7: dependencies: forwarded: 0.2.0 @@ -2707,6 +2907,16 @@ snapshots: string_decoder: 1.3.0 util-deprecate: 1.0.2 + readable-stream@4.7.0: + dependencies: + abort-controller: 3.0.0 + buffer: 6.0.3 + events: 3.3.0 + process: 0.11.10 + string_decoder: 1.3.0 + + rfdc@1.4.1: {} + rolldown@1.2.4: dependencies: '@oxc-project/types': 0.144.0 @@ -2840,6 +3050,8 @@ snapshots: side-channel-map: 1.0.1 side-channel-weakmap: 1.0.2 + smart-buffer@4.2.0: {} + socket.io-adapter@2.5.6: dependencies: debug: 4.4.3 @@ -2881,8 +3093,15 @@ snapshots: - supports-color - utf-8-validate + socks@2.8.9: + dependencies: + ip-address: 10.5.0 + smart-buffer: 4.2.0 + source-map-js@1.2.1: {} + split2@4.2.0: {} + statuses@2.0.2: {} streamsearch@1.1.0: {} @@ -2952,6 +3171,33 @@ snapshots: word-wrap@1.2.5: {} + worker-factory@7.0.50: + dependencies: + '@babel/runtime': 7.29.7 + fast-unique-numbers: 9.0.27 + tslib: 2.8.1 + + worker-timers-broker@8.0.18: + dependencies: + '@babel/runtime': 7.29.7 + broker-factory: 3.1.15 + fast-unique-numbers: 9.0.27 + tslib: 2.8.1 + worker-timers-worker: 9.0.15 + + worker-timers-worker@9.0.15: + dependencies: + '@babel/runtime': 7.29.7 + tslib: 2.8.1 + worker-factory: 7.0.50 + + worker-timers@8.0.34: + dependencies: + '@babel/runtime': 7.29.7 + tslib: 2.8.1 + worker-timers-broker: 8.0.18 + worker-timers-worker: 9.0.15 + wrappy@1.0.2: {} ws@8.18.3: {} diff --git a/server/package.json b/server/package.json index fa72769..dff0d09 100644 --- a/server/package.json +++ b/server/package.json @@ -17,6 +17,7 @@ "better-sqlite3": "13.0.2", "dotenv": "^16.5.0", "express": "^5.2.1", + "mqtt": "5.15.2", "multer": "2.0.2", "sharp": "0.35.3", "socket.io": "^4.8.3" diff --git a/server/src/app.js b/server/src/app.js index f2b77b8..e35f0ad 100644 --- a/server/src/app.js +++ b/server/src/app.js @@ -6,6 +6,7 @@ const { const { createAssetRouter } = require("./assets/routes"); const { createMediaRouter } = require("./assets/media-routes"); const { createAssetService } = require("./assets/service"); +const { createDisplayRouter } = require("./displays/routes"); const { createPlaybackService } = require("./playback"); const { createPlaybackRouter } = require("./playback/routes"); const { @@ -36,11 +37,14 @@ function safeNotifier(notifier, logger, failureMessage) { function createApp({ config, database, + displayService, imageProcessor, + integrations = {}, lifecycle, logger = console, migrations, now = Date.now, + onAssetProcessingFailed = () => {}, onPlaylistsChanged = () => {}, playbackService: suppliedPlaybackService, storage, @@ -72,9 +76,15 @@ function createApp({ notifyPlaylistsChanged(publicChange); }; const settingsService = createSettingsService(database); + const notifyAssetProcessingFailed = safeNotifier( + onAssetProcessingFailed, + logger, + "Asset processing failure notification failed", + ); const assetService = createAssetService({ database, imageProcessor, + onProcessingFailed: notifyAssetProcessingFailed, storage, }); const playlistService = createPlaylistService({ @@ -99,8 +109,11 @@ function createApp({ const detailedHealthRouterOptions = { config, database, + displayService, + integrations, lifecycle, migrations, + playbackService, storage, }; app.use( @@ -136,6 +149,9 @@ function createApp({ }), ); app.use("/api/playback", createPlaybackRouter(playbackService)); + if (displayService) { + app.use("/api/displays", createDisplayRouter(displayService)); + } app.use("/media", createMediaRouter(assetService)); app.use((request, response) => { diff --git a/server/src/assets/service.js b/server/src/assets/service.js index 4a828c3..d32bf80 100644 --- a/server/src/assets/service.js +++ b/server/src/assets/service.js @@ -163,6 +163,7 @@ function createAssetService({ createId = crypto.randomUUID, wait = delay, concurrentUploadWaitMs = CONCURRENT_UPLOAD_WAIT_MS, + onProcessingFailed = () => {}, }) { const assetSelect = ` SELECT a.*, @@ -434,11 +435,26 @@ function createAssetService({ return { created: true, duplicate: false, asset: mapAsset(row) }; } catch (error) { const failure = processingMetadata(id, plan.processed, keys, error); + let recorded = false; try { markFailed.run(failure); + recorded = true; } catch { // Preserve the original failure; the processing row remains diagnosable. } + if (recorded) { + try { + // One observable transition per newly failed asset. No polling scan + // is needed to notice historical failures. + onProcessingFailed({ + assetId: id, + displayName: plan.displayName || plan.file?.originalFilename || null, + message: failure.processingError, + }); + } catch { + // A notification hook must never mask the original upload error. + } + } try { await cleanObjects(keys); } catch { diff --git a/server/src/config/index.js b/server/src/config/index.js index c75357b..4a6244f 100644 --- a/server/src/config/index.js +++ b/server/src/config/index.js @@ -82,6 +82,23 @@ function url(env, name, { required = false, protocols } = {}) { return parsed.toString().replace(/\/$/, ""); } +function topicPrefix(env) { + const value = env.MQTT_TOPIC_PREFIX?.trim() || "announcements"; + if (/[+#]/.test(value)) { + throw new ConfigError( + "MQTT_TOPIC_PREFIX cannot contain the + or # wildcards", + "MQTT_TOPIC_PREFIX", + ); + } + if (value.startsWith("/") || value.endsWith("/")) { + throw new ConfigError( + "MQTT_TOPIC_PREFIX cannot start or end with /", + "MQTT_TOPIC_PREFIX", + ); + } + return value; +} + function bucketName(env) { const value = requiredString(env, "S3_BUCKET"); if (!/^[a-z0-9][a-z0-9.-]{1,61}[a-z0-9]$/.test(value)) { @@ -199,12 +216,31 @@ function loadConfig(env = process.env) { }), username: optionalString(env, "MQTT_USERNAME"), password: optionalString(env, "MQTT_PASSWORD"), + topicPrefix: topicPrefix(env), + reconnectPeriodMs: integer(env, "MQTT_RECONNECT_PERIOD_MS", 5_000, { + min: 250, + max: 300_000, + }), + connectTimeoutMs: integer(env, "MQTT_CONNECT_TIMEOUT_MS", 10_000, { + min: 250, + max: 120_000, + }), }), notifications: Object.freeze({ homeAssistantWebhookUrl: url(env, "HA_WEBHOOK_URL", { protocols: ["http:", "https:"], }), homeAssistantWebhookSecret: optionalString(env, "HA_WEBHOOK_SECRET"), + webhookTimeoutMs: integer(env, "HA_WEBHOOK_TIMEOUT_MS", 10_000, { + min: 250, + max: 120_000, + }), + }), + displays: Object.freeze({ + offlineThresholdMs: integer(env, "DISPLAY_OFFLINE_THRESHOLD_MS", 45_000, { + min: 5_000, + max: 3_600_000, + }), }), }); } @@ -240,11 +276,19 @@ function redactedConfig(config) { maxFiles: config.processing.maxFiles, maxInputPixels: config.processing.maxInputPixels, }, + // Status output must survive a partial or unexpected configuration shape: + // this mapping feeds the endpoints an operator uses when things are wrong. mqtt: { - configured: Boolean(config.mqtt.url), + configured: Boolean(config.mqtt?.url), + topicPrefix: config.mqtt?.topicPrefix ?? null, + reconnectPeriodMs: config.mqtt?.reconnectPeriodMs ?? null, }, notifications: { - configured: Boolean(config.notifications.homeAssistantWebhookUrl), + configured: Boolean(config.notifications?.homeAssistantWebhookUrl), + webhookTimeoutMs: config.notifications?.webhookTimeoutMs ?? null, + }, + displays: { + offlineThresholdMs: config.displays?.offlineThresholdMs ?? null, }, }; } diff --git a/server/src/displays/routes.js b/server/src/displays/routes.js new file mode 100644 index 0000000..6247d9f --- /dev/null +++ b/server/src/displays/routes.js @@ -0,0 +1,19 @@ +const express = require("express"); + +function createDisplayRouter(displayService) { + const router = express.Router(); + + router.get("/", (request, response, next) => { + try { + response.json({ displays: displayService.list() }); + } catch (error) { + next(error); + } + }); + + return router; +} + +module.exports = { + createDisplayRouter, +}; diff --git a/server/src/displays/service.js b/server/src/displays/service.js new file mode 100644 index 0000000..70310f7 --- /dev/null +++ b/server/src/displays/service.js @@ -0,0 +1,327 @@ +const HEARTBEAT_PROTOCOL_VERSION = 1; +const MAX_ID_LENGTH = 200; +const MAX_ERROR_LENGTH = 500; +const MAX_COUNT = 1_000_000; +const DEFAULT_OFFLINE_THRESHOLD_MS = 45_000; + +function optionalId(value, field) { + if (value === undefined || value === null) return null; + if (typeof value !== "string") throw new TypeError(`${field} must be a string`); + const trimmed = value.trim(); + if (!trimmed) return null; + if (trimmed.length > MAX_ID_LENGTH) { + throw new TypeError(`${field} is too long`); + } + return trimmed; +} + +function requiredId(value, field) { + const id = optionalId(value, field); + if (!id) throw new TypeError(`${field} is required`); + return id; +} + +function optionalCount(value, field) { + if (value === undefined || value === null) return null; + if (!Number.isSafeInteger(value) || value < 0 || value > MAX_COUNT) { + throw new TypeError(`${field} must be an integer from 0 through ${MAX_COUNT}`); + } + return value; +} + +function optionalRevision(value, field) { + if (value === undefined || value === null) return null; + if (!Number.isSafeInteger(value) || value < 0) { + throw new TypeError(`${field} must be a non-negative integer`); + } + return value; +} + +// Telemetry is attacker-adjacent input from a kiosk browser. It is bounded +// here so a looping client cannot grow the database without limit. +function boundedError(value) { + if (value === undefined || value === null) return null; + if (typeof value !== "string") throw new TypeError("mostRecentError must be a string"); + const trimmed = value.trim(); + if (!trimmed) return null; + return trimmed.length > MAX_ERROR_LENGTH + ? `${trimmed.slice(0, MAX_ERROR_LENGTH - 1)}…` + : trimmed; +} + +function validateHeartbeat(payload) { + if (payload === null || typeof payload !== "object" || Array.isArray(payload)) { + throw new TypeError("heartbeat must be an object"); + } + if (payload.protocolVersion !== HEARTBEAT_PROTOCOL_VERSION) { + throw new TypeError( + `protocolVersion must be ${HEARTBEAT_PROTOCOL_VERSION}`, + ); + } + return { + clientId: requiredId(payload.clientId, "clientId"), + outputId: optionalId(payload.outputId, "outputId") || "global", + activePlaylistId: optionalId(payload.activePlaylistId, "activePlaylistId"), + playlistRevision: optionalRevision(payload.playlistRevision, "playlistRevision"), + playbackRevision: optionalRevision(payload.playbackRevision, "playbackRevision"), + currentEntryId: optionalId(payload.currentEntryId, "currentEntryId"), + currentAssetId: optionalId(payload.currentAssetId, "currentAssetId"), + cachedAssetCount: optionalCount(payload.cachedAssetCount, "cachedAssetCount"), + playableAssetCount: optionalCount(payload.playableAssetCount, "playableAssetCount"), + lastError: boundedError(payload.mostRecentError), + }; +} + +function createDisplayService({ + database, + notifications = null, + now = Date.now, + setTimeoutFn = setTimeout, + clearTimeoutFn = clearTimeout, + offlineThresholdMs = DEFAULT_OFFLINE_THRESHOLD_MS, + onStatusChanged = () => {}, + logger = { warn() {} }, +} = {}) { + if (!database) throw new TypeError("database is required"); + + const findClient = database.prepare( + "SELECT * FROM display_clients WHERE client_id = ?", + ); + const listClients = database.prepare( + "SELECT * FROM display_clients ORDER BY first_seen_at, client_id", + ); + const insertClient = database.prepare(` + INSERT INTO display_clients ( + client_id, output_id, first_seen_at, last_seen_at, active_playlist_id, + playlist_revision, playback_revision, current_entry_id, current_asset_id, + cached_asset_count, playable_asset_count, last_error, updated_at + ) VALUES ( + @clientId, @outputId, @at, @at, @activePlaylistId, @playlistRevision, + @playbackRevision, @currentEntryId, @currentAssetId, @cachedAssetCount, + @playableAssetCount, @lastError, @at + ) + `); + const updateClient = database.prepare(` + UPDATE display_clients + SET output_id = @outputId, + last_seen_at = @at, + active_playlist_id = @activePlaylistId, + playlist_revision = @playlistRevision, + playback_revision = @playbackRevision, + current_entry_id = @currentEntryId, + current_asset_id = @currentAssetId, + cached_asset_count = @cachedAssetCount, + playable_asset_count = @playableAssetCount, + last_error = @lastError, + updated_at = @at + WHERE client_id = @clientId + `); + + let running = false; + let timerHandle = null; + // Restarting the API is not evidence that a display disappeared. Nothing is + // reported offline until a full threshold has elapsed since startup, so a + // deploy cannot manufacture offline notifications for healthy displays. + let graceUntil = now() + offlineThresholdMs; + let heartbeatCount = 0; + + const conditionKey = (clientId) => `display:${clientId}`; + + function warn(error, message) { + try { + logger.warn?.({ error }, message); + } catch { + // Telemetry logging must never break heartbeat ingestion. + } + } + + function notify(event) { + try { + onStatusChanged(event); + } catch (error) { + warn(error, "Display status callback failed"); + } + } + + function isOffline(row, at) { + return at - row.last_seen_at >= offlineThresholdMs; + } + + function publicClient(row, at) { + return { + clientId: row.client_id, + outputId: row.output_id, + status: isOffline(row, at) ? "offline" : "online", + firstSeenAt: new Date(row.first_seen_at).toISOString(), + lastSeenAt: new Date(row.last_seen_at).toISOString(), + activePlaylistId: row.active_playlist_id, + playlistRevision: row.playlist_revision, + playbackRevision: row.playback_revision, + currentEntryId: row.current_entry_id, + currentAssetId: row.current_asset_id, + cachedAssetCount: row.cached_asset_count, + playableAssetCount: row.playable_asset_count, + mostRecentError: row.last_error, + }; + } + + function openOffline(row, at) { + if (!notifications) return false; + const result = notifications.observeFailure(conditionKey(row.client_id), { + event: "display.offline", + severity: "critical", + message: `Display ${row.client_id} stopped sending heartbeats.`, + context: { + clientId: row.client_id, + outputId: row.output_id, + lastSeenAt: new Date(row.last_seen_at).toISOString(), + activePlaylistId: row.active_playlist_id, + currentEntryId: row.current_entry_id, + currentAssetId: row.current_asset_id, + }, + }); + if (result.opened) { + notify({ + reason: "offline", + ...publicClient(row, at), + status: "offline", + }); + } + return result.opened; + } + + function recoverOffline(row, at) { + if (!notifications) return false; + const result = notifications.observeRecovery(conditionKey(row.client_id), { + event: "display.recovered", + message: `Display ${row.client_id} is sending heartbeats again.`, + context: { + clientId: row.client_id, + outputId: row.output_id, + recoveredAt: new Date(at).toISOString(), + }, + }); + return result.recovered; + } + + function heartbeat(payload) { + const telemetry = validateHeartbeat(payload); + const at = now(); + const existing = findClient.get(telemetry.clientId); + const record = { ...telemetry, at }; + + if (existing) { + updateClient.run(record); + } else { + insertClient.run(record); + } + heartbeatCount += 1; + + const row = findClient.get(telemetry.clientId); + const wasOffline = existing ? isOffline(existing, at) : false; + const recovered = recoverOffline(row, at); + const errorChanged = (existing?.last_error ?? null) !== telemetry.lastError; + + if (!existing) { + notify({ reason: "first_seen", ...publicClient(row, at) }); + } else if (recovered || wasOffline) { + notify({ reason: "recovered", ...publicClient(row, at) }); + } else if (errorChanged) { + notify({ reason: "error_changed", ...publicClient(row, at) }); + } + // A steady heartbeat is intentionally silent: telemetry must not create + // realtime churn, and it never touches playback or playlist revisions. + return publicClient(row, at); + } + + function sweep() { + const at = now(); + if (at < graceUntil) return 0; + let opened = 0; + for (const row of listClients.all()) { + if (!isOffline(row, at)) continue; + if (openOffline(row, at)) opened += 1; + } + return opened; + } + + function schedule() { + if (!running) return; + if (timerHandle !== null) { + clearTimeoutFn(timerHandle); + timerHandle = null; + } + const at = now(); + const rows = listClients.all(); + // Wake only when the next display could actually cross the threshold + // instead of polling on a fixed interval. + const deadlines = rows + .map((row) => row.last_seen_at + offlineThresholdMs) + .filter((deadline) => deadline > at); + const next = Math.max( + graceUntil, + deadlines.length > 0 ? Math.min(...deadlines) : at + offlineThresholdMs, + ); + timerHandle = setTimeoutFn(() => { + timerHandle = null; + try { + sweep(); + } catch (error) { + warn(error, "Display offline sweep failed"); + } + schedule(); + }, Math.max(0, next - at)); + timerHandle?.unref?.(); + } + + function start() { + running = true; + graceUntil = now() + offlineThresholdMs; + schedule(); + } + + function stop() { + running = false; + if (timerHandle !== null) { + clearTimeoutFn(timerHandle); + timerHandle = null; + } + } + + function list() { + const at = now(); + return listClients.all().map((row) => publicClient(row, at)); + } + + function status() { + const at = now(); + const rows = listClients.all(); + const online = rows.filter((row) => !isOffline(row, at)).length; + return { + status: rows.length === 0 ? "none_seen" : online > 0 ? "ok" : "degraded", + known: rows.length, + online, + offline: rows.length - online, + heartbeatCount, + offlineThresholdMs, + }; + } + + return { + heartbeat, + list, + schedule, + start, + status, + stop, + sweep, + }; +} + +module.exports = { + DEFAULT_OFFLINE_THRESHOLD_MS, + HEARTBEAT_PROTOCOL_VERSION, + MAX_ERROR_LENGTH, + createDisplayService, + validateHeartbeat, +}; diff --git a/server/src/health/routes.js b/server/src/health/routes.js index 1663fdc..7ac1ff4 100644 --- a/server/src/health/routes.js +++ b/server/src/health/routes.js @@ -87,7 +87,46 @@ function createHealthRouter({ database, lifecycle, migrations, storage }) { return router; } -function createDetailedHealthRouter({ config, database, lifecycle, migrations, storage }) { +// Integration status is reported defensively: an integration that is down or +// throwing must still produce a status document rather than a 500. +function integrationStatus(service, fallbackConfigured) { + if (!service) { + return { status: "not_started", configured: fallbackConfigured }; + } + try { + return service.status(); + } catch { + return { status: "unknown", configured: fallbackConfigured }; + } +} + +function playbackStatus(playbackService) { + if (!playbackService) return { status: "not_started" }; + try { + const snapshot = playbackService.getSnapshot(); + return { + status: "ok", + mode: snapshot.mode, + transportState: snapshot.transportState, + activePlaylistId: snapshot.activePlaylistId ?? null, + playbackRevision: snapshot.playbackRevision, + hasPlayableTarget: Boolean(snapshot.normalTarget), + }; + } catch { + return { status: "unknown" }; + } +} + +function createDetailedHealthRouter({ + config, + database, + displayService, + integrations = {}, + lifecycle, + migrations, + playbackService, + storage, +}) { const router = express.Router(); router.get("/", async (request, response) => { @@ -99,11 +138,22 @@ function createDetailedHealthRouter({ config, database, lifecycle, migrations, s migrations.pendingVersions.length === 0; const configuration = redactedConfig(config); - const status = !ready - ? "not_ready" - : storageCheck.status === "healthy" - ? "ok" - : "degraded"; + const mqttCheck = integrationStatus( + integrations.mqtt, + configuration.mqtt.configured, + ); + const notificationCheck = integrationStatus( + integrations.notifications, + configuration.notifications.configured, + ); + const displayCheck = integrationStatus(displayService, true); + // External integrations degrade the reported status. They never make the + // API unready, because signage playback does not depend on them. + const degraded = + storageCheck.status !== "healthy" || + mqttCheck.status === "disconnected" || + notificationCheck.status === "degraded"; + const status = !ready ? "not_ready" : degraded ? "degraded" : "ok"; response.status(ready ? 200 : 503).json({ status, @@ -118,16 +168,10 @@ function createDetailedHealthRouter({ config, database, lifecycle, migrations, s ...storageCheck, configured: configuration.storage.configured, }, - mqtt: { - status: "not_started", - configured: configuration.mqtt.configured, - }, - notifications: { - status: "not_started", - configured: configuration.notifications.configured, - }, - playback: { status: "not_started" }, - displays: { status: "not_started" }, + mqtt: mqttCheck, + notifications: notificationCheck, + playback: playbackStatus(playbackService), + displays: displayCheck, }, }); }); diff --git a/server/src/mqtt/index.js b/server/src/mqtt/index.js new file mode 100644 index 0000000..e7a04fe --- /dev/null +++ b/server/src/mqtt/index.js @@ -0,0 +1,314 @@ +const PLAYBACK_STATE_VERSION = 1; + +// Commands that change durable state are delivered at least once. next and +// previous are not idempotent and carry no command ID, so a redelivered copy +// would silently skip a slide; they stay at QoS 0 deliberately. +const COMMANDS = Object.freeze([ + { name: "mode", suffix: "mode", qos: 1 }, + { name: "pause", suffix: "pause", qos: 1 }, + { name: "resume", suffix: "resume", qos: 1 }, + { name: "reload", suffix: "reload", qos: 1 }, + { name: "static", suffix: "static", qos: 1 }, + { name: "next", suffix: "next", qos: 0 }, + { name: "previous", suffix: "previous", qos: 0 }, +]); + +const UUID_PATTERN = + /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i; + +function transportOf(snapshot) { + if (snapshot.staticTarget) return "static"; + if (snapshot.transport?.rotationPaused) return "paused"; + return snapshot.normalTarget ? "rotating" : "idle"; +} + +function compactState(snapshot) { + return { + version: PLAYBACK_STATE_VERSION, + mode: snapshot.mode, + transport: transportOf(snapshot), + activePlaylistId: snapshot.activePlaylistId ?? null, + currentEntryId: snapshot.currentEntryId ?? null, + currentAssetId: snapshot.normalTarget?.assetId ?? null, + staticAssetId: snapshot.staticTarget?.assetId ?? null, + revision: snapshot.playbackRevision, + }; +} + +function createMqttService({ + config, + playbackService, + connect, + logger = console, +} = {}) { + if (!playbackService) throw new TypeError("playbackService is required"); + const settings = config?.mqtt || {}; + const prefix = settings.topicPrefix || "announcements"; + const topic = (suffix) => `${prefix}/${suffix}`; + const availabilityTopic = topic("availability"); + const modeStateTopic = topic("mode/state"); + const playbackStateTopic = topic("playback/state"); + + const configured = Boolean(settings.url); + let client = null; + let connected = false; + let connectionCount = 0; + let lastConnectedAt = null; + let lastErrorAt = null; + let lastErrorMessage = null; + let commandCount = 0; + let stopping = false; + + function warn(error, message, details) { + try { + logger.warn?.(`${message}${details ? ` ${JSON.stringify(details)}` : ""}`, error); + } catch { + // Logging failures must never propagate into the broker callbacks. + } + } + + function publish(topicName, payload, options) { + if (!client || !connected) return false; + try { + client.publish(topicName, payload, options, (error) => { + if (error) warn(error, "MQTT publish failed", { topic: topicName }); + }); + return true; + } catch (error) { + warn(error, "MQTT publish threw", { topic: topicName }); + return false; + } + } + + // Retained authoritative state is only ever published on dedicated state + // topics so a subscriber can never mistake an echo for a new command. + function publishState(snapshot) { + if (!snapshot) return; + publish(modeStateTopic, snapshot.mode, { qos: 1, retain: true }); + publish(playbackStateTopic, JSON.stringify(compactState(snapshot)), { + qos: 1, + retain: true, + }); + } + + function publishCurrentState() { + try { + publishState(playbackService.getSnapshot()); + } catch (error) { + warn(error, "MQTT could not read the authoritative playback snapshot"); + } + } + + function handleMode(payload) { + const value = payload.trim().toLowerCase(); + if (value !== "on" && value !== "off") { + warn(null, "Ignoring MQTT mode command with an unsupported payload", { + payload: value.slice(0, 40), + }); + return false; + } + playbackService.setMode(value); + return true; + } + + function handleStatic(payload) { + const assetId = payload.trim(); + if (!UUID_PATTERN.test(assetId)) { + warn(null, "Ignoring MQTT static command that is not a LibraryAsset UUID", { + payload: assetId.slice(0, 60), + }); + return false; + } + playbackService.enterStatic(assetId); + return true; + } + + const handlers = Object.freeze({ + mode: handleMode, + static: handleStatic, + next: () => (playbackService.next(), true), + previous: () => (playbackService.previous(), true), + pause: () => (playbackService.pause(), true), + resume: () => (playbackService.resume(), true), + reload: () => (playbackService.reload(), true), + }); + + const byTopic = new Map( + COMMANDS.map((command) => [topic(command.suffix), command]), + ); + + function handleMessage(topicName, buffer) { + const command = byTopic.get(topicName); + if (!command) { + warn(null, "Ignoring MQTT message on an unknown topic", { topic: topicName }); + return; + } + let payload = ""; + try { + payload = buffer ? buffer.toString("utf8") : ""; + } catch (error) { + warn(error, "Ignoring MQTT message with an undecodable payload", { + topic: topicName, + }); + return; + } + try { + // Every command runs through the one authoritative playback service. + // A rejected command leaves playback exactly as it was. + if (handlers[command.name](payload)) commandCount += 1; + } catch (error) { + warn(error, "MQTT command was rejected by the playback service", { + topic: topicName, + code: error?.code, + }); + } + } + + function subscribeAll() { + if (!client) return; + for (const command of COMMANDS) { + client.subscribe(topic(command.suffix), { qos: command.qos }, (error) => { + if (error) warn(error, "MQTT subscribe failed", { topic: topic(command.suffix) }); + }); + } + } + + function handleConnect() { + connected = true; + connectionCount += 1; + lastConnectedAt = new Date().toISOString(); + logger.info?.(`MQTT connected to the configured broker as ${prefix}`); + // Reconnect must restore both the subscriptions and the retained state a + // broker restart would otherwise have dropped. + publish(availabilityTopic, "online", { qos: 1, retain: true }); + subscribeAll(); + publishCurrentState(); + } + + function start() { + if (!configured) { + logger.info?.("MQTT is not configured; the integration stays disabled"); + return null; + } + if (client) return client; + const factory = connect || require("mqtt").connect; + client = factory(settings.url, { + username: settings.username || undefined, + password: settings.password || undefined, + reconnectPeriod: settings.reconnectPeriodMs ?? 5_000, + connectTimeout: settings.connectTimeoutMs ?? 10_000, + clean: true, + // The broker publishes this for us if the API disappears without a + // clean disconnect, which is exactly the case a client cannot report. + will: { + topic: availabilityTopic, + payload: "offline", + qos: 1, + retain: true, + }, + }); + + client.on("connect", handleConnect); + client.on("message", handleMessage); + client.on("reconnect", () => { + connected = false; + }); + client.on("close", () => { + connected = false; + }); + client.on("offline", () => { + connected = false; + }); + client.on("error", (error) => { + connected = false; + lastErrorAt = new Date().toISOString(); + lastErrorMessage = error?.message || "MQTT error"; + // A broker outage degrades the integration only. Playback, HTTP, and + // the display protocol continue untouched and the client reconnects. + warn(error, "MQTT connection error; the integration is degraded"); + }); + return client; + } + + async function stop() { + stopping = true; + if (!client) return; + const active = client; + client = null; + try { + if (connected) { + await new Promise((resolve) => { + let settled = false; + const finish = () => { + if (settled) return; + settled = true; + resolve(); + }; + const timer = setTimeout(finish, 1_000); + timer.unref?.(); + // A clean shutdown retracts availability itself; the will is only + // for the unexpected case. + active.publish(availabilityTopic, "offline", { qos: 1, retain: true }, () => { + clearTimeout(timer); + finish(); + }); + }); + for (const command of COMMANDS) { + try { + active.unsubscribe(topic(command.suffix)); + } catch { + // Shutdown continues even if the broker already went away. + } + } + } + } catch (error) { + warn(error, "MQTT shutdown publication failed"); + } + connected = false; + try { + await new Promise((resolve) => { + const timer = setTimeout(resolve, 1_000); + timer.unref?.(); + active.end(false, {}, () => { + clearTimeout(timer); + resolve(); + }); + }); + } catch (error) { + warn(error, "MQTT client did not close cleanly"); + } + } + + function status() { + return { + configured, + status: !configured ? "disabled" : connected ? "connected" : "disconnected", + connected, + topicPrefix: prefix, + connectionCount, + commandCount, + lastConnectedAt, + lastErrorAt, + lastErrorMessage, + stopping, + }; + } + + return { + availabilityTopic, + commandTopics: COMMANDS.map((command) => topic(command.suffix)), + compactState, + modeStateTopic, + playbackStateTopic, + publishState, + start, + status, + stop, + }; +} + +module.exports = { + COMMANDS, + PLAYBACK_STATE_VERSION, + createMqttService, +}; diff --git a/server/src/notifications/conditions.js b/server/src/notifications/conditions.js new file mode 100644 index 0000000..03a884d --- /dev/null +++ b/server/src/notifications/conditions.js @@ -0,0 +1,88 @@ +const STORAGE_CONDITION = "storage"; +const NO_PLAYABLE_CONDITION = "playback.no_playable"; + +// Storage health already debounces individual request failures into one +// coherent status, so an outage observed by many requests stays one incident. +function observeStorageHealth(notifications, health) { + if (!notifications || !health) return null; + if (health.status === "healthy") { + return notifications.observeRecovery(STORAGE_CONDITION, { + event: "storage.recovered", + message: "Announcement storage is reachable again.", + context: { checkedAt: health.checkedAt || null }, + }); + } + if (health.status === "degraded" && health.code === "STORAGE_UNAVAILABLE") { + return notifications.observeFailure(STORAGE_CONDITION, { + event: "storage.unavailable", + severity: "critical", + message: "Announcement storage is unavailable.", + context: { + checkedAt: health.checkedAt || null, + lastSuccessAt: health.lastSuccessAt || null, + }, + }); + } + // STORAGE_NOT_CHECKED is startup state, not an outage worth reporting. + return null; +} + +// A static override can keep the screen filled while the normal playlist is +// empty. That is still an operator-actionable condition, so static does not +// suppress it. +function observePlayback(notifications, snapshot) { + if (!notifications || !snapshot) return null; + if (snapshot.activePlaylistId === null || snapshot.activePlaylistId === undefined) { + // No playlist has been activated yet; there is nothing to repair. + return notifications.observeRecovery(NO_PLAYABLE_CONDITION, { + event: "playback.playable", + message: "Announcement playback has a playable slide again.", + context: { activePlaylistId: null }, + }); + } + if (snapshot.normalTarget) { + return notifications.observeRecovery(NO_PLAYABLE_CONDITION, { + event: "playback.playable", + message: "Announcement playback has a playable slide again.", + context: { + activePlaylistId: snapshot.activePlaylistId, + currentEntryId: snapshot.currentEntryId ?? null, + }, + }); + } + return notifications.observeFailure(NO_PLAYABLE_CONDITION, { + event: "playback.no_playable_slides", + severity: "warning", + message: "The active announcement playlist has no playable slides.", + context: { + activePlaylistId: snapshot.activePlaylistId, + staticActive: Boolean(snapshot.staticTarget), + mode: snapshot.mode, + }, + }); +} + +// An individual failed upload has no automatic recovery: the operator deletes +// it or uploads replacement bytes. One incident per asset means one warning, +// never a repeating one. +function observeAssetProcessingFailure(notifications, { assetId, displayName, message }) { + if (!notifications || !assetId) return null; + return notifications.observeFailure(`asset:${assetId}`, { + event: "asset.processing_failed", + severity: "warning", + message: `Image processing failed for ${displayName || assetId}.`, + context: { + assetId, + displayName: displayName || null, + processingError: notifications.boundedError(message), + }, + }); +} + +module.exports = { + NO_PLAYABLE_CONDITION, + STORAGE_CONDITION, + observeAssetProcessingFailure, + observePlayback, + observeStorageHealth, +}; diff --git a/server/src/notifications/index.js b/server/src/notifications/index.js new file mode 100644 index 0000000..6331a61 --- /dev/null +++ b/server/src/notifications/index.js @@ -0,0 +1,357 @@ +const crypto = require("node:crypto"); + +const PAYLOAD_VERSION = 1; +const MAX_ERROR_LENGTH = 500; + +// Immediate, then a capped schedule. Index is the attempt that just failed, so +// the first retry waits 30s and every attempt past the table waits 30 minutes. +const RETRY_BACKOFF_MS = Object.freeze([ + 30_000, + 60_000, + 300_000, + 900_000, + 1_800_000, +]); + +const SEVERITIES = Object.freeze(["info", "warning", "critical", "recovery"]); + +function boundedError(value) { + if (typeof value !== "string") return null; + const trimmed = value.trim(); + if (!trimmed) return null; + return trimmed.length > MAX_ERROR_LENGTH + ? `${trimmed.slice(0, MAX_ERROR_LENGTH - 1)}…` + : trimmed; +} + +function backoffMs(attemptCount) { + if (attemptCount <= 0) return 0; + return RETRY_BACKOFF_MS[ + Math.min(attemptCount - 1, RETRY_BACKOFF_MS.length - 1) + ]; +} + +function createNotificationService({ + database, + webhookUrl = null, + webhookSecret = null, + webhookTimeoutMs = 10_000, + now = Date.now, + setTimeoutFn = setTimeout, + clearTimeoutFn = clearTimeout, + createId = crypto.randomUUID, + fetchFn = (...args) => fetch(...args), + logger = { warn() {} }, +} = {}) { + if (!database) throw new TypeError("database is required"); + + const findOpen = database.prepare( + "SELECT * FROM notification_incidents WHERE condition_key = ? AND state = 'open'", + ); + const lastEpisode = database.prepare( + "SELECT COALESCE(MAX(episode), 0) AS episode FROM notification_incidents WHERE condition_key = ?", + ); + const insertIncident = database.prepare(` + INSERT INTO notification_incidents ( + id, condition_key, episode, event, severity, state, message, + context_json, opened_at, last_observed_at + ) VALUES ( + @id, @conditionKey, @episode, @event, @severity, 'open', @message, + @contextJson, @openedAt, @lastObservedAt + ) + `); + const observeIncident = database.prepare( + "UPDATE notification_incidents SET last_observed_at = ? WHERE id = ?", + ); + const recoverIncident = database.prepare(` + UPDATE notification_incidents + SET state = 'recovered', recovered_at = ?, last_observed_at = ? + WHERE id = ? AND state = 'open' + `); + const insertOutbox = database.prepare(` + INSERT INTO notification_outbox ( + id, incident_id, transition, payload_json, attempt_count, + next_attempt_at, created_at + ) VALUES ( + @id, @incidentId, @transition, @payloadJson, 0, @nextAttemptAt, @createdAt + ) + `); + const dueDeliveries = database.prepare(` + SELECT * FROM notification_outbox + WHERE delivered_at IS NULL AND next_attempt_at <= ? + ORDER BY next_attempt_at, created_at + LIMIT 20 + `); + const earliestPending = database.prepare(` + SELECT MIN(next_attempt_at) AS next_attempt_at + FROM notification_outbox + WHERE delivered_at IS NULL + `); + const markDelivered = database.prepare(` + UPDATE notification_outbox + SET delivered_at = ?, attempt_count = attempt_count + 1, last_error = NULL + WHERE id = ? + `); + const markFailedDelivery = database.prepare(` + UPDATE notification_outbox + SET attempt_count = attempt_count + 1, next_attempt_at = ?, last_error = ? + WHERE id = ? + `); + const countPending = database.prepare( + "SELECT COUNT(*) AS count FROM notification_outbox WHERE delivered_at IS NULL", + ); + const listOpenIncidents = database.prepare( + "SELECT * FROM notification_incidents WHERE state = 'open' ORDER BY opened_at", + ); + + const configured = Boolean(webhookUrl); + let running = false; + let timerHandle = null; + let delivering = false; + let lastSuccessAt = null; + let lastFailureAt = null; + let lastFailureMessage = null; + let deliveredCount = 0; + + function warn(error, message, details) { + try { + logger.warn?.({ error, ...details }, message); + } catch { + // Observability must never break an integration path. + } + } + + function payloadFor(incident, transition, at) { + return { + version: PAYLOAD_VERSION, + notificationId: null, + incidentId: incident.id, + event: incident.event, + severity: transition === "recovery" ? "recovery" : incident.severity, + state: transition === "recovery" ? "recovery" : "failure", + message: incident.message, + timestamp: new Date(at).toISOString(), + context: { + conditionKey: incident.condition_key, + episode: incident.episode, + ...JSON.parse(incident.context_json || "{}"), + }, + }; + } + + // Delivery work is only queued when a receiver is configured. Without a URL + // the incident history stays complete and useful while nothing retries + // forever against an endpoint that does not exist. + function enqueue(incident, transition, at) { + if (!configured) return null; + const id = createId(); + const payload = { ...payloadFor(incident, transition, at), notificationId: id }; + try { + insertOutbox.run({ + id, + incidentId: incident.id, + transition, + payloadJson: JSON.stringify(payload), + nextAttemptAt: at, + createdAt: at, + }); + } catch (error) { + // UNIQUE (incident_id, transition) makes duplicate notifications + // impossible even if a condition is observed repeatedly. + warn(error, "Notification was already queued for this transition", { + incidentId: incident.id, + transition, + }); + return null; + } + return id; + } + + const openTransaction = database.transaction( + (conditionKey, { event, severity, message, context }, at) => { + const existing = findOpen.get(conditionKey); + if (existing) { + observeIncident.run(at, existing.id); + return { incident: existing, opened: false, notificationId: null }; + } + const incident = { + id: createId(), + conditionKey, + episode: lastEpisode.get(conditionKey).episode + 1, + event, + severity, + message, + contextJson: JSON.stringify(context || {}), + openedAt: at, + lastObservedAt: at, + }; + insertIncident.run(incident); + const row = findOpen.get(conditionKey); + const notificationId = enqueue(row, "failure", at); + return { incident: row, opened: true, notificationId }; + }, + ); + + const recoverTransaction = database.transaction( + (conditionKey, { event, message, context }, at) => { + const existing = findOpen.get(conditionKey); + if (!existing) return { incident: null, recovered: false, notificationId: null }; + recoverIncident.run(at, at, existing.id); + const recoveredIncident = { + ...existing, + event: event || existing.event, + message: message || existing.message, + context_json: JSON.stringify({ + ...JSON.parse(existing.context_json || "{}"), + ...(context || {}), + }), + }; + const notificationId = enqueue(recoveredIncident, "recovery", at); + return { incident: recoveredIncident, recovered: true, notificationId }; + }, + ); + + function observeFailure(conditionKey, details) { + if (!SEVERITIES.includes(details?.severity)) { + throw new TypeError(`severity must be one of ${SEVERITIES.join(", ")}`); + } + const at = now(); + const result = openTransaction(conditionKey, details, at); + if (result.opened) scheduleDelivery(); + return result; + } + + function observeRecovery(conditionKey, details = {}) { + const at = now(); + const result = recoverTransaction(conditionKey, details, at); + if (result.recovered) scheduleDelivery(); + return result; + } + + function isOpen(conditionKey) { + return Boolean(findOpen.get(conditionKey)); + } + + async function deliverOne(row) { + const at = now(); + const headers = { "content-type": "application/json" }; + if (webhookSecret) headers["x-announcements-signature"] = webhookSecret; + try { + const response = await fetchFn(webhookUrl, { + method: "POST", + headers, + body: row.payload_json, + signal: AbortSignal.timeout(webhookTimeoutMs), + }); + if (response.status >= 200 && response.status < 300) { + markDelivered.run(at, row.id); + deliveredCount += 1; + lastSuccessAt = at; + return true; + } + throw new Error(`Webhook responded with HTTP ${response.status}`); + } catch (error) { + // Every non-2xx and transport failure is retried on the same capped + // schedule; the payload and its notification ID never change. + const attempt = row.attempt_count + 1; + lastFailureAt = at; + lastFailureMessage = boundedError(error?.message) || "Webhook delivery failed"; + markFailedDelivery.run(at + backoffMs(attempt), lastFailureMessage, row.id); + warn(error, "Home Assistant webhook delivery failed", { + notificationId: row.id, + attempt, + }); + return false; + } + } + + async function deliverDue() { + if (!configured || delivering) return 0; + delivering = true; + let processed = 0; + try { + const rows = dueDeliveries.all(now()); + for (const row of rows) { + await deliverOne(row); + processed += 1; + } + } finally { + delivering = false; + } + scheduleDelivery(); + return processed; + } + + function scheduleDelivery() { + if (!configured || !running) return; + if (timerHandle !== null) { + clearTimeoutFn(timerHandle); + timerHandle = null; + } + const next = earliestPending.get()?.next_attempt_at; + if (next === null || next === undefined) return; + const delay = Math.max(0, next - now()); + timerHandle = setTimeoutFn(() => { + timerHandle = null; + deliverDue().catch((error) => { + warn(error, "Notification delivery pass failed"); + }); + }, delay); + timerHandle?.unref?.(); + } + + function start() { + running = true; + scheduleDelivery(); + } + + function stop() { + running = false; + if (timerHandle !== null) { + clearTimeoutFn(timerHandle); + timerHandle = null; + } + } + + function status() { + const pending = countPending.get().count; + return { + // The webhook URL and secret are configuration, never status output. + configured, + status: !configured ? "disabled" : lastFailureAt && pending > 0 ? "degraded" : "ok", + pendingCount: pending, + deliveredCount, + lastSuccessAt: lastSuccessAt === null ? null : new Date(lastSuccessAt).toISOString(), + lastFailureAt: lastFailureAt === null ? null : new Date(lastFailureAt).toISOString(), + lastFailureMessage, + openIncidents: listOpenIncidents.all().map((incident) => ({ + id: incident.id, + conditionKey: incident.condition_key, + event: incident.event, + severity: incident.severity, + openedAt: new Date(incident.opened_at).toISOString(), + lastObservedAt: new Date(incident.last_observed_at).toISOString(), + })), + }; + } + + return { + boundedError, + configured, + deliverDue, + isOpen, + observeFailure, + observeRecovery, + start, + status, + stop, + }; +} + +module.exports = { + MAX_ERROR_LENGTH, + PAYLOAD_VERSION, + RETRY_BACKOFF_MS, + backoffMs, + createNotificationService, +}; diff --git a/server/src/runtime.js b/server/src/runtime.js index c7a39a9..0d44acf 100644 --- a/server/src/runtime.js +++ b/server/src/runtime.js @@ -3,7 +3,15 @@ const { Server: SocketServer } = require("socket.io"); const { createApp } = require("./app"); const { openDatabase } = require("./database"); const { runMigrations } = require("./database/migrate"); +const { createDisplayService } = require("./displays/service"); const { createImageProcessor } = require("./image-processing"); +const { createMqttService } = require("./mqtt"); +const { createNotificationService } = require("./notifications"); +const { + observeAssetProcessingFailure, + observePlayback, + observeStorageHealth, +} = require("./notifications/conditions"); const { createPlaybackService } = require("./playback"); const { createStorageService } = require("./storage"); @@ -27,6 +35,8 @@ function createRuntime({ config, imageProcessor: suppliedImageProcessor, logger = console, + mqttConnect, + notificationFetch, storage: suppliedStorage, }) { const database = openDatabase(config.database); @@ -42,10 +52,41 @@ function createRuntime({ ready: false, stopping: false, }; + const notifications = createNotificationService({ + database, + logger, + webhookUrl: config.notifications.homeAssistantWebhookUrl, + webhookSecret: config.notifications.homeAssistantWebhookSecret, + webhookTimeoutMs: config.notifications.webhookTimeoutMs, + ...(notificationFetch ? { fetchFn: notificationFetch } : {}), + }); + + function safely(action, message) { + try { + action(); + } catch (error) { + try { + logger.warn?.(message, error); + } catch { + // Integration observation is best effort after an authoritative commit. + } + } + } + let storage; let imageProcessor; try { - storage = suppliedStorage || createStorageService(config.storage, { logger }); + storage = + suppliedStorage || + createStorageService(config.storage, { + logger, + onHealthChanged(health) { + safely( + () => observeStorageHealth(notifications, health), + "Storage incident observation failed", + ); + }, + }); imageProcessor = suppliedImageProcessor || createImageProcessor({ maxInputPixels: config.processing.maxInputPixels }); @@ -56,6 +97,8 @@ function createRuntime({ let app; let io = null; let playbackService; + let mqttService; + let displayService; function emitSocketEvent(eventName, payload) { try { io?.emit(eventName, payload); @@ -74,17 +117,50 @@ function createRuntime({ onActivationChanged(event) { emitSocketEvent("playlists:changed", event); }, + // Phase 3B commit ordering is unchanged: this runs after the playback + // transaction has committed, and no integration failure can undo it. onStateChanged(snapshot) { emitSocketEvent("playback:state", snapshot); + safely( + () => mqttService?.publishState(snapshot), + "MQTT playback state publication failed", + ); + safely( + () => observePlayback(notifications, snapshot), + "Playback incident observation failed", + ); + }, + }); + mqttService = createMqttService({ + config, + logger, + playbackService, + ...(mqttConnect ? { connect: mqttConnect } : {}), + }); + displayService = createDisplayService({ + database, + logger, + notifications, + offlineThresholdMs: config.displays.offlineThresholdMs, + onStatusChanged(event) { + emitSocketEvent("display:status", event); }, }); app = createApp({ config, database, + displayService, imageProcessor, + integrations: { mqtt: mqttService, notifications }, lifecycle, logger, migrations, + onAssetProcessingFailed(failure) { + safely( + () => observeAssetProcessingFailure(notifications, failure), + "Asset processing incident observation failed", + ); + }, onPlaylistsChanged(event) { emitSocketEvent("playlists:changed", event); }, @@ -108,7 +184,20 @@ function createRuntime({ } catch (error) { logger.warn?.("Initial playback snapshot could not be sent", error); } + // Display telemetry is read-only reporting. It is validated, bounded, and + // can never mutate playback or playlist state. + socket.on("display:heartbeat", (payload) => { + try { + displayService.heartbeat(payload); + } catch (error) { + logger.warn?.( + `Ignoring malformed display heartbeat from ${socket.id}: ${error?.message}`, + ); + } + }); socket.on("disconnect", () => { + // A closed socket is not an offline display. Only heartbeat staleness + // decides that, because browsers reconnect with a new socket ID. logger.info?.(`Socket.IO client disconnected: ${socket.id}`); }); }); @@ -132,6 +221,11 @@ function createRuntime({ .then((listeningAddress) => { address = listeningAddress; lifecycle.ready = true; + // None of these are readiness-critical. A broker or webhook receiver + // that is down must never keep the API from serving playback. + notifications.start(); + displayService.start(); + safely(() => mqttService.start(), "MQTT integration could not be started"); Promise.resolve() .then(() => storage.start()) .catch(() => { @@ -141,6 +235,8 @@ function createRuntime({ }) .catch((error) => { playbackService.stop(); + notifications.stop(); + displayService.stop(); storage.stop(); database.close(); throw error; @@ -156,6 +252,13 @@ function createRuntime({ lifecycle.ready = false; lifecycle.stopping = true; playbackService.stop(); + notifications.stop(); + displayService.stop(); + const mqttClosed = Promise.resolve() + .then(() => mqttService.stop()) + .catch((error) => { + logger.warn?.("MQTT shutdown failed", error); + }); stopPromise = new Promise((resolve) => { let settled = false; const finish = () => { @@ -185,7 +288,10 @@ function createRuntime({ logger.info?.(`Stopping Announcements API (${reason})`); io.close(finish); - }); + }) + // Shutdown waits for MQTT to retract its retained availability so a + // planned restart is not reported as an unexpected disappearance. + .then(() => mqttClosed); return stopPromise; } @@ -193,10 +299,13 @@ function createRuntime({ app, config, database, + displayService, httpServer, io, lifecycle, migrations, + mqttService, + notifications, playbackService, storage, imageProcessor, diff --git a/server/src/storage/index.js b/server/src/storage/index.js index 36d7da3..3c5fbd0 100644 --- a/server/src/storage/index.js +++ b/server/src/storage/index.js @@ -137,7 +137,15 @@ function createS3Client(config) { return new S3Client(options); } -function createStorageService(config, { client, logger = console, now = () => new Date() } = {}) { +function createStorageService( + config, + { + client, + logger = console, + now = () => new Date(), + onHealthChanged = () => {}, + } = {}, +) { const s3 = client || createS3Client(config); let ensurePromise = null; let healthTimer = null; @@ -157,16 +165,33 @@ function createStorageService(config, { client, logger = console, now = () => ne return issuedHealthSequence; } + // Only a real status transition is reported. Many requests observing the + // same outage must not become many notifications. + function announceHealth(previousStatus) { + if (previousStatus === currentHealth.status) return; + try { + onHealthChanged(publicHealth(currentHealth)); + } catch (error) { + try { + logger.warn?.("Storage health observer failed", error); + } catch { + // Observability must never break a storage operation. + } + } + } + function markHealthy(sequence) { if (sequence < appliedHealthSequence) return publicHealth(currentHealth); appliedHealthSequence = sequence; bucketReady = true; + const previousStatus = currentHealth.status; const checkedAt = now().toISOString(); currentHealth = { status: "healthy", checkedAt, lastSuccessAt: checkedAt, }; + announceHealth(previousStatus); return publicHealth(currentHealth); } @@ -174,6 +199,7 @@ function createStorageService(config, { client, logger = console, now = () => ne if (sequence < appliedHealthSequence) return publicHealth(currentHealth); appliedHealthSequence = sequence; bucketReady = false; + const previousStatus = currentHealth.status; currentHealth = { status: "degraded", checkedAt: now().toISOString(), @@ -181,6 +207,7 @@ function createStorageService(config, { client, logger = console, now = () => ne code: "STORAGE_UNAVAILABLE", message: "Object storage is unavailable", }; + announceHealth(previousStatus); return publicHealth(currentHealth); } diff --git a/server/test/displays.test.js b/server/test/displays.test.js new file mode 100644 index 0000000..7c4c908 --- /dev/null +++ b/server/test/displays.test.js @@ -0,0 +1,376 @@ +const test = require("node:test"); +const assert = require("node:assert/strict"); +const { openDatabase } = require("../src/database"); +const { runMigrations } = require("../src/database/migrate"); +const { createDisplayService } = require("../src/displays/service"); +const { createNotificationService } = require("../src/notifications"); + +const BASE_TIME = 2_000_000_000_000; +const THRESHOLD = 45_000; + +function createClock(initial = BASE_TIME) { + let current = initial; + let sequence = 0; + const timers = new Set(); + return { + now: () => current, + set(value) { + current = value; + }, + setTimeout(callback, delay) { + const timer = { at: current + delay, callback, cleared: false, id: ++sequence }; + timers.add(timer); + return timer; + }, + clearTimeout(timer) { + if (timer) timer.cleared = true; + }, + runTo(value) { + current = value; + for (let guard = 0; guard < 200; guard += 1) { + const timer = [...timers] + .filter((candidate) => !candidate.cleared && !candidate.fired && candidate.at <= current) + .sort((left, right) => left.at - right.at || left.id - right.id)[0]; + if (!timer) return; + timer.fired = true; + timer.callback(); + } + throw new Error("timer loop did not settle"); + }, + }; +} + +function heartbeatPayload(overrides = {}) { + return { + protocolVersion: 1, + clientId: "lobby-display", + outputId: "global", + activePlaylistId: "playlist-a", + playlistRevision: 12, + playbackRevision: 42, + currentEntryId: "entry-a", + currentAssetId: "asset-a", + cachedAssetCount: 10, + playableAssetCount: 10, + mostRecentError: null, + ...overrides, + }; +} + +function createHarness({ start = true } = {}) { + const database = openDatabase({ path: ":memory:" }); + runMigrations(database); + const clock = createClock(); + const events = []; + let idCounter = 0; + const notifications = createNotificationService({ + database, + webhookUrl: "http://127.0.0.1:9/hook", + now: clock.now, + setTimeoutFn: clock.setTimeout, + clearTimeoutFn: clock.clearTimeout, + createId: () => `note-${++idCounter}`, + fetchFn: () => Promise.resolve({ status: 200 }), + logger: { warn() {} }, + }); + + function build() { + const service = createDisplayService({ + database, + notifications, + now: clock.now, + setTimeoutFn: clock.setTimeout, + clearTimeoutFn: clock.clearTimeout, + offlineThresholdMs: THRESHOLD, + onStatusChanged: (event) => events.push(event), + logger: { warn() {} }, + }); + if (start) service.start(); + return service; + } + + return { + build, + clock, + database, + events, + notifications, + service: build(), + outbox: () => + database.prepare("SELECT * FROM notification_outbox ORDER BY created_at, id").all(), + incidents: () => + database.prepare("SELECT * FROM notification_incidents ORDER BY opened_at").all(), + rows: () => database.prepare("SELECT * FROM display_clients").all(), + }; +} + +test("a first heartbeat creates the display and later heartbeats update the same record", (t) => { + const harness = createHarness(); + t.after(() => harness.database.close()); + + const first = harness.service.heartbeat(heartbeatPayload()); + assert.equal(first.status, "online"); + assert.equal(first.clientId, "lobby-display"); + assert.equal(harness.rows().length, 1); + assert.equal(harness.events.at(-1).reason, "first_seen"); + + harness.clock.set(BASE_TIME + 12_000); + const second = harness.service.heartbeat( + heartbeatPayload({ playbackRevision: 43, cachedAssetCount: 11 }), + ); + assert.equal(harness.rows().length, 1, "the persistent client ID is the display identity"); + assert.equal(second.playbackRevision, 43); + assert.equal(second.cachedAssetCount, 11); + assert.equal( + new Date(second.firstSeenAt).getTime(), + BASE_TIME, + "first_seen_at is preserved across heartbeats", + ); + assert.equal(new Date(second.lastSeenAt).getTime(), BASE_TIME + 12_000); + assert.equal( + harness.events.filter((event) => event.reason === "first_seen").length, + 1, + "a steady heartbeat creates no realtime churn", + ); +}); + +test("the same display keeps its identity across a new Socket.IO connection", (t) => { + const harness = createHarness(); + t.after(() => harness.database.close()); + + harness.service.heartbeat(heartbeatPayload()); + // A browser reconnect produces a new socket but the same persisted client ID. + harness.clock.set(BASE_TIME + 20_000); + harness.service.heartbeat(heartbeatPayload()); + + assert.equal(harness.rows().length, 1); + assert.equal(harness.service.list().length, 1); + assert.equal(harness.service.list()[0].status, "online"); +}); + +test("malformed telemetry is rejected safely and never creates a display", (t) => { + const harness = createHarness(); + t.after(() => harness.database.close()); + + const invalid = [ + null, + "string", + [], + {}, + heartbeatPayload({ protocolVersion: 2 }), + heartbeatPayload({ protocolVersion: undefined }), + heartbeatPayload({ clientId: "" }), + heartbeatPayload({ clientId: " " }), + heartbeatPayload({ clientId: 42 }), + heartbeatPayload({ clientId: "x".repeat(300) }), + heartbeatPayload({ cachedAssetCount: -1 }), + heartbeatPayload({ cachedAssetCount: 1.5 }), + heartbeatPayload({ playbackRevision: -5 }), + heartbeatPayload({ mostRecentError: { nested: true } }), + ]; + for (const payload of invalid) { + assert.throws(() => harness.service.heartbeat(payload), TypeError); + } + assert.equal(harness.rows().length, 0); +}); + +test("an oversized error string is bounded before it is stored", (t) => { + const harness = createHarness(); + t.after(() => harness.database.close()); + + const result = harness.service.heartbeat( + heartbeatPayload({ mostRecentError: "E".repeat(5_000_000) }), + ); + + assert.ok(result.mostRecentError.length <= 500, "a client cannot store megabytes"); + assert.ok(harness.rows()[0].last_error.length <= 500); +}); + +test("heartbeats never change playback or playlist revisions", (t) => { + const harness = createHarness(); + t.after(() => harness.database.close()); + const playbackBefore = harness.database + .prepare("SELECT revision, updated_at FROM playback_state WHERE output_id = 'global'") + .get(); + + for (let beat = 0; beat < 5; beat += 1) { + harness.clock.set(BASE_TIME + beat * 10_000); + harness.service.heartbeat(heartbeatPayload({ playbackRevision: 99 })); + } + + assert.deepEqual( + harness.database + .prepare("SELECT revision, updated_at FROM playback_state WHERE output_id = 'global'") + .get(), + playbackBefore, + "telemetry is read-only reporting", + ); +}); + +test("one missed heartbeat does not mark a display offline before the threshold", (t) => { + const harness = createHarness(); + t.after(() => harness.database.close()); + harness.service.heartbeat(heartbeatPayload()); + + // Two heartbeat intervals have passed but the 45s threshold has not. + harness.clock.runTo(BASE_TIME + 30_000); + assert.equal(harness.service.list()[0].status, "online"); + assert.equal(harness.incidents().length, 0); + + harness.clock.runTo(BASE_TIME + THRESHOLD - 1); + assert.equal(harness.service.list()[0].status, "online"); + assert.equal(harness.incidents().length, 0); +}); + +test("crossing the threshold opens exactly one offline incident that never repeats", (t) => { + const harness = createHarness(); + t.after(() => harness.database.close()); + harness.service.heartbeat(heartbeatPayload()); + + harness.clock.runTo(BASE_TIME + THRESHOLD); + assert.equal(harness.service.list()[0].status, "offline"); + const incidents = harness.incidents(); + assert.equal(incidents.length, 1); + assert.equal(incidents[0].event, "display.offline"); + assert.equal(incidents[0].condition_key, "display:lobby-display"); + assert.equal(JSON.parse(incidents[0].context_json).clientId, "lobby-display"); + assert.equal(harness.outbox().length, 1); + assert.equal(harness.events.at(-1).reason, "offline"); + + // Staying offline for a long time must not produce more notifications. + harness.clock.runTo(BASE_TIME + 30 * 60_000); + assert.equal(harness.incidents().length, 1); + assert.equal(harness.outbox().length, 1); + assert.equal( + harness.events.filter((event) => event.reason === "offline").length, + 1, + ); +}); + +test("a returning heartbeat recovers the display exactly once", (t) => { + const harness = createHarness(); + t.after(() => harness.database.close()); + harness.service.heartbeat(heartbeatPayload()); + harness.clock.runTo(BASE_TIME + THRESHOLD); + assert.equal(harness.outbox().length, 1); + + harness.clock.set(BASE_TIME + 60_000); + const recovered = harness.service.heartbeat(heartbeatPayload()); + assert.equal(recovered.status, "online"); + assert.equal(harness.incidents()[0].state, "recovered"); + assert.equal(harness.outbox().length, 2); + assert.equal(harness.outbox()[1].transition, "recovery"); + assert.equal(harness.events.at(-1).reason, "recovered"); + + // Continued healthy heartbeats add nothing further. + harness.clock.set(BASE_TIME + 70_000); + harness.service.heartbeat(heartbeatPayload()); + assert.equal(harness.outbox().length, 2); + assert.equal( + harness.events.filter((event) => event.reason === "recovered").length, + 1, + ); + + // A later disappearance is a new episode with its own single notification. + harness.clock.runTo(BASE_TIME + 70_000 + THRESHOLD); + assert.equal(harness.incidents().length, 2); + assert.equal(harness.outbox().length, 3); +}); + +test("a restart preserves known displays and cannot manufacture offline alerts", (t) => { + const harness = createHarness(); + t.after(() => harness.database.close()); + harness.service.heartbeat(heartbeatPayload()); + harness.service.stop(); + + // The API was down for well over the threshold. A healthy display simply + // has not had the chance to report yet, so startup stays silent until a + // full threshold has elapsed since the restart. + harness.clock.set(BASE_TIME + 10 * 60_000); + const restarted = harness.build(); + assert.equal(harness.rows().length, 1, "display history survives a restart"); + assert.equal(restarted.list()[0].status, "offline", "status still reflects staleness"); + assert.equal(harness.incidents().length, 0, "no notification storm after a restart"); + + harness.clock.runTo(BASE_TIME + 10 * 60_000 + 20_000); + assert.equal(harness.incidents().length, 0, "still inside the startup grace period"); + + restarted.heartbeat(heartbeatPayload()); + assert.equal(restarted.list()[0].status, "online"); + assert.equal(harness.incidents().length, 0, "a display that comes back was never offline"); +}); + +test("a display still missing after the startup grace period is reported once", (t) => { + const harness = createHarness(); + t.after(() => harness.database.close()); + harness.service.heartbeat(heartbeatPayload()); + harness.service.stop(); + + harness.clock.set(BASE_TIME + 10 * 60_000); + harness.build(); + harness.clock.runTo(BASE_TIME + 10 * 60_000 + THRESHOLD); + + assert.equal(harness.incidents().length, 1); + assert.equal(harness.incidents()[0].event, "display.offline"); + assert.equal(harness.outbox().length, 1); +}); + +test("an already-open offline incident is not duplicated by a restart", (t) => { + const harness = createHarness(); + t.after(() => harness.database.close()); + harness.service.heartbeat(heartbeatPayload()); + harness.clock.runTo(BASE_TIME + THRESHOLD); + assert.equal(harness.incidents().length, 1); + harness.service.stop(); + + harness.clock.set(BASE_TIME + 20 * 60_000); + harness.build(); + harness.clock.runTo(BASE_TIME + 20 * 60_000 + THRESHOLD + 1_000); + + assert.equal(harness.incidents().length, 1, "the same episode stays open"); + assert.equal(harness.outbox().length, 1); +}); + +test("a changed error is surfaced as a meaningful status update", (t) => { + const harness = createHarness(); + t.after(() => harness.database.close()); + harness.service.heartbeat(heartbeatPayload()); + harness.events.length = 0; + + harness.clock.set(BASE_TIME + 10_000); + harness.service.heartbeat(heartbeatPayload({ mostRecentError: "decode failed for asset-a" })); + assert.equal(harness.events.length, 1); + assert.equal(harness.events[0].reason, "error_changed"); + assert.equal(harness.events[0].mostRecentError, "decode failed for asset-a"); + + // The same error repeating is not a new event. + harness.clock.set(BASE_TIME + 20_000); + harness.service.heartbeat(heartbeatPayload({ mostRecentError: "decode failed for asset-a" })); + assert.equal(harness.events.length, 1); + + // An arbitrary browser string is telemetry only, never a notification. + assert.equal(harness.incidents().length, 0); + assert.equal(harness.outbox().length, 0); +}); + +test("display status derives from heartbeat freshness for every known client", (t) => { + const harness = createHarness(); + t.after(() => harness.database.close()); + + harness.service.heartbeat(heartbeatPayload({ clientId: "lobby" })); + harness.clock.set(BASE_TIME + 40_000); + harness.service.heartbeat(heartbeatPayload({ clientId: "narthex" })); + harness.clock.set(BASE_TIME + 50_000); + + const displays = Object.fromEntries( + harness.service.list().map((display) => [display.clientId, display]), + ); + assert.equal(displays.lobby.status, "offline"); + assert.equal(displays.narthex.status, "online"); + assert.equal(harness.service.status().online, 1); + assert.equal(harness.service.status().offline, 1); + assert.equal(harness.service.status().known, 2); + + const serialized = JSON.stringify(harness.service.list()); + assert.doesNotMatch(serialized, /socket|sid|transport/i); +}); diff --git a/server/test/integrations.test.js b/server/test/integrations.test.js new file mode 100644 index 0000000..5f5fc9b --- /dev/null +++ b/server/test/integrations.test.js @@ -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/); +}); diff --git a/server/test/mqtt-broker.test.js b/server/test/mqtt-broker.test.js new file mode 100644 index 0000000..689aa81 --- /dev/null +++ b/server/test/mqtt-broker.test.js @@ -0,0 +1,250 @@ +const test = require("node:test"); +const assert = require("node:assert/strict"); +const mqtt = require("mqtt"); +const { loadConfig } = require("../src/config"); +const { createRuntime } = require("../src/runtime"); + +// Protocol behaviour that a hand-written fake cannot prove — retained +// messages, subscription delivery, and a broker-published last will — is +// verified against a real broker. The isolated verification stack sets +// MQTT_TEST_URL; without it these checks are skipped rather than faked. +const BROKER_URL = process.env.MQTT_TEST_URL; +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: BROKER_URL, + ...overrides, + }); +} + +function testStorage() { + return { + async start() {}, + stop() {}, + health: () => ({ status: "healthy", checkedAt: null, lastSuccessAt: null }), + }; +} + +function connectObserver() { + return new Promise((resolve, reject) => { + const client = mqtt.connect(BROKER_URL, { reconnectPeriod: 0 }); + client.once("connect", () => resolve(client)); + client.once("error", reject); + }); +} + +function collect(client, topic) { + const received = []; + client.on("message", (messageTopic, payload) => { + if (messageTopic === topic) received.push(payload.toString("utf8")); + }); + return received; +} + +function waitFor(predicate, { timeoutMs = 5_000, label = "condition" } = {}) { + return new Promise((resolve, reject) => { + const deadline = Date.now() + timeoutMs; + const tick = () => { + let value; + try { + value = predicate(); + } catch (error) { + reject(error); + return; + } + if (value) { + resolve(value); + return; + } + if (Date.now() > deadline) { + reject(new Error(`Timed out waiting for ${label}`)); + return; + } + setTimeout(tick, 25); + }; + tick(); + }); +} + +async function subscribe(client, topic, options = {}) { + await new Promise((resolve, reject) => { + client.subscribe(topic, { qos: 1, ...options }, (error) => + error ? reject(error) : resolve(), + ); + }); +} + +async function publish(client, topic, payload, options = {}) { + await new Promise((resolve, reject) => { + client.publish(topic, payload, { qos: 1, ...options }, (error) => + error ? reject(error) : resolve(), + ); + }); +} + +async function startRuntime(t) { + const runtime = createRuntime({ + config: testConfig(), + logger: silentLogger, + storage: testStorage(), + }); + await runtime.start(); + t.after(async () => { + try { + await runtime.stop("test cleanup"); + } catch { + // A test that already stopped the runtime is fine. + } + }); + await waitFor(() => runtime.mqttService.status().connected, { + label: "the API to connect to the broker", + }); + return { runtime, baseUrl: `http://127.0.0.1:${runtime.address.port}` }; +} + +async function seedPlayback(runtime, baseUrl) { + const timestamp = Date.now(); + for (const [id, hash] of [ + ["asset-one", "a"], + ["asset-two", "c"], + ]) { + 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/x.webp', ?, 'image/webp', 10, 1920, 1080, ?, ?)`, + ) + .run(id, hash.repeat(64), id, `${id}.png`, hash.repeat(64), timestamp, timestamp); + } + const created = await fetch(`${baseUrl}/api/playlists`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ name: "Broker" }), + }); + const playlist = await created.json(); + for (const assetId of ["asset-one", "asset-two"]) { + await fetch(`${baseUrl}/api/playlists/${playlist.id}/entries`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ assetId, durationOverrideMs: 60_000 }), + }); + } + await fetch(`${baseUrl}/api/playlists/${playlist.id}/activate`, { method: "POST" }); + return playlist; +} + +const options = BROKER_URL ? {} : { skip: "MQTT_TEST_URL is not configured" }; + +test("a real broker delivers commands and retains authoritative state", options, async (t) => { + const { runtime, baseUrl } = await startRuntime(t); + await seedPlayback(runtime, baseUrl); + + const observer = await connectObserver(); + t.after(() => observer.endAsync?.() ?? observer.end(true)); + const availability = collect(observer, "announcements/availability"); + const modeState = collect(observer, "announcements/mode/state"); + await subscribe(observer, "announcements/availability"); + await subscribe(observer, "announcements/mode/state"); + + // A late subscriber receives retained state immediately, which is what a + // Home Assistant restart depends on. + await waitFor(() => availability.includes("online"), { + label: "retained availability", + }); + await waitFor(() => modeState.length > 0, { label: "retained mode state" }); + assert.equal(modeState.at(-1), "on"); + + const commander = await connectObserver(); + t.after(() => commander.endAsync?.() ?? commander.end(true)); + await publish(commander, "announcements/mode", "off"); + await waitFor( + () => + runtime.database + .prepare("SELECT mode FROM playback_state WHERE output_id = 'global'") + .get().mode === "off", + { label: "the mode command to commit" }, + ); + await waitFor(() => modeState.at(-1) === "off", { label: "retained mode state to follow" }); + + const before = runtime.playbackService.getSnapshot(); + await publish(commander, "announcements/next", "", { qos: 0 }); + await waitFor( + () => runtime.playbackService.getSnapshot().currentEntryId !== before.currentEntryId, + { label: "the next command to advance playback" }, + ); + + await publish(commander, "announcements/pause", ""); + await waitFor(() => runtime.playbackService.getSnapshot().transportState === "paused", { + label: "pause", + }); + await publish(commander, "announcements/resume", ""); + await waitFor(() => runtime.playbackService.getSnapshot().transportState === "playing", { + label: "resume", + }); + + // An invalid payload is ignored over the real broker exactly as it is + // locally: the last committed mode survives untouched. + await publish(commander, "announcements/mode", " ON "); + await waitFor(() => runtime.playbackService.getSnapshot().mode === "on", { + label: "a whitespace-padded mode command to commit", + }); + await publish(commander, "announcements/mode", "dim"); + await new Promise((resolve) => setTimeout(resolve, 250)); + assert.equal(runtime.playbackService.getSnapshot().mode, "on"); + assert.equal(modeState.at(-1), "on", "no fake state was published for a rejected command"); +}); + +test("a clean API shutdown retracts its retained availability", options, async (t) => { + const { runtime } = await startRuntime(t); + + const observer = await connectObserver(); + t.after(() => observer.endAsync?.() ?? observer.end(true)); + const availability = collect(observer, "announcements/availability"); + await subscribe(observer, "announcements/availability"); + await waitFor(() => availability.includes("online"), { label: "online availability" }); + + await runtime.stop("clean shutdown test"); + + await waitFor(() => availability.at(-1) === "offline", { + label: "availability to be retracted on a planned shutdown", + }); +}); + +test("an ungraceful disappearance makes the broker publish the last will", options, async (t) => { + const observer = await connectObserver(); + t.after(() => observer.endAsync?.() ?? observer.end(true)); + const availability = collect(observer, "test/availability"); + await subscribe(observer, "test/availability"); + + // A crashed or power-cut API never sends DISCONNECT. Only a will can report + // that, so the will configuration is exercised against a real broker here; + // the unit suite proves the service registers exactly these options. + const willed = mqtt.connect(BROKER_URL, { + reconnectPeriod: 0, + will: { topic: "test/availability", payload: "offline", qos: 1, retain: true }, + }); + await new Promise((resolve, reject) => { + willed.once("connect", resolve); + willed.once("error", reject); + }); + await publish(willed, "test/availability", "online", { retain: true }); + await waitFor(() => availability.includes("online"), { label: "online availability" }); + + willed.stream.destroy(); + + await waitFor(() => availability.at(-1) === "offline", { + label: "the broker to publish the last will", + timeoutMs: 10_000, + }); +}); diff --git a/server/test/mqtt.test.js b/server/test/mqtt.test.js new file mode 100644 index 0000000..6e6869a --- /dev/null +++ b/server/test/mqtt.test.js @@ -0,0 +1,403 @@ +const test = require("node:test"); +const assert = require("node:assert/strict"); +const { COMMANDS, createMqttService } = require("../src/mqtt"); + +const PREFIX = "announcements"; + +function createFakeClient() { + const listeners = new Map(); + const client = { + published: [], + subscriptions: [], + unsubscribed: [], + ended: false, + options: null, + 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) { + client.published.push({ topic, payload: String(payload), options }); + callback?.(null); + }, + subscribe(topic, options, callback) { + client.subscriptions.push({ topic, qos: options.qos }); + callback?.(null); + }, + unsubscribe(topic) { + client.unsubscribed.push(topic); + }, + end(force, options, callback) { + client.ended = true; + callback?.(); + }, + }; + return client; +} + +function createPlaybackStub(snapshot = {}) { + const calls = []; + const base = { + mode: "on", + transportState: "playing", + transport: { rotationPaused: false }, + activePlaylistId: "playlist-a", + currentEntryId: "entry-a", + normalTarget: { assetId: "asset-a" }, + staticTarget: null, + playbackRevision: 7, + ...snapshot, + }; + return { + calls, + getSnapshot: () => base, + setMode(mode) { + calls.push(["setMode", mode]); + }, + next() { + calls.push(["next"]); + }, + previous() { + calls.push(["previous"]); + }, + pause() { + calls.push(["pause"]); + }, + resume() { + calls.push(["resume"]); + }, + reload() { + calls.push(["reload"]); + }, + enterStatic(assetId) { + calls.push(["enterStatic", assetId]); + if (assetId === "11111111-1111-4111-8111-111111111111") { + const error = new Error("Library asset not found"); + error.code = "ASSET_NOT_FOUND"; + throw error; + } + if (assetId === "22222222-2222-4222-8222-222222222222") { + const error = new Error("Static playback requires a ready library asset"); + error.code = "ASSET_NOT_READY"; + throw error; + } + if (assetId === "33333333-3333-4333-8333-333333333333") { + const error = new Error("Expired library assets cannot be held statically"); + error.code = "ASSET_EXPIRED"; + throw error; + } + }, + }; +} + +function createHarness({ snapshot, url = "mqtt://broker:1883" } = {}) { + const client = createFakeClient(); + const playbackService = createPlaybackStub(snapshot); + const service = createMqttService({ + config: { + mqtt: { + url, + username: "user", + password: "must-not-leak", + topicPrefix: PREFIX, + reconnectPeriodMs: 5_000, + connectTimeoutMs: 10_000, + }, + }, + playbackService, + logger: { info() {}, warn() {} }, + connect: (connectUrl, options) => { + client.options = options; + client.url = connectUrl; + return client; + }, + }); + return { client, playbackService, service }; +} + +function connectHarness(options) { + const harness = createHarness(options); + harness.service.start(); + harness.client.emit("connect"); + return harness; +} + +test("the default configuration resolves the exact required command topics", () => { + const harness = createHarness(); + assert.deepEqual(harness.service.commandTopics.sort(), [ + "announcements/mode", + "announcements/next", + "announcements/pause", + "announcements/previous", + "announcements/reload", + "announcements/resume", + "announcements/static", + ]); + assert.equal(harness.service.availabilityTopic, "announcements/availability"); + assert.equal(harness.service.modeStateTopic, "announcements/mode/state"); + assert.equal(harness.service.playbackStateTopic, "announcements/playback/state"); +}); + +test("connecting configures a retained last will and publishes availability", () => { + const harness = connectHarness(); + + assert.deepEqual(harness.client.options.will, { + topic: "announcements/availability", + payload: "offline", + qos: 1, + retain: true, + }); + assert.equal(harness.client.options.reconnectPeriod, 5_000); + const availability = harness.client.published.find( + (message) => message.topic === "announcements/availability", + ); + assert.equal(availability.payload, "online"); + assert.equal(availability.options.retain, true); + assert.equal(availability.options.qos, 1); +}); + +test("every required command topic is subscribed with its delivery guarantee", () => { + const harness = connectHarness(); + const byTopic = Object.fromEntries( + harness.client.subscriptions.map((entry) => [entry.topic, entry.qos]), + ); + + assert.equal(byTopic["announcements/mode"], 1); + assert.equal(byTopic["announcements/pause"], 1); + assert.equal(byTopic["announcements/resume"], 1); + assert.equal(byTopic["announcements/reload"], 1); + assert.equal(byTopic["announcements/static"], 1); + // Redelivering a non-idempotent edge command would silently skip a slide. + assert.equal(byTopic["announcements/next"], 0); + assert.equal(byTopic["announcements/previous"], 0); + assert.equal(harness.client.subscriptions.length, COMMANDS.length); +}); + +test("mode commands delegate to the playback service and tolerate whitespace and case", () => { + const harness = connectHarness(); + + harness.client.emit("message", "announcements/mode", Buffer.from("off")); + harness.client.emit("message", "announcements/mode", Buffer.from(" ON \n")); + harness.client.emit("message", "announcements/mode", Buffer.from("OFF")); + + assert.deepEqual(harness.playbackService.calls, [ + ["setMode", "off"], + ["setMode", "on"], + ["setMode", "off"], + ]); +}); + +test("an invalid mode payload changes nothing and publishes no fake state", () => { + const harness = connectHarness(); + const publishedBefore = harness.client.published.length; + + for (const payload of ["dim", "", " ", "1", "{\"mode\":\"on\"}"]) { + harness.client.emit("message", "announcements/mode", Buffer.from(payload)); + } + + assert.deepEqual(harness.playbackService.calls, []); + assert.equal(harness.client.published.length, publishedBefore); + assert.equal(harness.service.status().commandCount, 0); +}); + +test("edge commands invoke the exact playback operation regardless of payload", () => { + const harness = connectHarness(); + + harness.client.emit("message", "announcements/next", Buffer.from("")); + harness.client.emit("message", "announcements/previous", Buffer.from(" ")); + harness.client.emit("message", "announcements/pause", Buffer.from("press")); + harness.client.emit("message", "announcements/resume", Buffer.from("\n")); + harness.client.emit("message", "announcements/reload", null); + + assert.deepEqual(harness.playbackService.calls, [ + ["next"], + ["previous"], + ["pause"], + ["resume"], + ["reload"], + ]); +}); + +test("static accepts a LibraryAsset UUID and rejects anything that is not one", () => { + const harness = connectHarness(); + const valid = "44444444-4444-4444-8444-444444444444"; + + harness.client.emit("message", "announcements/static", Buffer.from(` ${valid} `)); + for (const payload of [ + "not-a-uuid", + "https://example.com/image.png", + "private/display/asset.webp", + "bulletin.png", + "../../etc/passwd", + "", + ]) { + harness.client.emit("message", "announcements/static", Buffer.from(payload)); + } + + assert.deepEqual( + harness.playbackService.calls, + [["enterStatic", valid]], + "only a stable asset UUID may reach the playback service", + ); +}); + +test("a static command the playback service rejects leaves playback untouched", () => { + const harness = connectHarness(); + const publishedBefore = harness.client.published.length; + + for (const assetId of [ + "11111111-1111-4111-8111-111111111111", + "22222222-2222-4222-8222-222222222222", + "33333333-3333-4333-8333-333333333333", + ]) { + harness.client.emit("message", "announcements/static", Buffer.from(assetId)); + } + + // The rejection is logged; nothing is published and the output never goes + // black merely because an operator sent a bad asset ID. + assert.equal(harness.client.published.length, publishedBefore); + assert.equal(harness.service.status().connected, true); +}); + +test("authoritative state is retained on state topics and never echoed to commands", () => { + const harness = connectHarness(); + + harness.service.publishState({ + mode: "off", + transport: { rotationPaused: true }, + activePlaylistId: "playlist-b", + currentEntryId: "entry-b", + normalTarget: { assetId: "asset-b" }, + staticTarget: null, + playbackRevision: 12, + }); + + const modeState = harness.client.published.filter( + (message) => message.topic === "announcements/mode/state", + ); + assert.equal(modeState.at(-1).payload, "off"); + assert.equal(modeState.at(-1).options.retain, true); + assert.equal(modeState.at(-1).options.qos, 1); + + const compact = JSON.parse( + harness.client.published + .filter((message) => message.topic === "announcements/playback/state") + .at(-1).payload, + ); + assert.deepEqual(compact, { + version: 1, + mode: "off", + transport: "paused", + activePlaylistId: "playlist-b", + currentEntryId: "entry-b", + currentAssetId: "asset-b", + staticAssetId: null, + revision: 12, + }); + + const commandTopics = new Set(harness.service.commandTopics); + assert.ok( + harness.client.published.every((message) => !commandTopics.has(message.topic)), + "the server must never publish onto a command topic", + ); +}); + +test("the compact state topic never carries the manifest or private storage details", () => { + const harness = connectHarness(); + harness.service.publishState({ + mode: "on", + transport: { rotationPaused: false }, + activePlaylistId: "playlist-a", + currentEntryId: "entry-a", + normalTarget: { + assetId: "asset-a", + asset: { media: { url: "/media/assets/a/display/abc.webp" } }, + }, + staticTarget: null, + playbackRevision: 3, + manifest: { entries: [{ id: "entry-a" }, { id: "entry-b" }] }, + }); + + const payload = harness.client.published + .filter((message) => message.topic === "announcements/playback/state") + .at(-1).payload; + assert.ok(payload.length < 400, "the compact topic stays small"); + assert.doesNotMatch(payload, /manifest|media|object_key|bucket|private/i); +}); + +test("reconnecting resubscribes and republishes availability and authoritative state", () => { + const harness = connectHarness(); + harness.client.published.length = 0; + harness.client.subscriptions.length = 0; + + harness.client.emit("close"); + assert.equal(harness.service.status().connected, false); + assert.equal(harness.service.status().status, "disconnected"); + + harness.client.emit("connect"); + + assert.equal(harness.client.subscriptions.length, COMMANDS.length); + assert.equal( + harness.client.published.find( + (message) => message.topic === "announcements/availability", + ).payload, + "online", + ); + assert.equal( + harness.client.published.find( + (message) => message.topic === "announcements/mode/state", + ).payload, + "on", + ); + assert.equal(harness.service.status().connectionCount, 2); +}); + +test("a broker error degrades the integration without throwing or leaking the password", () => { + const harness = connectHarness(); + + assert.doesNotThrow(() => + harness.client.emit("error", new Error("connect ECONNREFUSED 10.0.0.5:1883")), + ); + + const status = harness.service.status(); + assert.equal(status.connected, false); + assert.equal(status.status, "disconnected"); + assert.match(status.lastErrorMessage, /ECONNREFUSED/); + assert.doesNotMatch(JSON.stringify(status), /must-not-leak/); +}); + +test("an unconfigured broker leaves the integration disabled and inert", async () => { + const harness = createHarness({ url: null }); + + assert.equal(harness.service.start(), null); + assert.equal(harness.service.status().status, "disabled"); + assert.equal(harness.service.status().configured, false); + assert.doesNotThrow(() => harness.service.publishState({ mode: "on" })); + await harness.service.stop(); + assert.equal(harness.client.published.length, 0); +}); + +test("messages on unknown topics are ignored without crashing", () => { + const harness = connectHarness(); + assert.doesNotThrow(() => { + harness.client.emit("message", "announcements/unknown", Buffer.from("x")); + harness.client.emit("message", "other/mode", Buffer.from("on")); + }); + assert.deepEqual(harness.playbackService.calls, []); +}); + +test("a clean shutdown retracts availability and unsubscribes every command", async () => { + const harness = connectHarness(); + await harness.service.stop(); + + const availability = harness.client.published + .filter((message) => message.topic === "announcements/availability") + .at(-1); + assert.equal(availability.payload, "offline"); + assert.equal(availability.options.retain, true); + assert.equal(harness.client.unsubscribed.length, COMMANDS.length); + assert.equal(harness.client.ended, true); +}); diff --git a/server/test/notifications.test.js b/server/test/notifications.test.js new file mode 100644 index 0000000..b22afec --- /dev/null +++ b/server/test/notifications.test.js @@ -0,0 +1,383 @@ +const test = require("node:test"); +const assert = require("node:assert/strict"); +const { openDatabase } = require("../src/database"); +const { runMigrations } = require("../src/database/migrate"); +const { + RETRY_BACKOFF_MS, + backoffMs, + createNotificationService, +} = require("../src/notifications"); +const { + observeAssetProcessingFailure, + observePlayback, + observeStorageHealth, +} = require("../src/notifications/conditions"); + +const BASE_TIME = 2_000_000_000_000; +const WEBHOOK = "http://127.0.0.1:9/announcements"; + +function createClock(initial = BASE_TIME) { + let current = initial; + let sequence = 0; + const timers = new Set(); + return { + now: () => current, + set(value) { + current = value; + }, + setTimeout(callback, delay) { + const timer = { at: current + delay, callback, cleared: false, id: ++sequence }; + timers.add(timer); + return timer; + }, + clearTimeout(timer) { + if (timer) timer.cleared = true; + }, + pending() { + return [...timers].filter((timer) => !timer.cleared && !timer.fired); + }, + async runTo(value) { + current = value; + for (let guard = 0; guard < 100; guard += 1) { + const timer = this.pending() + .filter((candidate) => candidate.at <= current) + .sort((left, right) => left.at - right.at || left.id - right.id)[0]; + if (!timer) return; + timer.fired = true; + await timer.callback(); + } + throw new Error("timer loop did not settle"); + }, + }; +} + +function createHarness({ webhookUrl = WEBHOOK, responses = [] } = {}) { + const database = openDatabase({ path: ":memory:" }); + runMigrations(database); + const clock = createClock(); + const calls = []; + let idCounter = 0; + + function fetchFn(url, init) { + calls.push({ url, body: JSON.parse(init.body), headers: init.headers }); + const next = responses.shift(); + if (typeof next === "function") return next(); + if (next instanceof Error) return Promise.reject(next); + return Promise.resolve({ status: next ?? 200 }); + } + + function build() { + return createNotificationService({ + database, + webhookUrl, + webhookSecret: "super-secret-webhook-id", + now: clock.now, + setTimeoutFn: clock.setTimeout, + clearTimeoutFn: clock.clearTimeout, + createId: () => `note-${++idCounter}`, + fetchFn, + logger: { warn() {} }, + }); + } + + return { + build, + calls, + clock, + database, + responses, + outbox: () => + database.prepare("SELECT * FROM notification_outbox ORDER BY created_at, id").all(), + incidents: () => + database + .prepare("SELECT * FROM notification_incidents ORDER BY opened_at, episode") + .all(), + }; +} + +const STORAGE_FAILURE = { + event: "storage.unavailable", + severity: "critical", + message: "Announcement storage is unavailable.", + context: { checkedAt: "now" }, +}; + +test("a persistent failure opens one incident and queues exactly one notification", async (t) => { + const harness = createHarness(); + const service = harness.build(); + t.after(() => harness.database.close()); + service.start(); + + const first = service.observeFailure("storage", STORAGE_FAILURE); + assert.equal(first.opened, true); + + // A long outage is observed many times; none of it may become new work. + for (let observation = 0; observation < 25; observation += 1) { + harness.clock.set(BASE_TIME + observation * 60_000); + assert.equal(service.observeFailure("storage", STORAGE_FAILURE).opened, false); + } + + assert.equal(harness.incidents().length, 1); + assert.equal(harness.outbox().length, 1); + assert.equal(harness.incidents()[0].state, "open"); + assert.equal( + harness.incidents()[0].last_observed_at, + BASE_TIME + 24 * 60_000, + "a continuing failure only updates its observation time", + ); +}); + +test("recovery queues exactly one recovery and a later failure is a new episode", async (t) => { + const harness = createHarness(); + const service = harness.build(); + t.after(() => harness.database.close()); + service.start(); + + service.observeFailure("storage", STORAGE_FAILURE); + harness.clock.set(BASE_TIME + 30 * 60_000); + const recovery = service.observeRecovery("storage", { + event: "storage.recovered", + message: "Announcement storage is reachable again.", + }); + assert.equal(recovery.recovered, true); + assert.equal(service.observeRecovery("storage", {}).recovered, false); + assert.equal(harness.outbox().length, 2); + + harness.clock.set(BASE_TIME + 60 * 60_000); + const second = service.observeFailure("storage", STORAGE_FAILURE); + assert.equal(second.opened, true); + const incidents = harness.incidents(); + assert.equal(incidents.length, 2); + assert.deepEqual( + incidents.map((incident) => incident.episode), + [1, 2], + "a repeat condition is a new episode, not a reopened incident", + ); + assert.equal(incidents[0].state, "recovered"); + assert.equal(incidents[1].state, "open"); + assert.equal(harness.outbox().length, 3); +}); + +test("a 2xx response marks delivery complete and carries the versioned payload", async (t) => { + const harness = createHarness({ responses: [204] }); + const service = harness.build(); + t.after(() => harness.database.close()); + service.start(); + + service.observeFailure("storage", STORAGE_FAILURE); + await harness.clock.runTo(BASE_TIME); + + assert.equal(harness.calls.length, 1); + const payload = harness.calls[0].body; + assert.equal(payload.version, 1); + assert.equal(payload.event, "storage.unavailable"); + assert.equal(payload.severity, "critical"); + assert.equal(payload.state, "failure"); + assert.equal(payload.message, "Announcement storage is unavailable."); + assert.match(payload.timestamp, /^\d{4}-\d{2}-\d{2}T/); + assert.ok(payload.notificationId && payload.incidentId); + assert.equal(payload.context.conditionKey, "storage"); + assert.equal(payload.context.episode, 1); + assert.equal(harness.outbox()[0].delivered_at, BASE_TIME); + assert.equal(service.status().pendingCount, 0); + assert.equal(service.status().deliveredCount, 1); +}); + +test("failed deliveries retry on a capped schedule with a stable notification id", async (t) => { + const harness = createHarness({ + // One attempt per documented backoff step: transport errors and both 4xx + // and 5xx responses follow the same retry policy. + responses: [ + new Error("ECONNREFUSED"), + 500, + 404, + new Error("ECONNREFUSED"), + new Error("ECONNREFUSED"), + 200, + ], + }); + const service = harness.build(); + t.after(() => harness.database.close()); + service.start(); + + service.observeFailure("storage", STORAGE_FAILURE); + let elapsed = 0; + await harness.clock.runTo(BASE_TIME); + assert.equal(harness.calls.length, 1); + + // Documented schedule: immediate, then 30s, 1m, 5m, 15m, 30m, capped at 30m. + for (const expected of RETRY_BACKOFF_MS) { + assert.equal( + harness.outbox()[0].next_attempt_at, + BASE_TIME + elapsed + expected, + `retry must be scheduled ${expected}ms after the failed attempt`, + ); + elapsed += expected; + await harness.clock.runTo(BASE_TIME + elapsed); + } + assert.equal(backoffMs(50), 1_800_000, "backoff is capped, not unbounded"); + + const ids = new Set(harness.calls.map((call) => call.body.notificationId)); + assert.equal(ids.size, 1, "retries must reuse one stable delivery identity"); + assert.equal(harness.outbox()[0].delivered_at, BASE_TIME + elapsed); + assert.equal(harness.outbox().length, 1, "retrying never creates new rows"); +}); + +test("open incidents and pending deliveries survive an API restart", async (t) => { + const harness = createHarness({ responses: [new Error("ECONNREFUSED")] }); + const first = harness.build(); + t.after(() => harness.database.close()); + first.start(); + first.observeFailure("storage", STORAGE_FAILURE); + await harness.clock.runTo(BASE_TIME); + assert.equal(harness.outbox()[0].delivered_at, null); + first.stop(); + + // A restart re-reads durable state; it must not re-notify or lose work. + harness.clock.set(BASE_TIME + 5 * 60_000); + const restarted = harness.build(); + restarted.start(); + assert.equal(restarted.isOpen("storage"), true); + assert.equal( + restarted.observeFailure("storage", STORAGE_FAILURE).opened, + false, + "a restart must not reopen an already-open incident", + ); + assert.equal(harness.outbox().length, 1); + assert.equal(restarted.status().pendingCount, 1); + + harness.responses.push(200); + await harness.clock.runTo(BASE_TIME + 10 * 60_000); + assert.equal(harness.outbox()[0].delivered_at, BASE_TIME + 10 * 60_000); + assert.equal(harness.calls.at(-1).body.notificationId, harness.calls[0].body.notificationId); +}); + +test("an unconfigured webhook records incidents without queuing endless delivery work", async (t) => { + const harness = createHarness({ webhookUrl: null }); + const service = harness.build(); + t.after(() => harness.database.close()); + service.start(); + + service.observeFailure("storage", STORAGE_FAILURE); + service.observeRecovery("storage", { event: "storage.recovered" }); + await harness.clock.runTo(BASE_TIME + 60 * 60_000); + + assert.equal(harness.incidents().length, 1, "incident history stays useful"); + assert.equal(harness.outbox().length, 0, "nothing retries against a receiver that does not exist"); + assert.equal(harness.calls.length, 0); + const status = service.status(); + assert.equal(status.configured, false); + assert.equal(status.status, "disabled"); +}); + +test("notification status never reveals the webhook URL or secret", async (t) => { + const harness = createHarness({ responses: [new Error("ECONNREFUSED")] }); + const service = harness.build(); + t.after(() => harness.database.close()); + service.start(); + service.observeFailure("storage", STORAGE_FAILURE); + await harness.clock.runTo(BASE_TIME); + + const serialized = JSON.stringify(service.status()); + assert.doesNotMatch(serialized, /super-secret-webhook-id/); + assert.doesNotMatch(serialized, /127\.0\.0\.1:9/); + assert.doesNotMatch(serialized, /announcements\/?"?$/); + assert.equal(service.status().configured, true); + assert.equal(service.status().openIncidents.length, 1); + assert.equal(service.status().openIncidents[0].conditionKey, "storage"); +}); + +test("storage health transitions drive exactly one incident pair", async (t) => { + const harness = createHarness(); + const service = harness.build(); + t.after(() => harness.database.close()); + service.start(); + + observeStorageHealth(service, { + status: "degraded", + code: "STORAGE_NOT_CHECKED", + }); + assert.equal(harness.incidents().length, 0, "startup state is not an outage"); + + for (let observation = 0; observation < 5; observation += 1) { + observeStorageHealth(service, { + status: "degraded", + code: "STORAGE_UNAVAILABLE", + checkedAt: "2026-08-16T12:00:00.000Z", + }); + } + assert.equal(harness.outbox().length, 1); + + observeStorageHealth(service, { status: "healthy", checkedAt: "2026-08-16T12:30:00.000Z" }); + assert.equal(harness.outbox().length, 2); + assert.equal(harness.outbox()[1].transition, "recovery"); +}); + +test("no playable slides is reported even while a static override fills the screen", async (t) => { + const harness = createHarness(); + const service = harness.build(); + t.after(() => harness.database.close()); + service.start(); + + observePlayback(service, { + activePlaylistId: "playlist-a", + normalTarget: { id: "entry" }, + mode: "on", + }); + assert.equal(harness.incidents().length, 0); + + observePlayback(service, { + activePlaylistId: "playlist-a", + normalTarget: null, + staticTarget: { assetId: "static-asset" }, + mode: "on", + }); + const open = harness.incidents(); + assert.equal(open.length, 1); + assert.equal(open[0].event, "playback.no_playable_slides"); + assert.equal(JSON.parse(open[0].context_json).staticActive, true); + + observePlayback(service, { + activePlaylistId: "playlist-a", + normalTarget: null, + staticTarget: { assetId: "static-asset" }, + mode: "on", + }); + assert.equal(harness.outbox().length, 1, "a continuing condition never re-notifies"); + + observePlayback(service, { + activePlaylistId: "playlist-a", + normalTarget: { id: "entry" }, + mode: "on", + }); + assert.equal(harness.incidents()[0].state, "recovered"); + assert.equal(harness.outbox().length, 2); +}); + +test("each failed upload opens one bounded warning incident", async (t) => { + const harness = createHarness(); + const service = harness.build(); + t.after(() => harness.database.close()); + service.start(); + + observeAssetProcessingFailure(service, { + assetId: "asset-1", + displayName: "Bulletin", + message: "x".repeat(5_000), + }); + observeAssetProcessingFailure(service, { + assetId: "asset-1", + displayName: "Bulletin", + message: "again", + }); + observeAssetProcessingFailure(service, { assetId: "asset-2", displayName: "Notice" }); + + assert.equal(harness.incidents().length, 2, "one incident per failed asset"); + assert.equal(harness.outbox().length, 2); + const context = JSON.parse(harness.incidents()[0].context_json); + assert.ok( + context.processingError.length <= 500, + "a processing error is bounded before it is stored", + ); + assert.equal(harness.incidents()[0].severity, "warning"); +}); diff --git a/server/test/storage-health.test.js b/server/test/storage-health.test.js index 14e146e..da4d240 100644 --- a/server/test/storage-health.test.js +++ b/server/test/storage-health.test.js @@ -30,8 +30,17 @@ function config() { maxFiles: 2, maxInputPixels: 1000, }, - mqtt: { url: null }, - notifications: { homeAssistantWebhookUrl: null }, + mqtt: { + url: null, + topicPrefix: "announcements", + reconnectPeriodMs: 5000, + connectTimeoutMs: 10000, + }, + notifications: { + homeAssistantWebhookUrl: null, + webhookTimeoutMs: 10000, + }, + displays: { offlineThresholdMs: 45000 }, }; }