M8: Real-Time Design

// v1, as found: each widget opens its own connection and hopes
useEffect(() => {
  const ws = new WebSocket(`${WS}/notifications`)            // one of five on this page; no reconnect; no heartbeat; no resume
  ws.onmessage = e => setCount(JSON.parse(e.data).count)
  return () => ws.close()
}, [])
// what happens: the balancer closes idle sockets at 60 s (no heartbeat); a sleep kills all five; the proxy customer gets nothing;
// chat on a drop loses whatever was in flight; a tab left open all day has reconnected zero times and shows yesterday

// v2: one connection per tab; topics with refcounts and lastSeq; backoff; resubscribe; heartbeat; a snapshot fallback
class Realtime {
  private ws?: WebSocket; private topics = new Map<string, { refs: number; seq: number; subs: Set<(e: Event) => void> }>()
  private attempt = 0; private pingTimer?: number; private pongDeadline?: number
  subscribe(topic: string, onEvent: (e: Event) => void, since = 0) {
    const t = this.topics.get(topic) ?? { refs: 0, seq: since, subs: new Set() }
    t.refs++; t.subs.add(onEvent); this.topics.set(topic, t)
    if (t.refs === 1 && this.ws?.readyState === WebSocket.OPEN) this.send({ type: 'sub', topic, since: t.seq })
    this.ensureOpen()
    return () => { t.refs--; t.subs.delete(onEvent); if (t.refs === 0) { this.topics.delete(topic); this.send({ type: 'unsub', topic }) } }
  }
  private ensureOpen() {
    if (this.ws && this.ws.readyState <= WebSocket.OPEN) return
    const ws = this.ws = new WebSocket(URL)
    ws.onopen = () => { this.attempt = 0; for (const [topic, t] of this.topics) this.send({ type: 'sub', topic, since: t.seq }); this.heartbeat() }
    ws.onmessage = ({ data }) => {
      const m = JSON.parse(data)
      if (m.type === 'pong') { this.pongDeadline = undefined; return }
      if (m.type === 'snapshot_required') { this.topics.get(m.topic)?.subs.forEach(fn => fn({ kind: 'resnapshot', topic: m.topic })); return }   // caller refetches via the query layer, then resumes
      if (m.type === 'event') {
        const t = this.topics.get(m.topic); if (!t) return
        if (t.seq && m.seq > t.seq + 1) { this.send({ type: 'sub', topic: m.topic, since: t.seq }); return }   // gap: ask for replay; drop this one (it will come back in order)
        t.seq = m.seq; t.subs.forEach(fn => fn(m))
      }
    }
    ws.onclose = () => { clearInterval(this.pingTimer); const delay = Math.min(30_000, 1000 * 2 ** this.attempt++) * (0.7 + Math.random() * 0.6); setTimeout(() => this.ensureOpen(), delay) }
  }
  private heartbeat() { this.pingTimer = window.setInterval(() => { if (this.pongDeadline && Date.now() > this.pongDeadline) return this.ws?.close(); this.pongDeadline = Date.now() + 10_000; this.send({ type: 'ping' }) }, 25_000) }
  private send(m: unknown) { if (this.ws?.readyState === WebSocket.OPEN) this.ws.send(JSON.stringify(m)) }
}
document.addEventListener('visibilitychange', () => { if (!document.hidden) rt.ensureOpenNow() })   // foreground: assume dead, reconnect now
// React: useRealtime(topic, handler) = useEffect(() => rt.subscribe(topic, handler), [topic]); the handler updates a store or patches the query cache

// v3: per-topic policy. state-like topics coalesce by key; event-like topics batch per frame; hidden tabs downgrade
const latest = new Map<string, unknown>(); const batches = new Map<string, unknown[]>(); let raf = 0
function onEvent(topic: string, e: Event) {
  const kind = policy(topic)                                   // 'state' | 'events' from the topic registry
  if (kind === 'state') latest.set(e.key, e.data) ; else (batches.get(topic) ?? batches.set(topic, []).get(topic)!).push(e.data)
  if (!raf) raf = requestAnimationFrame(flush)
}
function flush() {
  raf = 0
  if (latest.size) { store.applyLatest(latest); latest.clear() }              // one apply per frame, newest only
  for (const [topic, items] of batches) { store.applyBatch(topic, items); batches.delete(topic) }   // every item, one commit
}
// throttle at the source: rt.subscribe('px:EURUSD', h, { maxRate: 10 }) → the server conflates per client
// hidden tab: on hide, unsubscribe high-rate topics (keep their lastSeq); on show, resubscribe with since (replay or snapshot)
// instrumentation: apply lag = now − event.ts (server clock, offset-corrected); queue size; alert at 1 s of lag