claude-code-best/packages/remote-control-server/src/__tests__/event-bus.test.ts

import { describe, test, expect, beforeEach } from "bun:test";
import { EventBus, getEventBus, removeEventBus, getAllEventBuses } from "../transport/event-bus";

describe("EventBus", () => {
  let bus: EventBus;

  beforeEach(() => {
    bus = new EventBus();
  });

  describe("publish", () => {
    test("publishes event with seqNum starting at 1", () => {
      const event = bus.publish({
        id: "e1",
        sessionId: "s1",
        type: "user",
        payload: { content: "hello" },
        direction: "outbound",
      });
      expect(event.seqNum).toBe(1);
      expect(event.createdAt).toBeGreaterThan(0);
    });

    test("increments seqNum on each publish", () => {
      bus.publish({ id: "e1", sessionId: "s1", type: "user", payload: {}, direction: "outbound" });
      bus.publish({ id: "e2", sessionId: "s1", type: "assistant", payload: {}, direction: "inbound" });
      const event = bus.publish({ id: "e3", sessionId: "s1", type: "result", payload: {}, direction: "inbound" });
      expect(event.seqNum).toBe(3);
    });

    test("throws when publishing to a closed bus", () => {
      bus.close();
      expect(() =>
        bus.publish({ id: "e1", sessionId: "s1", type: "user", payload: {}, direction: "outbound" }),
      ).toThrow("EventBus is closed");
    });
  });

  describe("subscribe", () => {
    test("receives published events", () => {
      const received: unknown[] = [];
      bus.subscribe((event) => received.push(event));

      bus.publish({ id: "e1", sessionId: "s1", type: "user", payload: { content: "hi" }, direction: "outbound" });
      expect(received).toHaveLength(1);
      expect((received[0] as any).payload).toEqual({ content: "hi" });
    });

    test("unsubscribe stops receiving events", () => {
      const received: unknown[] = [];
      const unsub = bus.subscribe((event) => received.push(event));
      unsub();
      bus.publish({ id: "e1", sessionId: "s1", type: "user", payload: {}, direction: "outbound" });
      expect(received).toHaveLength(0);
    });

    test("multiple subscribers all receive events", () => {
      const r1: unknown[] = [];
      const r2: unknown[] = [];
      bus.subscribe((e) => r1.push(e));
      bus.subscribe((e) => r2.push(e));
      bus.publish({ id: "e1", sessionId: "s1", type: "user", payload: {}, direction: "outbound" });
      expect(r1).toHaveLength(1);
      expect(r2).toHaveLength(1);
    });

    test("subscriber error does not affect other subscribers", () => {
      const received: unknown[] = [];
      bus.subscribe(() => {
        throw new Error("boom");
      });
      bus.subscribe((e) => received.push(e));
      bus.publish({ id: "e1", sessionId: "s1", type: "user", payload: {}, direction: "outbound" });
      expect(received).toHaveLength(1);
    });

    test("subscriberCount", () => {
      expect(bus.subscriberCount()).toBe(0);
      const unsub1 = bus.subscribe(() => {});
      expect(bus.subscriberCount()).toBe(1);
      const unsub2 = bus.subscribe(() => {});
      expect(bus.subscriberCount()).toBe(2);
      unsub1();
      expect(bus.subscriberCount()).toBe(1);
    });
  });

  describe("getEventsSince", () => {
    test("returns events after given seqNum", () => {
      bus.publish({ id: "e1", sessionId: "s1", type: "user", payload: {}, direction: "outbound" });
      bus.publish({ id: "e2", sessionId: "s1", type: "assistant", payload: {}, direction: "inbound" });
      bus.publish({ id: "e3", sessionId: "s1", type: "result", payload: {}, direction: "inbound" });

      const events = bus.getEventsSince(1);
      expect(events).toHaveLength(2);
      expect(events[0].seqNum).toBe(2);
      expect(events[1].seqNum).toBe(3);
    });

    test("returns empty for seqNum beyond last", () => {
      bus.publish({ id: "e1", sessionId: "s1", type: "user", payload: {}, direction: "outbound" });
      expect(bus.getEventsSince(1)).toHaveLength(0);
    });

    test("returns all events when seqNum is 0", () => {
      bus.publish({ id: "e1", sessionId: "s1", type: "user", payload: {}, direction: "outbound" });
      bus.publish({ id: "e2", sessionId: "s1", type: "assistant", payload: {}, direction: "inbound" });
      expect(bus.getEventsSince(0)).toHaveLength(2);
    });
  });

  describe("getLastSeqNum", () => {
    test("returns 0 for empty bus", () => {
      expect(bus.getLastSeqNum()).toBe(0);
    });

    test("returns last seqNum after publishes", () => {
      bus.publish({ id: "e1", sessionId: "s1", type: "user", payload: {}, direction: "outbound" });
      bus.publish({ id: "e2", sessionId: "s1", type: "user", payload: {}, direction: "outbound" });
      expect(bus.getLastSeqNum()).toBe(2);
    });
  });

  describe("close", () => {
    test("clears subscribers and prevents publishing", () => {
      bus.subscribe(() => {});
      bus.close();
      expect(bus.subscriberCount()).toBe(0);
      expect(() => bus.publish({ id: "e1", sessionId: "s1", type: "user", payload: {}, direction: "outbound" })).toThrow();
    });
  });
});

describe("EventBus registry", () => {
  beforeEach(() => {
    // Clean up global registry
    for (const [key] of getAllEventBuses()) {
      removeEventBus(key);
    }
  });

  describe("getEventBus", () => {
    test("creates new bus for unknown session", () => {
      const bus = getEventBus("s1");
      expect(bus).toBeInstanceOf(EventBus);
      expect(getAllEventBuses().has("s1")).toBe(true);
    });

    test("returns same bus for same session", () => {
      const bus1 = getEventBus("s1");
      const bus2 = getEventBus("s1");
      expect(bus1).toBe(bus2);
    });
  });

  describe("removeEventBus", () => {
    test("removes and closes bus", () => {
      const bus = getEventBus("s2");
      removeEventBus("s2");
      expect(getAllEventBuses().has("s2")).toBe(false);
      expect(() => bus.publish({ id: "e1", sessionId: "s2", type: "user", payload: {}, direction: "outbound" })).toThrow();
    });

    test("no-op for non-existent bus", () => {
      expect(() => removeEventBus("nonexistent")).not.toThrow();
    });
  });

  describe("getAllEventBuses", () => {
    test("returns all registered buses", () => {
      getEventBus("a");
      getEventBus("b");
      expect(getAllEventBuses().size).toBeGreaterThanOrEqual(2);
    });
  });
});