diff --git a/package-lock.json b/package-lock.json index 02a002b..8d66fde 100644 --- a/package-lock.json +++ b/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 + } + } } } } diff --git a/package.json b/package.json index 7bed1b9..ba036e0 100644 --- a/package.json +++ b/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" } } diff --git a/public/app.js b/public/app.js index c939b9f..d45aca8 100644 --- a/public/app.js +++ b/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(); })(); diff --git a/src/server.js b/src/server.js index b5e29a7..0c6abb3 100644 --- a/src/server.js +++ b/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(); }); diff --git a/src/streamMixer.js b/src/streamMixer.js index 2329a61..c9247c8 100644 --- a/src/streamMixer.js +++ b/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(); } } diff --git a/src/wsHub.js b/src/wsHub.js new file mode 100644 index 0000000..d1eb2ec --- /dev/null +++ b/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 };