agentic-dev/packages/core-events/src/in-memory-event-bus.test.ts

import { describe, it, expect, vi } from "vitest";
import { z } from "zod";
import { defineEvent } from "@/event-descriptor";
import { InMemoryEventBus } from "@/in-memory-event-bus";

const evt = defineEvent("test.thing", z.object({ id: z.string() }).strict());

describe("InMemoryEventBus", () => {
  it("validates the payload via the descriptor's schema before fanout", async () => {
    const bus = new InMemoryEventBus();
    const handler = vi.fn();
    bus.subscribe(evt, "test-consumer", handler);
    await expect(bus.publish(evt, { id: 123 } as unknown as { id: string })).rejects.toThrow();
    expect(handler).not.toHaveBeenCalled();
  });

  it("delivers to all registered handlers in parallel", async () => {
    const bus = new InMemoryEventBus();
    const a = vi.fn();
    const b = vi.fn();
    bus.subscribe(evt, "consumer-a", a);
    bus.subscribe(evt, "consumer-b", b);
    await bus.publish(evt, { id: "x" });
    expect(a).toHaveBeenCalledWith({ id: "x" });
    expect(b).toHaveBeenCalledWith({ id: "x" });
  });

  it("swallows handler errors by default (publisher's publish does not throw)", async () => {
    const bus = new InMemoryEventBus();
    bus.subscribe(evt, "boom", async () => {
      throw new Error("subscriber blew up");
    });
    await expect(bus.publish(evt, { id: "x" })).resolves.toBeUndefined();
  });

  it("rethrows the first handler error when failFast is true", async () => {
    const bus = new InMemoryEventBus({ failFast: true });
    bus.subscribe(evt, "first", async () => {
      throw new Error("first failure");
    });
    bus.subscribe(evt, "second", vi.fn());
    await expect(bus.publish(evt, { id: "x" })).rejects.toThrow("first failure");
  });

  it("delivers nothing when no handlers are registered", async () => {
    const bus = new InMemoryEventBus();
    await expect(bus.publish(evt, { id: "x" })).resolves.toBeUndefined();
  });
});