// Live check-in updates: server-sent events with replay from the last event id, // falling back to polling every 3 seconds when the stream is unavailable. import { eventFromWire } from "./adapt-checkins"; import { clinicianApi } from "./client"; import type { CheckInEvent } from "./types"; import type { CheckInEvent as WireEvent } from "./wire"; export type StreamMode = "stream" | "polling" | "closed"; const TYPES = ["status", "note", "note_updated", "transcript", "summary", "urgent", "alert"] as const; export const POLL_MS = 3_000; export type Subscription = { close: () => void }; /** * Follow one check-in from `after` (the snapshot's lastEventId). EventSource resends * Last-Event-ID on its own reconnects; the first connection passes it as a query value. * Two stream failures within a minute switch to polling for the rest of the view. */ export function followCheckIn( checkInId: string, after: number, onEvents: (events: CheckInEvent[]) => void, onMode: (mode: StreamMode) => void, options: { forcePolling?: boolean } = {}, ): Subscription { let last = after; let closed = false; let source: EventSource | null = null; let timer: ReturnType | null = null; let controller: AbortController | null = null; const failures: number[] = []; const deliver = (events: CheckInEvent[]) => { const fresh = events.filter((e) => e.id > last).sort((a, b) => a.id - b.id); if (!fresh.length) return; last = fresh[fresh.length - 1].id; onEvents(fresh); }; const poll = async () => { if (closed) return; controller = new AbortController(); try { const result = await clinicianApi.updates(checkInId, last, controller.signal); deliver(result.events); } catch { // A failed poll is retried on the next tick; the snapshot stays as it was. } if (!closed) timer = setTimeout(poll, POLL_MS); }; const startPolling = () => { source?.close(); source = null; onMode("polling"); void poll(); }; if (options.forcePolling || typeof EventSource === "undefined") startPolling(); else { source = new EventSource(clinicianApi.eventsUrl(checkInId, last), { withCredentials: true }); source.onopen = () => onMode("stream"); for (const type of TYPES) source.addEventListener(type, (message) => { try { deliver([eventFromWire(JSON.parse((message as MessageEvent).data) as WireEvent)]); } catch { // An unreadable event is skipped; the next poll or replay brings it back. } }); source.onerror = () => { const now = Date.now(); failures.push(now); while (failures.length && now - failures[0] > 60_000) failures.shift(); if (source?.readyState === EventSource.CLOSED || failures.length >= 2) startPolling(); }; } return { close() { closed = true; source?.close(); controller?.abort(); if (timer) clearTimeout(timer); onMode("closed"); }, }; }