diff --git a/public/shell.js b/public/shell.js index 88cf4f0..a7f33df 100644 --- a/public/shell.js +++ b/public/shell.js @@ -794,80 +794,130 @@ function tileMessage(entry, text) { entry.body.appendChild(msg); } -/* ── shared event streams ───────────────────────────────────────────── */ +/* ── one stream per page ─────────────────────────────────────────────── */ /* Browsers allow about six HTTP/1.1 connections to one host, and an open - EventSource holds one of them for as long as it lives. One stream per - tile plus the shell's own /api/events reaches that ceiling at five or six - tiles, and from then on every further request — the + picker's module - list, a new tile's /state, its stream — sits in the browser's queue - waiting for a connection that is never handed back. So tiles that ask - for the same stream share one connection: the pool keeps one EventSource - per URL, fans its events out to every subscriber, and reconnects it (on - the same 3-second timer the per-tile version used) while anyone is still - listening. A tile joining an already-open stream is told `open` straight - away, so its backfill runs exactly as if the connection were its own. */ + EventSource holds one of them for as long as it lives. Sharing one + stream per URL was not enough: a dashboard with six modules still ran + out (Safari first), and from then on every further request — a remounted + tile's /state, an avatar, the + picker's list — sat in the browser's + queue waiting for a connection that was never handed back. So the page + holds ONE EventSource, GET /api/stream?s=, and the server relays + every stream named in it (a module's, or `shell:` for the shell's own) + with the key prefixed to each event name. Subscribers still see the + events they asked for by their own names. Adding or dropping a stream + remakes the connection with the new list, a moment later so several + tiles mounting together cause one reconnect; every stream announces + `open` again after it, so backfills run as before. */ -const ssePool = new Map(); // url -> shared source +const muxSubs = new Map(); // key -> { subs: Set<{ handlers }> } +let muxEs = null; // the page's one EventSource +let muxKeys = ''; // the key list it was opened with +let muxTimer = null; // the pending (re)connect +let muxRetry = null; // retry after the connection gave up +const muxListened = new Set(); // "key|name" pairs with a listener on muxEs + +/** The stream key for a URL: `/api/modules/` → `:`, `/api/events` → `shell:`. */ +function muxKeyFor(url) { + if (url === '/api/events') return 'shell:'; + const m = /^\/api\/modules\/([^/?#]+)(\/[^?#]*)?$/.exec(url); + return m ? `${decodeURIComponent(m[1])}:${m[2] || '/'}` : null; +} + +function muxFanOut(key, kind, e) { + for (const sub of [...(muxSubs.get(key)?.subs || [])]) { + const fn = kind === 'event' ? sub.handlers.events?.[e.type] : sub.handlers[kind]; + try { fn?.(e); } catch (err) { console.error('[shell] stream handler failed:', err); } + } +} + +/** Listen for `|` on the open connection and hand it to that key's subscribers as `name`. */ +function muxListen(key, name) { + const full = `${key}|${name}`; + if (!muxEs || muxListened.has(full)) return; + muxListened.add(full); + muxEs.addEventListener(full, (e) => { + if (name === '__open') return muxFanOut(key, 'open', new Event('open')); + if (name === '__down') return muxFanOut(key, 'error', new Event('error')); + const ev = { type: name, data: e.data, lastEventId: e.lastEventId }; + muxFanOut(key, name === 'message' ? 'message' : 'event', ev); + }); +} + +function muxConnect() { + clearTimeout(muxTimer); + muxTimer = null; + clearTimeout(muxRetry); + muxRetry = null; + const keys = [...muxSubs.keys()].sort(); + const wanted = keys.join(','); + if (!keys.length) { + try { muxEs?.close(); } catch { /* not open */ } + muxEs = null; + muxKeys = ''; + return; + } + if (muxEs && muxKeys === wanted && muxEs.readyState !== EventSource.CLOSED) return; + try { muxEs?.close(); } catch { /* not open */ } + muxListened.clear(); + muxKeys = wanted; + const es = new EventSource(`/api/stream?s=${encodeURIComponent(wanted)}`); + muxEs = es; + // Each stream says `open` for itself (the server's __open), so nothing to do on the connection's own open. + es.onerror = () => { + if (es !== muxEs) return; + for (const key of muxSubs.keys()) muxFanOut(key, 'error', new Event('error')); + // EventSource retries transient drops itself but gives up for good on a + // completed non-SSE response (a 502 while the server restarts). + if (es.readyState === EventSource.CLOSED && muxSubs.size) { + clearTimeout(muxRetry); + muxRetry = setTimeout(muxConnect, 3000); + } + }; + for (const [key, entry] of muxSubs) { + muxListen(key, '__open'); + muxListen(key, '__down'); + muxListen(key, 'message'); + for (const sub of entry.subs) for (const name of Object.keys(sub.handlers.events || {})) muxListen(key, name); + } +} function sseSubscribe(url, handlers = {}) { - let shared = ssePool.get(url); - if (!shared) { - shared = { subs: new Set(), es: null, retryTimer: null, names: new Set() }; - const fanOut = (kind, e) => { - for (const sub of [...shared.subs]) { - const fn = kind === 'event' ? sub.handlers.events?.[e.type] : sub.handlers[kind]; - try { fn?.(e); } catch (err) { console.error('[shell] stream handler failed:', err); } - } - }; - shared.listen = (name) => shared.es.addEventListener(name, (e) => fanOut('event', e)); - shared.connect = () => { - clearTimeout(shared.retryTimer); - shared.retryTimer = null; - try { shared.es?.close(); } catch { /* already closed */ } - const es = new EventSource(url); - shared.es = es; - es.onopen = (e) => fanOut('open', e); - es.onmessage = (e) => fanOut('message', e); - es.onerror = (e) => { - fanOut('error', e); - // EventSource retries transient drops itself but gives up for good on - // a completed non-SSE response (a 502 while the upstream is down). - if (es.readyState === EventSource.CLOSED && shared.subs.size) { - clearTimeout(shared.retryTimer); - shared.retryTimer = setTimeout(shared.connect, 3000); - } - }; - for (const name of shared.names) shared.listen(name); - }; - ssePool.set(url, shared); + const key = muxKeyFor(url); + if (!key) { + console.warn('[shell] not a stream this page can carry:', url); + return { close() {} }; + } + let entry = muxSubs.get(key); + if (!entry) { + entry = { subs: new Set() }; + muxSubs.set(key, entry); } - const sub = { handlers }; - shared.subs.add(sub); - for (const name of Object.keys(handlers.events || {})) { - if (shared.names.has(name)) continue; - shared.names.add(name); - if (shared.es) shared.listen(name); + entry.subs.add(sub); + const carried = Boolean(muxEs) && muxKeys.split(',').includes(key) && muxEs.readyState !== EventSource.CLOSED; + if (carried) { + // Joining a stream the connection already carries: listeners for any + // new event names, and `open` a moment later so the caller's own setup + // finishes first — exactly as a fresh connection would have reported it. + for (const name of Object.keys(handlers.events || {})) muxListen(key, name); + if (muxEs.readyState === EventSource.OPEN) { + setTimeout(() => { + if (!entry.subs.has(sub)) return; + try { handlers.open?.(new Event('open')); } catch (err) { console.error('[shell] stream handler failed:', err); } + }, 0); + } + } else { + clearTimeout(muxTimer); + muxTimer = setTimeout(muxConnect, 20); } - if (!shared.es || shared.es.readyState === EventSource.CLOSED) { - // first subscriber, or the stream had given up while nobody listened - shared.connect(); - } else if (shared.es.readyState === EventSource.OPEN) { - // Joining mid-flight: report the open the way a fresh connection would - // have, a moment later, so the caller's own setup finishes first. - setTimeout(() => { - if (!shared.subs.has(sub)) return; - try { handlers.open?.(new Event('open')); } catch (err) { console.error('[shell] stream handler failed:', err); } - }, 0); - } - return { close() { - if (!shared.subs.delete(sub) || shared.subs.size) return; - clearTimeout(shared.retryTimer); - try { shared.es?.close(); } catch { /* already closed */ } - ssePool.delete(url); + if (!entry.subs.delete(sub) || entry.subs.size) return; + muxSubs.delete(key); + // A remount closes and reopens within milliseconds: wait before remaking the connection. + clearTimeout(muxTimer); + muxTimer = setTimeout(muxConnect, 500); }, }; } @@ -1959,20 +2009,14 @@ new ResizeObserver(() => { config or enabled flags change. Tiles of a changed module get onConfigChange(cfg) if they implement it, otherwise a clean remount. */ -let eventsSource = null; -let eventsRetry = null; - function subscribeEvents() { - try { eventsSource?.close(); } catch { /* not open */ } - eventsSource = new EventSource('/api/events'); - eventsSource.addEventListener('modules-changed', () => handleModulesChanged()); - eventsSource.addEventListener('theme', handleThemeEvent); - eventsSource.onerror = () => { - if (eventsSource.readyState === EventSource.CLOSED) { - clearTimeout(eventsRetry); - eventsRetry = setTimeout(subscribeEvents, 4000); - } - }; + // Rides on the page's one stream connection (see sseSubscribe) as `shell:`. + sseSubscribe('/api/events', { + events: { + 'modules-changed': () => handleModulesChanged(), + theme: handleThemeEvent, + }, + }); } async function handleModulesChanged() { diff --git a/server.js b/server.js index 9b7de98..4c77a57 100644 --- a/server.js +++ b/server.js @@ -559,6 +559,142 @@ setInterval(() => { } }, 25000).unref(); +/* ── one stream per page ───────────────────────────────────────────── + Browsers allow about six HTTP/1.1 connections to a host and an open + EventSource holds one for as long as it lives. A dashboard with six + modules ran out — Safari first — and from then on every further request + (a remounted tile's /state, an avatar) waited in the browser's queue for + ever. So a page opens ONE EventSource, GET /api/stream?s=, naming + the streams it wants: `slack:/stream`, `pco-plan:/timers/stream`, `shell:` + for the shell's own events. The server opens each module stream for it + over loopback (the module's route is untouched and still serves curl and + other tools), and forwards every frame with the event name prefixed by + its key — `event: slack:/stream|state`. Two frames of its own bracket a + stream's life: `|__open` when the upstream answers, `|__down` + when it ends (it is re-opened after a pause while the page stays). */ + +const MUX_MAX_STREAMS = 32; +const MUX_RETRY_MS = 3000; +const MUX_RETRY_REFUSED_MS = 10000; +const MUX_PING_MS = 15000; +const MUX_KEY_RE = /^([A-Za-z0-9_-]+):(\/[A-Za-z0-9_\-./]*)?$/; + +function parseMuxKeys(search) { + const params = new URLSearchParams(search || ''); + const out = []; + const seen = new Set(); + for (const raw of String(params.get('s') || '').split(',')) { + const key = raw.trim(); + const m = MUX_KEY_RE.exec(key); + if (!m || seen.has(key)) continue; + seen.add(key); + out.push({ key, id: m[1], subPath: m[2] || '' }); + if (out.length >= MUX_MAX_STREAMS) break; + } + return out; +} + +function handleMuxStream(req, res, search) { + const wanted = parseMuxKeys(search); + 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'); + try { req.socket.setKeepAlive(true, 15000); } catch { /* gone */ } + let closed = false; + const send = (frame) => { + if (closed) return; + try { + res.write(frame); + } catch { + /* the client is gone; 'close' follows */ + } + }; + + // The shell's own events ride along under `shell:` — a stand-in sits in + // shellStreams and rewrites the event line of every frame it is handed. + let shellProxy = null; + const upstreams = []; + for (const entry of wanted) { + if (entry.id === 'shell') { + if (shellProxy) continue; + shellProxy = { write(frame) { send(String(frame).replace(/^event: /m, `event: ${entry.key}|`)); } }; + shellStreams.add(shellProxy); + send(`event: ${entry.key}|__open\ndata: {}\n\n`); + continue; + } + upstreams.push({ ...entry, req: null, timer: null, gone: false }); + } + + function forwardBlock(entry, block) { + let name = 'message'; + const data = []; + for (const line of block.split('\n')) { + if (line.startsWith(':')) continue; // the upstream's keepalive; this connection pings on its own + if (line.startsWith('event:')) name = line.slice(6).trim() || 'message'; + else if (line.startsWith('data:')) data.push(line.slice(5).replace(/^ /, '')); + // id: and retry: belong to the upstream connection, not to the page's + } + if (!data.length) return; + send(`event: ${entry.key}|${name}\n${data.map((d) => `data: ${d}`).join('\n')}\n\n`); + } + + function down(entry, why, delay) { + if (closed || entry.gone) return; + entry.gone = true; + send(`event: ${entry.key}|__down\ndata: ${JSON.stringify({ why })}\n\n`); + try { entry.req?.destroy(); } catch { /* already gone */ } + entry.req = null; + entry.timer = setTimeout(() => { + entry.gone = false; + open(entry); + }, delay); + } + + function open(entry) { + if (closed) return; + const path = `/api/modules/${encodeURIComponent(entry.id)}${entry.subPath || '/'}`; + const up = http.request({ host: '127.0.0.1', port: PORT, path, method: 'GET', headers: { accept: 'text/event-stream' } }, (upRes) => { + if (upRes.statusCode !== 200 || !/text\/event-stream/.test(String(upRes.headers['content-type'] || ''))) { + upRes.resume(); + return down(entry, `HTTP ${upRes.statusCode}`, MUX_RETRY_REFUSED_MS); + } + send(`event: ${entry.key}|__open\ndata: {}\n\n`); + let buf = ''; + upRes.setEncoding('utf8'); + upRes.on('data', (chunk) => { + buf += chunk; + let i; + while ((i = buf.indexOf('\n\n')) >= 0) { + forwardBlock(entry, buf.slice(0, i)); + buf = buf.slice(i + 2); + } + if (buf.length > 1e6) buf = ''; // a frame that never ends is dropped, not kept + }); + upRes.on('end', () => down(entry, 'ended', MUX_RETRY_MS)); + upRes.on('error', (err) => down(entry, err.message, MUX_RETRY_MS)); + }); + up.on('error', (err) => down(entry, err.message, MUX_RETRY_MS)); + up.end(); + entry.req = up; + } + + for (const entry of upstreams) open(entry); + const ping = setInterval(() => send(': ping\n\n'), MUX_PING_MS); + req.on('close', () => { + closed = true; + clearInterval(ping); + if (shellProxy) shellStreams.delete(shellProxy); + for (const entry of upstreams) { + clearTimeout(entry.timer); + try { entry.req?.destroy(); } catch { /* already gone */ } + } + }); +} + /* ── themes ─────────────────────────────────────────────────────────── */ /* Preset palettes for the whole dashboard, chosen in Admin → Theme. The @@ -1448,6 +1584,11 @@ function handleRequest(req, res) { return sendJson(res, 200, themeView()); } + /* — every stream a page wants, over one connection — */ + if (urlPath === '/api/stream' && req.method === 'GET') { + return handleMuxStream(req, res, search); + } + /* — shell events: open dashboards learn about admin changes live — */ if (urlPath === '/api/events' && req.method === 'GET') { res.writeHead(200, {