maestro/web/src/events.test.ts

import { describe, expect, it, vi } from "vitest";

import {
  CatalogStore,
  INITIAL_BACKOFF_MS,
  MAX_BACKOFF_MS,
  type CatalogState,
  type Scheduler,
} from "./events.js";
import type { CatalogEvent, Session } from "./types.js";

function session(id: string): Session {
  return {
    id,
    cwd: "/home/n",
    command: "bash",
    env: {},
    metadata: {},
    status: "Running",
    created_at: "2026-08-22T00:00:00Z",
  } as Session;
}

/** Just enough of a WebSocket for the store to drive, plus a way to push frames. */
class FakeSocket {
  onmessage: ((ev: { data: string }) => void) | null = null;
  onopen: (() => void) | null = null;
  onclose: (() => void) | null = null;
  onerror: (() => void) | null = null;
  closed = false;

  close(): void {
    this.closed = true;
  }

  open(): void {
    this.onopen?.();
  }

  emit(event: CatalogEvent): void {
    this.onmessage?.({ data: JSON.stringify(event) });
  }

  drop(): void {
    this.onclose?.();
  }
}

/** Scheduler that records delays instead of waiting them out. */
function fakeScheduler(): Scheduler & { delays: number[]; runNext(): void } {
  const queued: Array<() => void> = [];
  return {
    delays: [] as number[],
    setTimer(fn, ms) {
      this.delays.push(ms);
      queued.push(fn);
      return queued.length;
    },
    clearTimer() {},
    runNext() {
      queued.shift()?.();
    },
  };
}

function storeWith(sockets: FakeSocket[], refetch?: () => Promise<Session[]>) {
  let i = 0;
  const scheduler = fakeScheduler();
  const store = new CatalogStore({
    socket: () => sockets[i++] as unknown as WebSocket,
    scheduler,
    refetch,
  });
  return { store, scheduler };
}

describe("CatalogStore reducer", () => {
  it("replaces state on Snapshot and applies deltas", () => {
    const ws = new FakeSocket();
    const { store } = storeWith([ws]);
    store.start();

    ws.emit({ type: "Snapshot", sessions: [session("a")] } as CatalogEvent);
    expect([...store.snapshot().sessions.keys()]).toEqual(["a"]);

    ws.emit({ type: "SessionAdded", session: session("b") } as CatalogEvent);
    expect([...store.snapshot().sessions.keys()].sort()).toEqual(["a", "b"]);

    ws.emit({ type: "SessionRemoved", id: "a" } as CatalogEvent);
    expect([...store.snapshot().sessions.keys()]).toEqual(["b"]);
  });

  it("ignores a frame that is not JSON", () => {
    const ws = new FakeSocket();
    const { store } = storeWith([ws]);
    store.start();
    ws.emit({ type: "Snapshot", sessions: [session("a")] } as CatalogEvent);

    ws.onmessage?.({ data: "not json" });

    expect([...store.snapshot().sessions.keys()]).toEqual(["a"]);
  });

  it("refetches over REST when the server reports dropped frames", async () => {
    const ws = new FakeSocket();
    const refetch = vi.fn(async () => [session("fresh")]);
    const { store } = storeWith([ws], refetch);
    store.start();
    ws.emit({ type: "Snapshot", sessions: [session("stale")] } as CatalogEvent);

    ws.emit({ type: "Lagged", skipped: 4 } as CatalogEvent);
    await vi.waitFor(() => expect(refetch).toHaveBeenCalledOnce());

    expect([...store.snapshot().sessions.keys()]).toEqual(["fresh"]);
  });

  it("hands each subscriber a copy it can diff against the next one", () => {
    const ws = new FakeSocket();
    const { store } = storeWith([ws]);
    const seen: CatalogState[] = [];
    store.subscribe((s) => seen.push(s));
    store.start();

    ws.emit({ type: "Snapshot", sessions: [session("a")] } as CatalogEvent);
    ws.emit({ type: "SessionAdded", session: session("b") } as CatalogEvent);

    const [, afterSnapshot, afterAdd] = seen;
    expect(afterSnapshot.sessions).not.toBe(afterAdd.sessions);
    expect(afterSnapshot.sessions.size).toBe(1);
    expect(afterAdd.sessions.size).toBe(2);
  });
});

describe("CatalogStore reconnect", () => {
  it("doubles the delay per attempt and caps it", () => {
    const sockets = Array.from({ length: 8 }, () => new FakeSocket());
    const { store, scheduler } = storeWith(sockets);
    store.start();

    for (let i = 0; i < 7; i++) {
      sockets[i].drop();
      scheduler.runNext();
    }

    expect(scheduler.delays[0]).toBe(INITIAL_BACKOFF_MS);
    expect(scheduler.delays[1]).toBe(INITIAL_BACKOFF_MS * 2);
    expect(scheduler.delays.at(-1)).toBe(MAX_BACKOFF_MS);
  });

  it("resets the delay once a connection opens", () => {
    const sockets = [new FakeSocket(), new FakeSocket(), new FakeSocket()];
    const { store, scheduler } = storeWith(sockets);
    store.start();

    sockets[0].drop();
    scheduler.runNext();
    expect(store.backoff()).toBe(INITIAL_BACKOFF_MS * 2);

    sockets[1].open();
    expect(store.backoff()).toBe(INITIAL_BACKOFF_MS);
  });
});

describe("connection listeners", () => {
  it("report the socket opening and dropping", () => {
    const sock = new FakeSocket();
    const store = new CatalogStore({
      socket: () => sock as unknown as WebSocket,
      scheduler: fakeScheduler(),
    });
    const seen: boolean[] = [];
    store.subscribeConnection((up) => seen.push(up));
    store.start();
    // Nothing on subscribe — the daemon link's own probe sets the first state.
    expect(seen).toEqual([]);
    sock.open();
    sock.drop();
    expect(seen).toEqual([true, false]);
  });
});