ProdCom 1.3.0: one connection to ProdCom, shared by every dashboard
The module used to proxy each browser's transcript stream to ProdCom one to one, so N screens meant N upstream connections and N history reads. It now follows ProdCom once: it keeps the channel list, the groups and the last 300 entries, and gives every browser a snapshot on connect (`state`) plus each changed entry as it happens (`entry`), with `clear` and `status` frames as the link changes. Reconnects back off from 2 s to 15 s, re-read the lists and history, and re-send the snapshot. The tile no longer loads anything itself; the admin readout says how many channels and entries the server holds. The /prodcom/* proxy is gone. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Waters Church
co-authored by
Claude Fable 5.1
Claude Fable 5.1
parent
ec178e9ccd
commit
96eff1a5c5
@@ -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));
|
||||
},
|
||||
};
|
||||
},
|
||||
};
|
||||
|
||||
Reference in New Issue
Block a user