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) { 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]); }); });