diff --git a/packages/core-events/src/in-memory-event-bus.test.ts b/packages/core-events/src/in-memory-event-bus.test.ts new file mode 100644 index 0000000..1ce7f4f --- /dev/null +++ b/packages/core-events/src/in-memory-event-bus.test.ts @@ -0,0 +1,49 @@ +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(); + }); +}); diff --git a/packages/core-events/src/in-memory-event-bus.ts b/packages/core-events/src/in-memory-event-bus.ts new file mode 100644 index 0000000..e429ffd --- /dev/null +++ b/packages/core-events/src/in-memory-event-bus.ts @@ -0,0 +1,40 @@ +import type { z } from "zod"; +import type { EventDescriptor } from "./event-descriptor"; +import type { EventHandler, IEventBus } from "./event-bus.interface"; + +export type InMemoryEventBusOptions = { + /** When true, rethrow the first handler error (default: false — errors swallowed). */ + failFast?: boolean; +}; + +export class InMemoryEventBus implements IEventBus { + private readonly handlers = new Map[]>(); + + constructor(private readonly options: InMemoryEventBusOptions = {}) {} + + async publish( + descriptor: EventDescriptor>, + payload: T, + ): Promise { + descriptor.schema.parse(payload); + const subscribers = this.handlers.get(descriptor.name) ?? []; + if (subscribers.length === 0) return; + const settled = await Promise.allSettled( + subscribers.map((h) => h(payload)), + ); + if (this.options.failFast) { + const failure = settled.find((s) => s.status === "rejected"); + if (failure && failure.status === "rejected") throw failure.reason; + } + } + + subscribe( + descriptor: EventDescriptor>, + _consumerFeature: string, + handler: EventHandler, + ): void { + const arr = this.handlers.get(descriptor.name) ?? []; + arr.push(handler as EventHandler); + this.handlers.set(descriptor.name, arr); + } +}