Fix live stream buffering and add websocket status

package-lock.json

@@ -8,7 +8,8 @@
       "name": "arcana-fx",
       "version": "0.1.0",
       "dependencies": {
-        "express": "^4.19.2"
+        "express": "^4.19.2",
+        "ws": "^8.18.0"
       },
       "engines": {
         "node": ">=20"
@@ -825,6 +826,27 @@
       "engines": {
         "node": ">= 0.8"
       }
+    },
+    "node_modules/ws": {
+      "version": "8.20.1",
+      "resolved": "https://registry.npmjs.org/ws/-/ws-8.20.1.tgz",
+      "integrity": "sha512-It4dO0K5v//JtTXuPkfEOaI3uUN87iYPnqo/ZzqCoG3g8uhA66QUMs/SrM0YK7/NAu+r4LMh/9dq2A7k+rHs+w==",
+      "license": "MIT",
+      "engines": {
+        "node": ">=10.0.0"
+      },
+      "peerDependencies": {
+        "bufferutil": "^4.0.1",
+        "utf-8-validate": ">=5.0.2"
+      },
+      "peerDependenciesMeta": {
+        "bufferutil": {
+          "optional": true
+        },
+        "utf-8-validate": {
+          "optional": true
+        }
+      }
     }
   }
 }

package.json

@@ -7,12 +7,13 @@
   "scripts": {
     "start": "node src/server.js",
     "dev": "node --watch src/server.js",
-    "check": "node --check src/server.js && node --check src/filestash.js && node --check src/effects.js && node --check src/homeassistant.js && node --check src/streamMixer.js && node --check src/routes.js"
+    "check": "node --check src/server.js && node --check src/filestash.js && node --check src/effects.js && node --check src/homeassistant.js && node --check src/streamMixer.js && node --check src/routes.js && node --check src/wsHub.js"
   },
   "engines": {
     "node": ">=20"
   },
   "dependencies": {
-    "express": "^4.19.2"
+    "express": "^4.19.2",
+    "ws": "^8.18.0"
   }
 }

public/app.js

@@ -46,6 +46,10 @@
     busy: false,
     streamOn: false,
     active: [],
+    statusSocket: null,
+    statusFallbackTimer: null,
+    statusReconnectTimer: null,
+    statusReconnectDelay: 500,
     sets: [],
     currentSetId: null,
     lastVariantPerGroup: new Map(), // anti-immediate-repeat memory
@@ -354,9 +358,65 @@
   async function refreshActive() {
     try {
       const res = await fetchJson('/api/stream/status');
-      state.active = (res.stream && res.stream.active) || [];
-      renderActive();
-    } catch { /* status polling is best-effort */ }
+      applyStreamStatus(res.stream);
+    } catch { /* status fallback is best-effort */ }
+  }
+
+  function applyStreamStatus(stream) {
+    if (!stream) return;
+    state.active = stream.active || [];
+    renderActive();
+  }
+
+  function statusWsUrl() {
+    const proto = window.location.protocol === 'https:' ? 'wss:' : 'ws:';
+    return `${proto}//${window.location.host}/ws/status`;
+  }
+
+  function stopStatusFallback() {
+    if (state.statusFallbackTimer) clearInterval(state.statusFallbackTimer);
+    state.statusFallbackTimer = null;
+  }
+
+  function startStatusFallback() {
+    if (state.statusFallbackTimer) return;
+    refreshActive();
+    state.statusFallbackTimer = setInterval(refreshActive, 5000);
+  }
+
+  function connectStatusSocket() {
+    if (state.statusSocket && (
+      state.statusSocket.readyState === WebSocket.OPEN ||
+      state.statusSocket.readyState === WebSocket.CONNECTING
+    )) return;
+
+    const ws = new WebSocket(statusWsUrl());
+    state.statusSocket = ws;
+
+    ws.addEventListener('open', () => {
+      state.statusReconnectDelay = 500;
+      stopStatusFallback();
+      ws.send(JSON.stringify({ type: 'getStatus' }));
+    });
+
+    ws.addEventListener('message', (event) => {
+      let msg = null;
+      try { msg = JSON.parse(event.data); } catch { return; }
+      if (msg && msg.type === 'streamStatus') applyStreamStatus(msg.stream);
+    });
+
+    ws.addEventListener('close', () => {
+      if (state.statusSocket === ws) state.statusSocket = null;
+      startStatusFallback();
+      clearTimeout(state.statusReconnectTimer);
+      const delay = state.statusReconnectDelay;
+      state.statusReconnectDelay = Math.min(10000, Math.round(state.statusReconnectDelay * 1.7));
+      state.statusReconnectTimer = setTimeout(connectStatusSocket, delay);
+    });
+
+    ws.addEventListener('error', () => {
+      try { ws.close(); } catch { /* noop */ }
+    });
   }
 
   function toggleSet(groupId) {
@@ -453,5 +513,5 @@
   loadEffects();
   renderSetsSelect();
   renderActive();
-  setInterval(refreshActive, 1500);
+  connectStatusSocket();
 })();

src/server.js

@@ -9,6 +9,7 @@ const { EffectsService } = require('./effects');
 const { HomeAssistantClient } = require('./homeassistant');
 const { StreamMixer } = require('./streamMixer');
 const { createApiRouter } = require('./routes');
+const { attachStatusWebSocket } = require('./wsHub');
 
 function buildApp(config) {
   const filestash = new FilestashClient({
@@ -93,7 +94,7 @@ function buildApp(config) {
 
 function main() {
   const config = loadConfig();
-  const { app, effects } = buildApp(config);
+  const { app, effects, mixer } = buildApp(config);
 
   const server = app.listen(config.env.port, () => {
     // eslint-disable-next-line no-console
@@ -108,6 +109,12 @@ function main() {
     }
   });
 
+  const wsHub = attachStatusWebSocket({
+    server,
+    mixer,
+    entity: config.homeAssistant.allowedMediaPlayer,
+  });
+
   // Warm the cache in the background; ignore failures, they will surface on /api/effects.
   effects.list().catch((err) => {
     // eslint-disable-next-line no-console
@@ -118,6 +125,7 @@ function main() {
     process.on(sig, () => {
       // eslint-disable-next-line no-console
       console.log(`Received ${sig}, shutting down.`);
+      wsHub.close();
       server.close(() => process.exit(0));
       setTimeout(() => process.exit(1), 5000).unref();
     });

src/streamMixer.js

@@ -1,5 +1,6 @@
 'use strict';
 
+const EventEmitter = require('node:events');
 const { spawn } = require('node:child_process');
 const crypto = require('node:crypto');
 
@@ -11,8 +12,17 @@ const FRAME_BYTES = Math.floor(SAMPLE_RATE * CHANNELS * BYTES_PER_SAMPLE * FRAME
 const MAX_ACTIVE = 24;
 const MAX_QUEUE_BYTES = FRAME_BYTES * 250; // 5 seconds of decoded audio buffering per effect.
 
-class StreamMixer {
+// Drop a frame for any client whose response socket has more than ~60ms
+// (3 frames) of audio still queued in Node's writable buffer. Real-time
+// soundboard contract: the live edge always advances; if the player is
+// slow we shed silence/mix frames instead of letting the TCP buffer grow
+// into seconds of stale audio. That growth is what caused button-to-sound
+// latency to drift upward the longer the device stream ran.
+const MAX_CLIENT_BUFFER_BYTES = FRAME_BYTES * 3;
+
+class StreamMixer extends EventEmitter {
   constructor({ effects, localMediaBaseUrl }) {
+    super();
     if (!effects) throw new Error('StreamMixer requires effects');
     if (!localMediaBaseUrl) throw new Error('StreamMixer requires localMediaBaseUrl');
     this.effects = effects;
@@ -21,6 +31,8 @@ class StreamMixer {
     this.active = new Map();
     this.timer = null;
     this.sequence = 0;
+    this.nextFrameAt = 0;
+    this.setMaxListeners(64);
   }
 
   stream(res) {
@@ -30,15 +42,27 @@ class StreamMixer {
     res.setHeader('Pragma', 'no-cache');
     res.setHeader('Expires', '0');
     res.setHeader('Connection', 'keep-alive');
+    // Disable Nagle: 20ms PCM frames should be flushed immediately rather
+    // than coalesced by the kernel, which would add a frame or two of jitter.
+    if (res.socket && typeof res.socket.setNoDelay === 'function') {
+      try { res.socket.setNoDelay(true); } catch { /* noop */ }
+    }
     res.write(wavHeader());
 
-    const client = { res, connectedAt: Date.now(), mutedUntil: Date.now() + 80 };
+    const client = {
+      res,
+      connectedAt: Date.now(),
+      mutedUntil: Date.now() + 80,
+      droppedFrames: 0,
+    };
     this.clients.add(client);
     this._ensureTimer();
+    this.emit('change', this._reason('clientConnect'));
 
     res.on('close', () => {
-      this.clients.delete(client);
+      const had = this.clients.delete(client);
       this._maybeStopTimer();
+      if (had) this.emit('change', this._reason('clientDisconnect'));
     });
   }
 
@@ -102,17 +126,21 @@ class StreamMixer {
     });
 
     proc.on('close', (code) => {
+      const wasActive = this.active.has(id);
       item.ended = true;
       if (code && code !== 0 && !item.error) item.error = `ffmpeg exited ${code}`;
+      if (wasActive) this.emit('change', this._reason('end', { id, error: item.error || null }));
     });
 
     proc.on('error', (err) => {
       item.error = err.message;
       item.ended = true;
+      this.emit('change', this._reason('error', { id, error: err.message }));
     });
 
     this.active.set(id, item);
     this._ensureTimer();
+    this.emit('change', this._reason('play', { id }));
     return publicItem(item);
   }
 
@@ -126,6 +154,7 @@ class StreamMixer {
     this.active.delete(id);
     if (item.proc && !item.proc.killed) item.proc.kill('SIGTERM');
     this._maybeStopTimer();
+    this.emit('change', this._reason('stop', { id }));
     return true;
   }
 
@@ -136,6 +165,7 @@ class StreamMixer {
     }
     this.active.clear();
     this._maybeStopTimer();
+    if (count) this.emit('change', this._reason('stopAll', { count }));
     return count;
   }
 
@@ -144,13 +174,19 @@ class StreamMixer {
       sampleRate: SAMPLE_RATE,
       channels: CHANNELS,
       clients: this.clients.size,
+      droppedFrames: [...this.clients].reduce((n, c) => n + c.droppedFrames, 0),
       active: this.listActive(),
     };
   }
 
+  _reason(kind, extra = null) {
+    return { kind, at: Date.now(), extra };
+  }
+
   _ensureTimer() {
     if (this.timer) return;
-    this.timer = setInterval(() => this._tick(), FRAME_MS);
+    this.nextFrameAt = Date.now() + FRAME_MS;
+    this.timer = setInterval(() => this._tick(), Math.max(2, Math.floor(FRAME_MS / 2)));
     this.timer.unref?.();
   }
 
@@ -158,14 +194,29 @@ class StreamMixer {
     if (this.clients.size || this.active.size) return;
     if (this.timer) clearInterval(this.timer);
     this.timer = null;
+    this.nextFrameAt = 0;
   }
 
   _tick() {
+    const now = Date.now();
+    // Wall-clock pacing: never emit PCM frames ahead of real time. If the
+    // process was delayed, resume at the live edge instead of trying to catch
+    // up by writing a burst, which would create exactly the queued-delay
+    // behavior the soundboard is trying to avoid.
+    if (this.nextFrameAt && now + 1 < this.nextFrameAt) return;
+    if (!this.nextFrameAt || now - this.nextFrameAt > FRAME_MS * 2) {
+      this.nextFrameAt = now + FRAME_MS;
+    } else {
+      this.nextFrameAt += FRAME_MS;
+    }
+
     const output = Buffer.alloc(FRAME_BYTES);
     const mix = new Int32Array(FRAME_BYTES / BYTES_PER_SAMPLE);
-    const now = Date.now();
+    let endedSomething = false;
 
     for (const item of [...this.active.values()]) {
+      // If a decoder is late its slot stays silent for this tick rather
+      // than holding the whole mix back — the live edge always advances.
       const frame = consumeFrame(item, FRAME_BYTES);
       if (frame) {
         for (let i = 0; i < mix.length; i += 1) {
@@ -174,6 +225,7 @@ class StreamMixer {
       }
       if (item.ended && item.queueBytes === 0) {
         this.active.delete(item.id);
+        endedSomething = true;
       }
     }
 
@@ -181,19 +233,37 @@ class StreamMixer {
       output.writeInt16LE(clamp(mix[i], -32768, 32767), i * 2);
     }
 
+    let clientChanged = false;
     for (const client of [...this.clients]) {
       if (now < client.mutedUntil) continue;
       if (client.res.destroyed || client.res.writableEnded) {
         this.clients.delete(client);
+        clientChanged = true;
+        continue;
+      }
+      // Backpressure-aware drop. If the player is slow and the writable
+      // buffer is already holding more than MAX_CLIENT_BUFFER_BYTES
+      // (~60ms) of audio, skip this frame instead of stacking onto it.
+      // Without this, the WAV stream silently turns into a multi-second
+      // delay line: button presses still mix correctly but the listener
+      // hears them only after the backlog drains.
+      const buffered = client.res.writableLength | 0;
+      const needsDrain = client.res.writableNeedDrain === true;
+      if (needsDrain || buffered > MAX_CLIENT_BUFFER_BYTES) {
+        client.droppedFrames += 1;
         continue;
       }
       try {
         client.res.write(output);
       } catch {
         this.clients.delete(client);
+        clientChanged = true;
       }
     }
 
+    if (endedSomething || clientChanged) {
+      this.emit('change', this._reason(endedSomething ? 'effectEnded' : 'clientChanged'));
+    }
     this._maybeStopTimer();
   }
 }

src/wsHub.js

@@ -0,0 +1,88 @@
+'use strict';
+
+const { WebSocketServer } = require('ws');
+
+const STATUS_PATH = '/ws/status';
+const AGE_TICK_MS = 1000;
+
+function attachStatusWebSocket({ server, mixer, entity }) {
+  if (!server) throw new Error('attachStatusWebSocket requires server');
+  if (!mixer) throw new Error('attachStatusWebSocket requires mixer');
+
+  const wss = new WebSocketServer({ noServer: true });
+  let ageTimer = null;
+
+  function payload(reason = 'snapshot') {
+    return JSON.stringify({
+      type: 'streamStatus',
+      reason,
+      entity,
+      stream: mixer.stats(),
+      time: new Date().toISOString(),
+    });
+  }
+
+  function send(ws, reason = 'snapshot') {
+    if (ws.readyState !== ws.OPEN) return;
+    try { ws.send(payload(reason)); } catch { /* noop */ }
+  }
+
+  function broadcast(reason = 'change') {
+    for (const ws of wss.clients) send(ws, reason);
+    manageAgeTimer();
+  }
+
+  function manageAgeTimer() {
+    const hasClients = wss.clients.size > 0;
+    const hasActive = mixer.stats().active.length > 0;
+    if (hasClients && hasActive && !ageTimer) {
+      ageTimer = setInterval(() => broadcast('age'), AGE_TICK_MS);
+      ageTimer.unref?.();
+    } else if ((!hasClients || !hasActive) && ageTimer) {
+      clearInterval(ageTimer);
+      ageTimer = null;
+    }
+  }
+
+  server.on('upgrade', (req, socket, head) => {
+    let pathname = '';
+    try {
+      pathname = new URL(req.url || '/', 'http://localhost').pathname;
+    } catch {
+      socket.destroy();
+      return;
+    }
+    if (pathname !== STATUS_PATH) return;
+
+    wss.handleUpgrade(req, socket, head, (ws) => {
+      wss.emit('connection', ws, req);
+    });
+  });
+
+  wss.on('connection', (ws) => {
+    send(ws, 'connect');
+    manageAgeTimer();
+    ws.on('message', (raw) => {
+      let msg = null;
+      try { msg = JSON.parse(String(raw)); } catch { /* ignore */ }
+      if (msg && msg.type === 'getStatus') send(ws, 'requested');
+    });
+    ws.on('close', manageAgeTimer);
+    ws.on('error', () => { try { ws.close(); } catch { /* noop */ } });
+  });
+
+  const onMixerChange = (evt) => broadcast((evt && evt.kind) || 'change');
+  mixer.on('change', onMixerChange);
+
+  return {
+    wss,
+    broadcast,
+    close() {
+      mixer.off('change', onMixerChange);
+      if (ageTimer) clearInterval(ageTimer);
+      wss.close();
+    },
+  };
+}
+
+module.exports = { attachStatusWebSocket, STATUS_PATH };