fusion/packages/dashboard/app/__tests__/sse-bus.test.ts

import { describe, it, expect, afterEach, vi, beforeEach } from "vitest";
import { MockEventSource } from "../../vitest.setup";
import { subscribeSse, __resetSseBus, __sseBusChannelCount } from "../sse-bus";
import { clearTraces, getTraces } from "../utils/dashboardTraceBuffer";

function expectEventsUrl(url: string, projectId?: string): void {
  const parsed = new URL(url, "http://localhost");
  expect(parsed.pathname).toBe("/api/events");
  expect(parsed.searchParams.get("projectId")).toBe(projectId ?? null);
  expect(parsed.searchParams.get("clientId")).toBeTruthy();
}

beforeEach(() => {
  window.sessionStorage.clear();
  clearTraces();
});

afterEach(() => {
  __resetSseBus();
});

describe("sse-bus", () => {
  it("opens one EventSource per URL regardless of subscriber count", () => {
    const url = "/api/events?projectId=p1";
    const unsubA = subscribeSse(url, { events: { "task:created": () => {} } });
    const unsubB = subscribeSse(url, { events: { "task:updated": () => {} } });
    const unsubC = subscribeSse(url, { events: { "task:deleted": () => {} } });

    const sources = MockEventSource.instances.filter((es) => {
      expectEventsUrl(es.url, "p1");
      return true;
    });
    expect(sources).toHaveLength(1);

    unsubA();
    unsubB();
    unsubC();
  });

  it("opens separate EventSources for different URLs", () => {
    const unsubA = subscribeSse("/api/events", {});
    const unsubB = subscribeSse("/api/events?projectId=p1", {});
    expect(MockEventSource.instances).toHaveLength(2);
    expectEventsUrl(MockEventSource.instances[0]!.url);
    expectEventsUrl(MockEventSource.instances[1]!.url, "p1");
    unsubA();
    unsubB();
  });

  it("dispatches events to every subscriber of the same URL", () => {
    const url = "/api/events";
    const receivedA: unknown[] = [];
    const receivedB: unknown[] = [];
    const unsubA = subscribeSse(url, {
      events: { "task:created": (e) => receivedA.push(JSON.parse(e.data)) },
    });
    const unsubB = subscribeSse(url, {
      events: { "task:created": (e) => receivedB.push(JSON.parse(e.data)) },
    });

    const es = MockEventSource.instances[0];
    es._emit("task:created", { id: "t-1" });

    expect(receivedA).toEqual([{ id: "t-1" }]);
    expect(receivedB).toEqual([{ id: "t-1" }]);

    unsubA();
    unsubB();
  });

  it("closes the EventSource when the last subscriber unsubscribes", () => {
    const url = "/api/events";
    const unsubA = subscribeSse(url, {});
    const unsubB = subscribeSse(url, {});
    expect(__sseBusChannelCount()).toBe(1);

    const es = MockEventSource.instances[0];
    unsubA();
    expect(es.close).not.toHaveBeenCalled();

    unsubB();
    expect(es.close).toHaveBeenCalledTimes(1);
    expect(__sseBusChannelCount()).toBe(0);
  });

  it("stops dispatching to a subscriber after it unsubscribes", () => {
    const url = "/api/events";
    const received: unknown[] = [];
    const unsub = subscribeSse(url, {
      events: { "task:created": (e) => received.push(JSON.parse(e.data)) },
    });

    const es = MockEventSource.instances[0];
    es._emit("task:created", { id: "t-1" });
    expect(received).toEqual([{ id: "t-1" }]);

    unsub();
    // Another subscriber keeps the channel alive so we can assert the old handler is gone.
    const unsub2 = subscribeSse(url, {});
    es._emit("task:created", { id: "t-2" });
    expect(received).toEqual([{ id: "t-1" }]);
    unsub2();
  });

  it("attaches native listeners lazily as new event types are subscribed", () => {
    const url = "/api/events";
    const unsubA = subscribeSse(url, { events: { "task:created": () => {} } });
    const es = MockEventSource.instances[0];
    expect(Object.keys(es.listeners)).toEqual(
      expect.arrayContaining(["task:created"])
    );
    expect(es.listeners["task:updated"]).toBeUndefined();

    const unsubB = subscribeSse(url, { events: { "task:updated": () => {} } });
    expect(es.listeners["task:updated"]).toBeDefined();

    unsubA();
    unsubB();
  });

  it("fires onOpen for every subscriber on initial connect", () => {
    const url = "/api/events";
    let opensA = 0;
    let opensB = 0;
    const unsubA = subscribeSse(url, { onOpen: () => opensA++ });
    const unsubB = subscribeSse(url, { onOpen: () => opensB++ });

    const es = MockEventSource.instances[0];
    es._emit("open");

    expect(opensA).toBe(1);
    expect(opensB).toBe(1);

    unsubA();
    unsubB();
  });

  it("fires onReconnect whenever the channel is rebuilt", () => {
    const url = "/api/events";
    let reconnects = 0;
    const unsub = subscribeSse(url, {
      onReconnect: () => reconnects++,
    });
    const es = MockEventSource.instances[0];
    // An error that tears down the connection triggers a resync signal.
    es._emit("error");
    expect(reconnects).toBe(1);
    unsub();
  });

  it("does not set a reconnect timer after closeChannel is called", () => {
    vi.useFakeTimers();
    const url = "/api/events";
    const unsub = subscribeSse(url, { events: { "task:created": () => {} } });
    const es = MockEventSource.instances[0];

    // Simulate an error which triggers forceReconnect (and schedules reconnect timer)
    es._emit("error");

    // Immediately unsubscribe (triggers closeChannel which sets closed=true)
    unsub();

    // Advance timers past RECONNECT_DELAY_MS (3 seconds)
    vi.advanceTimersByTime(4_000);

    // No new EventSource should be created — the closed flag prevented reconnect
    expect(MockEventSource.instances).toHaveLength(1);
    vi.useRealTimers();
  });

  it("does not leak channels on rapid subscribe/unsubscribe cycles", () => {
    const url = "/api/events";
    for (let i = 0; i < 5; i++) {
      const unsub = subscribeSse(url, { events: { "task:created": () => {} } });
      unsub();
    }
    expect(__sseBusChannelCount()).toBe(0);
    const countBeforeTimers = MockEventSource.instances.length;
    vi.useFakeTimers();
    vi.advanceTimersByTime(4_000);
    vi.useRealTimers();
    expect(MockEventSource.instances.length).toBe(countBeforeTimers);
  });

  it("reopens subscribed channel on pageshow even when event.persisted is false", () => {
    subscribeSse("/api/events?projectId=p1", {});
    expect(MockEventSource.instances).toHaveLength(1);

    window.dispatchEvent(new Event("pagehide"));
    window.dispatchEvent(new PageTransitionEvent("pageshow", { persisted: false }));

    expect(MockEventSource.instances).toHaveLength(2);
  });

  it("does not reopen on pageshow when there are no subscribers", () => {
    const unsub = subscribeSse("/api/events", {});
    unsub();
    const count = MockEventSource.instances.length;

    window.dispatchEvent(new PageTransitionEvent("pageshow", { persisted: false }));

    expect(MockEventSource.instances).toHaveLength(count);
  });

  it("reopens null channels on visibilitychange visible", () => {
    subscribeSse("/api/events", {});
    window.dispatchEvent(new Event("pagehide"));

    Object.defineProperty(document, "visibilityState", { value: "visible", configurable: true });
    document.dispatchEvent(new Event("visibilitychange"));

    expect(MockEventSource.instances).toHaveLength(2);
  });

  it("reopens CLOSED channels on visibilitychange visible", () => {
    subscribeSse("/api/events", {});
    const first = MockEventSource.instances[0]!;
    first.readyState = MockEventSource.CLOSED;

    Object.defineProperty(document, "visibilityState", { value: "visible", configurable: true });
    document.dispatchEvent(new Event("visibilitychange"));

    expect(MockEventSource.instances).toHaveLength(2);
  });

  it("does not reopen on visibilitychange hidden", () => {
    subscribeSse("/api/events", {});
    const count = MockEventSource.instances.length;

    Object.defineProperty(document, "visibilityState", { value: "hidden", configurable: true });
    document.dispatchEvent(new Event("visibilitychange"));

    expect(MockEventSource.instances).toHaveLength(count);
  });

  it("pushes traces for pageshow, visibilitychange, and forceReconnect", () => {
    subscribeSse("/api/events", {});
    const first = MockEventSource.instances[0]!;

    window.dispatchEvent(new PageTransitionEvent("pageshow", { persisted: true }));
    Object.defineProperty(document, "visibilityState", { value: "visible", configurable: true });
    document.dispatchEvent(new Event("visibilitychange"));
    first._emit("error");

    const events = getTraces().map((entry) => entry.event);
    expect(events).toContain("pageshow");
    expect(events).toContain("visibilitychange");
    expect(events).toContain("forceReconnect");
  });
});