fabro/apps/fabro-web/app/lib/live-events.test.tsx

import { describe, expect, test } from "bun:test";

import {
  createCrossTabSseCoordinator,
  type BroadcastChannelLike,
} from "./cross-tab-sse";
import { subscribeToLiveEvents, type LiveEventPayload } from "./live-events";
import type { EventSourceLike } from "./sse";

type MessageHandler = ((event: { data: string }) => void) | null;

class FakeEventSource {
  onmessage: MessageHandler = null;
  closed = false;

  emit(payload: unknown) {
    this.onmessage?.({ data: JSON.stringify(payload) });
  }

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

class FakeBroadcastChannel implements BroadcastChannelLike {
  onmessage: ((event: { data: unknown }) => void) | null = null;

  postMessage() {}

  close() {}
}

describe("subscribeToLiveEvents", () => {
  test("coordinated mode opens /api/v1/attach and forwards every payload", async () => {
    const source = new FakeEventSource();
    const created: string[] = [];
    const seen: LiveEventPayload[] = [];
    const coordinator = createCoordinator((url) => {
      created.push(url);
      return source;
    });

    const cleanup = subscribeToLiveEvents(
      (payload) => seen.push(payload),
      () => {
        throw new Error("source should be created by coordinator");
      },
      { coordinator },
    );

    await waitFor(() => created.length === 1);

    source.emit({ id: "evt-1", event: "stage.started", run_id: "run-a" });
    source.emit({ id: "evt-2", event: "agent.message", run_id: "run-b" });

    expect(created).toEqual(["/api/v1/attach"]);
    expect(seen.map((p) => p.run_id)).toEqual(["run-a", "run-b"]);

    cleanup();
    coordinator.close();
  });

  test("fallback mode opens /api/v1/attach (not a per-run URL) and forwards payloads", () => {
    const source = new FakeEventSource();
    const created: string[] = [];
    const seen: LiveEventPayload[] = [];
    const coordinator = createFallbackCoordinator();

    const cleanup = subscribeToLiveEvents(
      (payload) => seen.push(payload),
      (url) => {
        created.push(url);
        return source;
      },
      { coordinator },
    );

    source.emit({ id: "evt-1", event: "run.completed", run_id: "run-a" });
    source.emit({ id: "evt-2", event: "run.failed", run_id: "run-b" });

    expect(created).toEqual(["/api/v1/attach"]);
    expect(seen.map((p) => p.run_id)).toEqual(["run-a", "run-b"]);

    cleanup();
    coordinator.close();
  });

  test("fallback closes the shared source on the final unsubscribe", () => {
    const source = new FakeEventSource();
    const coordinator = createFallbackCoordinator();
    const cleanup = subscribeToLiveEvents(() => {}, () => source, { coordinator });

    expect(source.closed).toBe(false);
    cleanup();
    expect(source.closed).toBe(true);

    coordinator.close();
  });
});

function createCoordinator(eventSourceFactory: (url: string) => EventSourceLike) {
  return createCrossTabSseCoordinator({
    tabId: "live-events-test",
    channelFactory: () => new FakeBroadcastChannel(),
    eventSourceFactory,
    addVisibilityChangeListener: () => () => {},
    addPagehideListener: () => () => {},
    timing: {
      heartbeatMs: 10,
      leaderStaleMs: 50,
      electionJitterMs: 0,
    },
  });
}

function createFallbackCoordinator() {
  return createCrossTabSseCoordinator({
    channelFactory: () => {
      throw new Error("BroadcastChannel unavailable");
    },
  });
}

async function waitFor(condition: () => boolean, timeoutMs = 200) {
  const deadline = Date.now() + timeoutMs;
  while (Date.now() < deadline) {
    if (condition()) return;
    await new Promise((resolve) => setTimeout(resolve, 2));
  }
  throw new Error("condition did not become true before timeout");
}