maestro/web/src/events.ts

import { eventsWsUrl, listSessions } from "./api.js";
import type { CatalogEvent, Session, Uuid } from "./types.js";

export interface CatalogState {
  sessions: Map<Uuid, Session>;
}

export type CatalogListener = (state: CatalogState) => void;

/** Opens the events socket. Injectable so the reducer and the backoff schedule
 * can be driven without a server. */
export type SocketFactory = () => WebSocket;

/** Schedules a reconnect attempt; returns a handle `clearTimer` understands. */
export interface Scheduler {
  setTimer(fn: () => void, ms: number): number;
  clearTimer(handle: number): void;
}

const browserScheduler: Scheduler = {
  setTimer: (fn, ms) => window.setTimeout(fn, ms),
  clearTimer: (h) => window.clearTimeout(h),
};

/** Initial reconnect delay, doubling to `MAX_BACKOFF_MS`. */
export const INITIAL_BACKOFF_MS = 500;
export const MAX_BACKOFF_MS = 10_000;

export interface CatalogStoreOptions {
  socket?: SocketFactory;
  scheduler?: Scheduler;
  /** Full-state refetch used when the server reports it dropped frames. */
  refetch?: () => Promise<Session[]>;
}

/** Called with `true` when the events socket opens, `false` when it drops.
 *  Drives the daemon link (daemonLink.ts) — the socket dying is the first sign
 *  the daemon went down or was upgraded under the page. */
export type ConnectionListener = (up: boolean) => void;

export class CatalogStore {
  private state: CatalogState = { sessions: new Map() };
  private listeners = new Set<CatalogListener>();
  private connectionListeners = new Set<ConnectionListener>();
  private ws: WebSocket | null = null;
  private reconnectTimer: number | null = null;
  private backoffMs = INITIAL_BACKOFF_MS;
  private readonly socket: SocketFactory;
  private readonly scheduler: Scheduler;
  private readonly refetch: () => Promise<Session[]>;

  constructor(opts: CatalogStoreOptions = {}) {
    this.socket = opts.socket ?? (() => new WebSocket(eventsWsUrl()));
    this.scheduler = opts.scheduler ?? browserScheduler;
    this.refetch = opts.refetch ?? listSessions;
  }

  start(): void {
    this.connect();
  }

  /** Current reconnect delay, for tests asserting the backoff schedule. */
  backoff(): number {
    return this.backoffMs;
  }

  stop(): void {
    if (this.reconnectTimer !== null) {
      this.scheduler.clearTimer(this.reconnectTimer);
      this.reconnectTimer = null;
    }
    if (this.ws) {
      this.ws.onclose = null;
      this.ws.close();
      this.ws = null;
    }
  }

  /** Watch the socket's up/down transitions. Not called on subscribe: the
   *  caller's own probe decides the starting state. */
  subscribeConnection(listener: ConnectionListener): () => void {
    this.connectionListeners.add(listener);
    return () => this.connectionListeners.delete(listener);
  }

  subscribe(listener: CatalogListener): () => void {
    this.listeners.add(listener);
    listener(this.snapshot());
    return () => this.listeners.delete(listener);
  }

  /** A copy, so a listener can diff it against the previous one. Returning the
   * live object meant in-place mutation was invisible and every event forced a
   * full re-render. */
  snapshot(): CatalogState {
    return { sessions: new Map(this.state.sessions) };
  }

  private connect(): void {
    const ws = this.socket();
    this.ws = ws;

    ws.onmessage = (ev) => {
      let event: CatalogEvent;
      try {
        event = JSON.parse(ev.data) as CatalogEvent;
      } catch {
        return;
      }
      this.apply(event);
    };

    ws.onopen = () => {
      this.backoffMs = INITIAL_BACKOFF_MS;
      this.emitConnection(true);
    };

    ws.onclose = () => {
      this.ws = null;
      this.emitConnection(false);
      this.scheduleReconnect();
    };

    ws.onerror = () => {
      try {
        ws.close();
      } catch {
        // ignore
      }
    };
  }

  private scheduleReconnect(): void {
    if (this.reconnectTimer !== null) return;
    const delay = Math.min(this.backoffMs, MAX_BACKOFF_MS);
    this.reconnectTimer = this.scheduler.setTimer(() => {
      this.reconnectTimer = null;
      this.backoffMs = Math.min(this.backoffMs * 2, MAX_BACKOFF_MS);
      this.connect();
    }, delay);
  }

  private apply(event: CatalogEvent): void {
    switch (event.type) {
      case "Snapshot":
        this.state = {
          sessions: new Map(event.sessions.map((s) => [s.id, s])),
        };
        break;
      case "SessionAdded":
        this.state.sessions.set(event.session.id, event.session);
        break;
      case "SessionRemoved":
        this.state.sessions.delete(event.id);
        break;
      case "SessionUpdated":
        this.state.sessions.set(event.session.id, event.session);
        break;
      case "Lagged":
        // Re-derive from REST to avoid showing stale state.
        void this.resync();
        return;
    }
    this.emit();
  }

  private async resync(): Promise<void> {
    try {
      const sessions = await this.refetch();
      this.state = {
        sessions: new Map(sessions.map((s) => [s.id, s])),
      };
      this.emit();
    } catch {
      // surface in UI on next reconnect cycle; nothing else useful to do
    }
  }

  private emitConnection(up: boolean): void {
    for (const fn of this.connectionListeners) fn(up);
  }

  private emit(): void {
    const snap = this.snapshot();
    for (const fn of this.listeners) fn(snap);
  }
}