111 lines
4.3 KiB
JavaScript
111 lines
4.3 KiB
JavaScript
// SharedWorker that holds ONE EventSource per stream URL and fans
|
||
// every server-sent event out to every connected tab via MessagePort.
|
||
// Design rationale: docs/web-ui/shape.md (SSE multiplexing paragraph).
|
||
//
|
||
// Wire protocol (port.postMessage payloads):
|
||
//
|
||
// tab → worker
|
||
// { kind: 'subscribe', url: '/api/dashboard/stream' }
|
||
// { kind: 'unsubscribe', url: '/api/dashboard/stream' }
|
||
//
|
||
// worker → tab
|
||
// { kind: 'open', url: '...' } relayed from EventSource.onopen; also
|
||
// fired synthetically to new subscribers
|
||
// when upstream is already OPEN (so
|
||
// onStreamOpen runs + re-syncs state).
|
||
// { kind: 'message', url: '...', data: '<raw SSE data string>' }
|
||
// { kind: 'error', url: '...' } relayed from EventSource.onerror.
|
||
// { kind: 'ping' } heartbeat every PING_INTERVAL_MS.
|
||
// Silence past ~3× triggers a re-subscribe
|
||
// (guards Firefox silently GCing the
|
||
// SharedWorker under memory pressure).
|
||
//
|
||
// Per (port, url) subscriptions; unsubscribing the last port for a URL closes
|
||
// the EventSource. One URL in use today; design allows multiple.
|
||
|
||
const streams = new Map();
|
||
// All currently-connected ports. Used by the heartbeat tick to fan
|
||
// pings out across every tab regardless of which URLs each port is
|
||
// subscribed to. Stays disjoint from per-stream `entry.ports` (which
|
||
// is URL-scoped); a port may be in `allPorts` without any active
|
||
// subscription (e.g. between a tab loading and its first subscribe).
|
||
const allPorts = new Set();
|
||
const PING_INTERVAL_MS = 30_000;
|
||
setInterval(() => {
|
||
for (const port of allPorts) {
|
||
try { port.postMessage({ kind: 'ping' }); }
|
||
catch { /* port dead — left in the Set; see onconnect's closing comment */ }
|
||
}
|
||
}, PING_INTERVAL_MS);
|
||
|
||
function getOrCreateStream(url) {
|
||
let entry = streams.get(url);
|
||
if (entry) return entry;
|
||
const es = new EventSource(url);
|
||
entry = { es, url, ports: new Set() };
|
||
es.onopen = () => {
|
||
for (const port of entry.ports) {
|
||
try { port.postMessage({ kind: 'open', url }); }
|
||
catch { /* port dead — cleanup happens on unsubscribe / next subscribe */ }
|
||
}
|
||
};
|
||
es.onmessage = (e) => {
|
||
for (const port of entry.ports) {
|
||
try { port.postMessage({ kind: 'message', url, data: e.data }); }
|
||
catch { /* same */ }
|
||
}
|
||
};
|
||
es.onerror = () => {
|
||
for (const port of entry.ports) {
|
||
try { port.postMessage({ kind: 'error', url }); }
|
||
catch { /* same */ }
|
||
}
|
||
};
|
||
streams.set(url, entry);
|
||
return entry;
|
||
}
|
||
|
||
function unsubscribe(port, url) {
|
||
const entry = streams.get(url);
|
||
if (!entry) return;
|
||
entry.ports.delete(port);
|
||
if (entry.ports.size === 0) {
|
||
entry.es.close();
|
||
streams.delete(url);
|
||
}
|
||
}
|
||
|
||
self.onconnect = (connectEvent) => {
|
||
const port = connectEvent.ports[0];
|
||
allPorts.add(port);
|
||
const subscribedUrls = new Set();
|
||
port.onmessage = (e) => {
|
||
const msg = e.data;
|
||
if (!msg || typeof msg.url !== 'string') return;
|
||
if (msg.kind === 'subscribe') {
|
||
if (subscribedUrls.has(msg.url)) return; // idempotent
|
||
const entry = getOrCreateStream(msg.url);
|
||
entry.ports.add(port);
|
||
subscribedUrls.add(msg.url);
|
||
// Synthetic open for late subscribers — the upstream EventSource
|
||
// may already be OPEN when this tab joins, in which case the
|
||
// native onopen has long since fired and won't fire again until
|
||
// the next reconnect. Hand the new tab the open event explicitly
|
||
// so its onStreamOpen handler runs.
|
||
if (entry.es.readyState === EventSource.OPEN) {
|
||
try { port.postMessage({ kind: 'open', url: msg.url }); }
|
||
catch { /* port dead immediately — give up */ }
|
||
}
|
||
} else if (msg.kind === 'unsubscribe') {
|
||
if (!subscribedUrls.has(msg.url)) return;
|
||
unsubscribe(port, msg.url);
|
||
subscribedUrls.delete(msg.url);
|
||
}
|
||
};
|
||
// A port has no explicit "disconnect" event in the SharedWorker API
|
||
// — tabs close, the GC eventually reclaims the port, but postMessage
|
||
// to a dead port throws which the senders above catch. We don't
|
||
// proactively prune ports on a timer because the cost is bounded
|
||
// (one dead Set entry per stale tab) and the next subscribe / catch
|
||
// catches it.
|
||
};
|