spb/claude-remote

1/**2 * In-process typed event bus + WS `/events` (text JSON `RtEvent` to every authenticated3 * server client). Slow consumers are dropped, never buffered without bound.4 */5import type { WebSocket } from 'ws';6import { RtEvent, WS_CLOSE } from '@cr/shared';7import type { RuntimePlugin } from './plugins.js';8import { getCore } from './core-util.js';910export type RtEventListener = (event: RtEvent) => void;1112export class EventBus {13  private readonly listeners = new Set<RtEventListener>();1415  /** Validates (the bus is the single choke point for outbound events) and fans out synchronously. */16  emit(event: RtEvent): void {17    const parsed = RtEvent.parse(event);18    for (const l of [...this.listeners]) {19      try {20        l(parsed);21      } catch {22        /* a failing listener must not break the others */23      }24    }25  }2627  subscribe(listener: RtEventListener): () => void {28    this.listeners.add(listener);29    return () => {30      this.listeners.delete(listener);31    };32  }3334  get listenerCount(): number {35    return this.listeners.size;36  }37}3839/** A consumer whose outbound buffer exceeds this is disconnected (it re-syncs over HTTP on reconnect). */40export const EVENTS_MAX_BUFFERED_BYTES = 1024 * 1024;4142const eventsPlugin: RuntimePlugin = async (app) => {43  const core = getCore(app);44  const clients = new Set<WebSocket>();4546  const unsubscribe = core.bus.subscribe((event) => {47    if (clients.size === 0) return;48    const text = JSON.stringify(event);49    for (const ws of clients) {50      if (ws.readyState !== ws.OPEN) continue;51      if (ws.bufferedAmount > EVENTS_MAX_BUFFERED_BYTES) {52        clients.delete(ws);53        app.log.warn({ buffered: ws.bufferedAmount }, 'events: dropping slow consumer');54        ws.close(WS_CLOSE.BACKPRESSURE, 'slow consumer');55        setTimeout(() => ws.terminate(), 2000).unref();56        continue;57      }58      ws.send(text);59    }60  });6162  app.get('/events', { websocket: true }, (ws) => {63    clients.add(ws);64    ws.on('close', () => clients.delete(ws));65    ws.on('error', () => clients.delete(ws));66    // Clients only listen; anything they send is ignored.67    ws.on('message', () => undefined);68  });6970  app.addHook('onClose', async () => {71    unsubscribe();72    for (const ws of clients) ws.close(1001, 'runtime shutting down');73    clients.clear();74  });75};7677export default eventsPlugin;78