/** * WebSocket fan-out for /api/v1/live. * * Subscribes to the StateStore (which is backed by Redis pubsub in * prod, in-process events in dev) and re-broadcasts to all connected * clients. * * Hono's @hono/node-ws gives us a `WSContext` that wraps a `ws` WebSocket * with extra Hono-specific methods. We use a thin adapter type so this * module doesn't need to import from hono. */ import type { LiveMessage, UniverseSnapshot } from '@kerbal-rt/shared-types'; import type { StateStore } from '@kerbal-rt/db'; /** Anything with .send(string) and a .readyState we can inspect. */ export interface WsLike { send(data: string): void; readonly readyState: number; } const READY_STATE_OPEN = 1; export interface LiveSocketOpts { store: StateStore; /** Channel label for diagnostics only. */ channel: string; } export class LiveSocketHub { private clients = new Set(); private unsubscribe: (() => Promise) | null = null; constructor(private opts: LiveSocketOpts) {} async start(): Promise { // Subscribe once; share the subscription across all clients. this.unsubscribe = await this.opts.store.subscribe((snap) => { this.broadcast({ type: 'snapshot', snapshot: snap }); }); } async stop(): Promise { if (this.unsubscribe) { await this.unsubscribe(); this.unsubscribe = null; } this.clients.clear(); } onOpen(ws: WsLike): void { this.clients.add(ws); // Send the latest snapshot on connect so the client can render immediately. void this.opts.store.latestSnapshot().then((snap) => { if (snap) this.send(ws, { type: 'snapshot', snapshot: snap }); }); } onMessage(_evt: { data: unknown }, _ws: WsLike): void { // Phase 1: clients are read-only. // Phase 2 will accept subscriptions like { type: 'subscribe', vesselId: '...' } } onClose(ws: WsLike): void { this.clients.delete(ws); } broadcast(msg: LiveMessage): void { const data = JSON.stringify(msg); for (const ws of this.clients) { if (ws.readyState === READY_STATE_OPEN) ws.send(data); } } private send(ws: WsLike, msg: LiveMessage): void { if (ws.readyState === READY_STATE_OPEN) ws.send(JSON.stringify(msg)); } get clientCount(): number { return this.clients.size; } } /** Backward-compat shim for the Phase 0 module shape. */ export function makeLiveHub(opts: LiveSocketOpts): LiveSocketHub { return new LiveSocketHub(opts); } export type { UniverseSnapshot };