diff --git a/README.md b/README.md index a56fa13..8856226 100644 --- a/README.md +++ b/README.md @@ -15,9 +15,10 @@ touching the shell (see [docs/MODULE-GUIDE.md](docs/MODULE-GUIDE.md)). 2. Discovers modules in `modules/` and serves each one's client at `/modules//`. 3. Mounts each enabled module's server routes under `/api/modules//` — - that's where module servers proxy ProdCom, poll ProPresenter, and stream - live updates over SSE, so browsers never talk to production gear - directly (no CORS, and API keys never leave the server). + that's where module servers follow ProdCom, poll ProPresenter, and stream + live updates over SSE — one upstream connection per module, however many + dashboards are open — so browsers never talk to production gear directly + (no CORS, and API keys never leave the server). 4. Stores server-wide module config and named layouts in a per-machine data directory outside the app folder, so updating ProdDash never loses them. diff --git a/modules/prodcom-transcript/client.js b/modules/prodcom-transcript/client.js index 439f487..87a1980 100644 --- a/modules/prodcom-transcript/client.js +++ b/modules/prodcom-transcript/client.js @@ -3,8 +3,9 @@ Ported from prodcom-listener/public/app.js into the ProdDash module contract: everything lives inside the tile's root, per-tile state (hidden channels, channel icons, timestamps, text size, flow direction) - persists in instance settings, and all network traffic goes through - moduleApi (the module's server proxy at /prodcom/*). */ + persists in instance settings, and the data comes from the module's own + server, which keeps one connection to ProdCom for every dashboard: a + snapshot on connect (`state`), then each changed entry (`entry`). */ /* the text-size buttons: a drawn − and + at the icon size, like every other header icon */ const MINUS_SVG = ''; @@ -19,7 +20,6 @@ export default function create({ root, moduleApi }) { let pinnedToLatest = true; let hiddenChannels = new Set(moduleApi.instanceSettings.hiddenChannels || []); let stream = null; // moduleApi.sse handle - let bootTimer = null; let stopped = false; /* ── DOM skeleton ───────────────────────────────────────────────── */ @@ -353,120 +353,68 @@ export default function create({ root, moduleApi }) { /* The stream sends JSON objects for new/updated/completed transcript entries. Be liberal about the envelope shape. */ - function extractEntries(obj) { - if (!obj || typeof obj !== 'object') return []; - if (Array.isArray(obj)) return obj.flatMap(extractEntries); - if (typeof obj.text === 'string' && (obj.id || obj.channelId)) return [obj]; - for (const key of ['data', 'payload', 'entry', 'entries', 'transcript']) { - if (obj[key] && typeof obj[key] === 'object') { - const found = extractEntries(obj[key]); - if (found.length) return found; - } - } - return []; + /* ── the module's feed: one snapshot, then each change ─────────── */ + + function applyStatus(st) { + if (!st || typeof st !== 'object') return; + if (st.state === 'ok') moduleApi.setStatus('ok', st.message || 'Live'); + else moduleApi.setStatus(st.state === 'error' ? 'error' : 'connecting', st.message || 'Connecting to ProdCom…'); } - function handleEventData(raw) { - let obj; - try { obj = JSON.parse(raw); } catch { return; } - const kind = String(obj.type || obj.event || '').toLowerCase(); - if (kind.includes('clear')) { - entryEls.forEach((el) => el.remove()); - entryEls.clear(); - emptyEl.hidden = false; - return; - } - extractEntries(obj).forEach(upsertEntry); - } - - /* ── data loading (all via the module's proxy) ──────────────────── */ - - async function loadChannels() { - const res = await moduleApi.fetch('/prodcom/api/v1/channels'); - if (!res.ok) throw new Error('channels ' + res.status); - const body = await res.json(); + function applySnapshot(snap) { + if (!snap || typeof snap !== 'object') return; channels.clear(); - for (const ch of body.data || []) { - channels.set(ch.id, { name: ch.name, color: ch.color || '#8b98a5' }); - } - rebuildChannelMenu(); - } - - async function loadGroups() { - const res = await moduleApi.fetch('/prodcom/api/v1/groups'); - if (!res.ok) throw new Error('groups ' + res.status); - const body = await res.json(); + for (const ch of snap.channels || []) channels.set(String(ch.id), { name: ch.name, color: ch.color || '#8b98a5' }); groups.clear(); - for (const g of body.data || []) { - groups.set(g.id, { name: g.name, channelIds: g.channelIds || [] }); - } + for (const g of snap.groups || []) groups.set(String(g.id), { name: g.name, channelIds: g.channelIds || [] }); rebuildChannelMenu(); + const list = (snap.entries || []).slice().sort((a, b) => new Date(a.date) - new Date(b.date)); + list.forEach(upsertEntry); // dedupes by id, so a snapshot after a reconnect is harmless + applyStatus(snap.status); } - async function loadHistory() { - const res = await moduleApi.fetch('/prodcom/api/v1/transcript?limit=100'); - if (!res.ok) throw new Error('transcript ' + res.status); - const body = await res.json(); - const entries = (body.data || []).slice(); - entries.sort((a, b) => new Date(a.date) - new Date(b.date)); - entries.forEach(upsertEntry); + async function loadState() { + try { + const res = await moduleApi.fetch('/state'); + if (!res.ok) throw new Error(String(res.status)); + applySnapshot(await res.json()); + } catch { /* the stream's own snapshot covers it */ } } - /* ── live stream ────────────────────────────────────────────────── */ - - const SSE_EVENT_NAMES = [ - 'transcript', 'transcript.new', 'transcript.updated', 'transcript.update', - 'transcript.completed', 'transcript.complete', 'entry', 'new', 'update', - 'updated', 'complete', 'completed', - ]; - function connect() { - const onEvent = (ev) => handleEventData(ev.data); - const events = {}; - for (const name of SSE_EVENT_NAMES) events[name] = onEvent; - - stream = moduleApi.sse('/prodcom/api/v1/transcript/stream', { + stream = moduleApi.sse('/stream', { open() { - moduleApi.setStatus('ok', 'Live'); - // refresh channels/groups and backfill entries missed while - // disconnected (upsertEntry dedupes by id, so this is harmless) - loadChannels().catch(() => {}); - loadGroups().catch(() => {}); - loadHistory().catch(() => {}); + loadState(); // a late joiner, or a reconnect: catch up on what was missed }, error() { - moduleApi.setStatus('error', 'ProdCom unreachable — reconnecting…'); + moduleApi.setStatus('error', 'ProdDash ProdCom module unreachable — reconnecting…'); + }, + events: { + state(ev) { + try { applySnapshot(JSON.parse(ev.data)); } catch { /* a bad frame is skipped */ } + }, + entry(ev) { + try { upsertEntry(JSON.parse(ev.data)); } catch { /* skipped */ } + }, + clear() { + clearView(); + }, + status(ev) { + try { applyStatus(JSON.parse(ev.data)); } catch { /* skipped */ } + }, }, - message: onEvent, - events, }); } /* ── lifecycle ──────────────────────────────────────────────────── */ - async function boot() { - if (stopped) return; - moduleApi.setStatus('connecting', 'Connecting to ProdCom…'); - try { - await loadChannels(); - await loadGroups().catch(() => {}); // groups are optional - await loadHistory(); - } catch { - moduleApi.setStatus('error', 'ProdCom unreachable — retrying…'); - bootTimer = setTimeout(boot, 5000); - return; - } - scrollToLatest(); - connect(); - } - return { start() { - boot(); + moduleApi.setStatus('connecting', 'Connecting to ProdCom…'); + connect(); }, stop() { stopped = true; - clearTimeout(bootTimer); stream?.close(); root.innerHTML = ''; // header controls are removed by the shell }, diff --git a/modules/prodcom-transcript/module.json b/modules/prodcom-transcript/module.json index cf5250b..55c4e04 100644 --- a/modules/prodcom-transcript/module.json +++ b/modules/prodcom-transcript/module.json @@ -1,9 +1,9 @@ { "id": "prodcom-transcript", "name": "ProdCom", - "version": "1.2.2", + "version": "1.3.0", "proddash": ">=1.1.0", - "description": "Live ProdCom transcript stream", + "description": "Live ProdCom transcript", "permissions": [ { "kind": "local-network", diff --git a/modules/prodcom-transcript/server.js b/modules/prodcom-transcript/server.js index 0ee3aaa..3d9531f 100644 --- a/modules/prodcom-transcript/server.js +++ b/modules/prodcom-transcript/server.js @@ -3,73 +3,34 @@ /** * ProdCom Transcript — server part. * - * One wildcard route that proxies /prodcom/* (including the SSE transcript - * stream) to the ProdCom API configured in the admin page, so browsers never - * talk to ProdCom directly (no CORS, and the API key never reaches clients). - * Ported from prodcom-listener/server.js. + * One connection to ProdCom for the whole server, however many dashboards + * show the transcript. The module holds the channel list, the groups and + * the last entries, follows ProdCom's SSE stream, and hands every browser a + * snapshot on connect plus each change as it happens. (It used to proxy each + * browser's stream to ProdCom one to one — N screens meant N upstream + * connections; ProdCom now sees one.) + * + * Routes (under /api/modules/prodcom-transcript): + * GET /state → { status, channels, groups, entries } (entries by date) + * GET /stream → SSE: `state` (the snapshot, first), then `entry` for each + * changed entry, `clear`, and `status` when the link changes + * + * ProdCom's API, as the reference listener speaks it: + * GET /api/v1/channels, /api/v1/groups, /api/v1/transcript?limit=N and the + * SSE /api/v1/transcript/stream, whose frames carry an entry, or a + * { type, entry | data } wrapper, in a few historical shapes (see + * extractEntries). A bearer API key is sent when one is configured. */ const http = require('http'); -// Hop-by-hop headers that must not be forwarded -const HOP_HEADERS = new Set([ - 'connection', 'keep-alive', 'proxy-authenticate', 'proxy-authorization', - 'te', 'trailer', 'transfer-encoding', 'upgrade', -]); - -/** Health shown in the admin page: outcome of the most recent proxied request. */ -let health = { status: 'connecting', message: 'No traffic yet' }; - -function proxy(req, res, target, apiKey) { - const upstreamPath = (req.wildcard || '/') + (req.search || ''); - const headers = {}; - for (const [k, v] of Object.entries(req.headers)) { - if (!HOP_HEADERS.has(k.toLowerCase())) headers[k] = v; - } - headers.host = target.host; - delete headers['accept-encoding']; // keep SSE unbuffered and simple - if (apiKey) headers.authorization = 'Bearer ' + apiKey; - - const preq = http.request( - { - hostname: target.hostname, - port: target.port || 80, - path: upstreamPath, - method: req.method, - headers, - }, - (pres) => { - health = { status: 'ok', message: `ProdCom reachable (last: HTTP ${pres.statusCode})` }; - const outHeaders = {}; - for (const [k, v] of Object.entries(pres.headers)) { - if (!HOP_HEADERS.has(k.toLowerCase())) outHeaders[k] = v; - } - res.writeHead(pres.statusCode, outHeaders); - // Flush immediately — SSE streams may send no bytes for a while, and - // the browser needs the headers to consider the stream open. - res.flushHeaders(); - pres.pipe(res); - // If ProdCom dies mid-stream, pipe() unpipes on error without ending - // the browser's response, leaving SSE clients hanging on a dead - // stream. Destroy it so they notice immediately and reconnect. - pres.on('close', () => { - if (!res.writableEnded) res.destroy(); - }); - } - ); - - preq.on('error', (e) => { - health = { status: 'error', message: 'ProdCom unreachable: ' + (e.code || String(e)) }; - if (!res.headersSent) { - res.writeHead(502, { 'Content-Type': 'application/json' }); - } - res.end(JSON.stringify({ error: 'ProdCom unreachable', detail: e.code || String(e) })); - }); - - // Tear down the upstream connection when the browser disconnects - res.on('close', () => preq.destroy()); - req.pipe(preq); -} +const HISTORY_LIMIT = 100; // entries asked for on (re)connect +const ENTRIES_MAX = 300; // entries kept in memory +const RETRY_MIN_MS = 2000; +const RETRY_MAX_MS = 15000; +const REQUEST_TIMEOUT_MS = 8000; +const REFRESH_LISTS_MS = 5 * 60 * 1000; // channels and groups re-read +const HEARTBEAT_MS = 15000; /** The ProdCom base URL from the endpoint config. A legacy prodcomUrl (saved before the endpoint field existed) can only still be present until the @@ -82,23 +43,281 @@ function resolveTarget(config) { return null; } +/** Entries inside whatever ProdCom sent: a bare entry, or one wrapped in data / payload / entry / entries / transcript. */ +function extractEntries(obj) { + if (!obj || typeof obj !== 'object') return []; + if (Array.isArray(obj)) return obj.flatMap(extractEntries); + if (typeof obj.text === 'string' && (obj.id || obj.channelId)) return [obj]; + for (const key of ['data', 'payload', 'entry', 'entries', 'transcript']) { + if (obj[key] && typeof obj[key] === 'object') { + const found = extractEntries(obj[key]); + if (found.length) return found; + } + } + return []; +} + +/** One entry as the tiles read it; an entry without an id is keyed by channel and time. */ +function normalizeEntry(e) { + const id = e.id !== undefined && e.id !== null && e.id !== '' ? String(e.id) : `${e.channelId || ''}|${e.date || ''}`; + return { + id, + channelId: e.channelId === undefined || e.channelId === null ? '' : String(e.channelId), + channelName: e.channelName ? String(e.channelName) : '', + text: String(e.text ?? ''), + date: e.date ? String(e.date) : new Date().toISOString(), + inProgress: Boolean(e.inProgress), + }; +} + +let current = null; // the running instance, for routes() + module.exports = { - init({ config }) { - health = resolveTarget(config) - ? { status: 'connecting', message: 'No traffic yet' } - : { status: 'error', message: 'No ProdCom server configured' }; + init({ config, log }) { + const target = resolveTarget(config); + const apiKey = String(config.apiKey || ''); + const channels = []; + const groups = []; + const entries = new Map(); // id → entry, in arrival order + const streams = new Set(); // browsers' SSE responses + let status = { state: 'connecting', message: target ? 'Connecting to ProdCom…' : 'No ProdCom server configured' }; + let stopped = false; + let upstream = null; + let generation = 0; // callbacks from a superseded connection are ignored + let retryTimer = null; + let retryMs = RETRY_MIN_MS; + + const authHeaders = () => (apiKey ? { authorization: `Bearer ${apiKey}` } : {}); + + async function getJson(pathname) { + const res = await fetch(new URL(pathname, target), { + headers: { accept: 'application/json', ...authHeaders() }, + signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS), + }); + if (!res.ok) throw new Error(`HTTP ${res.status} for ${pathname}`); + return res.json(); + } + + function sortedEntries() { + return [...entries.values()].sort((a, b) => new Date(a.date) - new Date(b.date)); + } + + function snapshot() { + return { status, channels, groups, entries: sortedEntries() }; + } + + function broadcast(event, data) { + if (!streams.size) return; + const frame = `event: ${event}\ndata: ${JSON.stringify(data)}\n\n`; + for (const res of streams) { + try { + res.write(frame); + } catch { + streams.delete(res); + } + } + } + + function setStatus(state, message) { + if (status.state === state && status.message === message) return; + status = { state, message }; + broadcast('status', status); + } + + function trim() { + while (entries.size > ENTRIES_MAX) entries.delete(entries.keys().next().value); + } + + /** A changed entry: kept, and told to every browser. */ + function upsert(raw) { + const e = normalizeEntry(raw); + const prev = entries.get(e.id); + if (prev && prev.text === e.text && prev.inProgress === e.inProgress && prev.channelId === e.channelId) return; + entries.set(e.id, e); + trim(); + broadcast('entry', e); + } + + function clearAll() { + entries.clear(); + broadcast('clear', {}); + } + + async function loadLists() { + const [ch, gr] = await Promise.all([ + getJson('/api/v1/channels'), + getJson('/api/v1/groups').catch(() => ({ data: [] })), // groups are optional in ProdCom + ]); + channels.splice(0, channels.length, ...(ch.data || []).map((c) => ({ id: String(c.id), name: String(c.name || c.id), color: c.color || '#8b98a5' }))); + groups.splice(0, groups.length, ...(gr.data || []).map((g) => ({ id: String(g.id), name: String(g.name || g.id), channelIds: Array.isArray(g.channelIds) ? g.channelIds.map(String) : [] }))); + } + + async function loadHistory() { + const body = await getJson(`/api/v1/transcript?limit=${HISTORY_LIMIT}`); + for (const e of extractEntries(body).map(normalizeEntry)) entries.set(e.id, e); + trim(); + } + + function handleFrame(block) { + const data = []; + for (const line of block.split('\n')) { + if (line.startsWith('data:')) data.push(line.slice(5).replace(/^ /, '')); + } + if (!data.length) return; + let obj; + try { + obj = JSON.parse(data.join('\n')); + } catch { + return; + } + const kind = String(obj?.type || obj?.event || '').toLowerCase(); + if (kind.includes('clear')) return clearAll(); + extractEntries(obj).forEach(upsert); + } + + function scheduleRetry() { + if (stopped) return; + clearTimeout(retryTimer); + retryTimer = setTimeout(connect, retryMs); + retryMs = Math.min(RETRY_MAX_MS, retryMs * 2); + } + + function fail(gen, why) { + if (stopped || gen !== generation) return; + generation += 1; // whatever else this connection reports is stale now + try { upstream?.destroy(); } catch { /* already gone */ } + upstream = null; + setStatus('error', `ProdCom unreachable (${why}) — retrying`); + log(`ProdCom ${target.origin}: ${why} — retrying in ${retryMs / 1000} s`); + scheduleRetry(); + } + + /** Lists and history first (a fresh snapshot for the tiles), then the live stream. */ + async function connect() { + if (stopped || !target) return; + const gen = ++generation; + try { + await loadLists(); + await loadHistory(); + } catch (err) { + if (gen !== generation || stopped) return; + return fail(gen, err.cause?.code || err.code || err.message); + } + if (gen !== generation || stopped) return; + broadcast('state', snapshot()); + const url = new URL('/api/v1/transcript/stream', target); + const req = http.request({ + hostname: url.hostname, + port: url.port || 80, + path: url.pathname, + method: 'GET', + headers: { accept: 'text/event-stream', ...authHeaders() }, + }, (res) => { + if (gen !== generation) return void res.destroy(); + if (res.statusCode !== 200) { + res.resume(); + return fail(gen, `HTTP ${res.statusCode} from the transcript stream`); + } + retryMs = RETRY_MIN_MS; + setStatus('ok', 'Live'); + log(`following ${target.origin} — ${channels.length} channels, ${entries.size} entries`); + let buf = ''; + res.setEncoding('utf8'); + res.on('data', (chunk) => { + if (gen !== generation) return; + buf += chunk; + let i; + while ((i = buf.indexOf('\n\n')) >= 0) { + handleFrame(buf.slice(0, i)); + buf = buf.slice(i + 2); + } + if (buf.length > 1e6) buf = ''; // a frame that never ends is dropped + }); + res.on('end', () => fail(gen, 'stream ended')); + res.on('error', (err) => fail(gen, err.code || err.message)); + }); + req.on('error', (err) => fail(gen, err.code || err.message)); + req.end(); + upstream = req; + } + + const listsTimer = setInterval(() => { + if (stopped || status.state !== 'ok') return; + loadLists().then(() => broadcast('state', snapshot())).catch(() => { /* the next reconnect re-reads them */ }); + }, REFRESH_LISTS_MS); + listsTimer.unref?.(); + const heartbeat = setInterval(() => { + for (const res of streams) { + try { + res.write(': ping\n\n'); + } catch { + streams.delete(res); + } + } + }, HEARTBEAT_MS); + heartbeat.unref?.(); + + if (target) { + log(`one connection to ProdCom at ${target.origin}, shared by every dashboard`); + connect(); + } else { + log('no ProdCom server configured — set the endpoint in /admin'); + } + + current = { + snapshot, + streams, + health() { + if (status.state === 'ok') return { status: 'ok', message: `Live — ${channels.length} channel${channels.length === 1 ? '' : 's'}, ${entries.size} entries` }; + return { status: status.state === 'error' ? 'error' : 'connecting', message: status.message }; + }, + }; + return { - stop() { /* nothing persistent to tear down — streams die with their sockets */ }, - health: () => health, + stop() { + stopped = true; + generation += 1; + clearTimeout(retryTimer); + clearInterval(listsTimer); + clearInterval(heartbeat); + try { upstream?.destroy(); } catch { /* already gone */ } + for (const res of streams) { + try { res.end(); } catch { /* gone */ } + } + streams.clear(); + if (current?.streams === streams) current = null; + }, + health: () => (current ? current.health() : { status: 'connecting', message: 'Restarting…' }), }; }, - routes({ config, log }) { - // An invalid host/URL throws here, which surfaces as a mount error in /admin. - const target = resolveTarget(config) || new URL('http://10.3.11.152:24480'); - log(`proxying /prodcom/* → ${target.origin}`); + routes() { + const sendJson = (res, code, body) => { + res.writeHead(code, { 'Content-Type': 'application/json; charset=utf-8', 'Cache-Control': 'no-store' }); + res.end(JSON.stringify(body)); + }; return { - '* /prodcom/*': (req, res) => proxy(req, res, target, config.apiKey || ''), + 'GET /state': (req, res) => { + if (!current) return sendJson(res, 503, { error: 'ProdCom module is restarting — try again' }); + sendJson(res, 200, current.snapshot()); + }, + 'GET /stream': (req, res) => { + res.writeHead(200, { + 'Content-Type': 'text/event-stream; charset=utf-8', + 'Cache-Control': 'no-cache, no-transform', + Connection: 'keep-alive', + 'X-Accel-Buffering': 'no', + }); + res.write('retry: 3000\n\n'); + if (!current) { + res.write(`event: state\ndata: ${JSON.stringify({ status: { state: 'connecting', message: 'ProdCom module is restarting…' }, channels: [], groups: [], entries: [] })}\n\n`); + return void res.end(); + } + res.write(`event: state\ndata: ${JSON.stringify(current.snapshot())}\n\n`); + try { req.socket.setKeepAlive(true, 15000); } catch { /* gone */ } + current.streams.add(res); + req.on('close', () => current && current.streams.delete(res)); + }, }; }, };