diff --git a/packages/core-events/src/payload-jobs-event-bus.test.ts b/packages/core-events/src/payload-jobs-event-bus.test.ts new file mode 100644 index 0000000..87b5a06 --- /dev/null +++ b/packages/core-events/src/payload-jobs-event-bus.test.ts @@ -0,0 +1,51 @@ +import { describe, it, expect, vi } from "vitest"; +import { z } from "zod"; +import { defineEvent } from "@/event-descriptor"; +import { PayloadJobsEventBus } from "@/payload-jobs-event-bus"; +import type { IJobQueue } from "@repo/core-shared/jobs"; + +const evt = defineEvent("auth.user.signed-up", z.object({ userId: z.string() }).strict()); + +function recordingQueue(): IJobQueue & { enqueued: { taskSlug: string; input: unknown }[] } { + const enqueued: { taskSlug: string; input: unknown }[] = []; + const q: IJobQueue = { + async enqueue(taskSlug, input) { + enqueued.push({ taskSlug, input }); + return { jobId: `recording-${enqueued.length}` }; + }, + }; + return Object.assign(q, { enqueued }); +} + +describe("PayloadJobsEventBus", () => { + it("validates the payload before enqueueing", async () => { + const queue = recordingQueue(); + const bus = new PayloadJobsEventBus(queue); + bus.subscribe(evt, "marketing-pages", vi.fn()); + await expect( + bus.publish(evt, { userId: 42 } as unknown as { userId: string }), + ).rejects.toThrow(); + expect(queue.enqueued).toHaveLength(0); + }); + + it("enqueues one task per subscriber, naming `__events..`", async () => { + const queue = recordingQueue(); + const bus = new PayloadJobsEventBus(queue); + bus.subscribe(evt, "marketing-pages", vi.fn()); + bus.subscribe(evt, "blog", vi.fn()); + await bus.publish(evt, { userId: "u1" }); + expect(queue.enqueued).toHaveLength(2); + expect(queue.enqueued.map((e) => e.taskSlug).sort()).toEqual([ + "__events.auth.user.signed-up.blog", + "__events.auth.user.signed-up.marketing-pages", + ]); + expect(queue.enqueued[0]!.input).toEqual({ userId: "u1" }); + }); + + it("enqueues nothing when no subscribers are registered", async () => { + const queue = recordingQueue(); + const bus = new PayloadJobsEventBus(queue); + await bus.publish(evt, { userId: "u1" }); + expect(queue.enqueued).toHaveLength(0); + }); +}); diff --git a/packages/core-events/src/payload-jobs-event-bus.ts b/packages/core-events/src/payload-jobs-event-bus.ts new file mode 100644 index 0000000..20227bb --- /dev/null +++ b/packages/core-events/src/payload-jobs-event-bus.ts @@ -0,0 +1,43 @@ +import type { z } from "zod"; +import type { IJobQueue } from "@repo/core-shared/jobs"; +import type { EventDescriptor } from "./event-descriptor"; +import type { EventHandler, IEventBus } from "./event-bus.interface"; + +/** + * Production-grade bus: for each subscriber, enqueues one Payload task per + * `publish()` call. Subscribers register with their consumer-feature name so + * fan-out tasks are named deterministically: `__events..`. + * The actual handler invocation happens inside Payload's job runner — see the + * matching task config generated by `gen event consume` (Task 39). + */ +export class PayloadJobsEventBus implements IEventBus { + private readonly subscribers = new Map(); + + constructor(private readonly queue: IJobQueue) {} + + async publish( + descriptor: EventDescriptor>, + payload: T, + ): Promise { + descriptor.schema.parse(payload); + const consumers = this.subscribers.get(descriptor.name) ?? []; + await Promise.all( + consumers.map((consumerFeature) => + this.queue.enqueue( + `__events.${descriptor.name}.${consumerFeature}`, + payload, + ), + ), + ); + } + + subscribe( + descriptor: EventDescriptor>, + consumerFeature: string, + _handler: EventHandler, + ): void { + const arr = this.subscribers.get(descriptor.name) ?? []; + if (!arr.includes(consumerFeature)) arr.push(consumerFeature); + this.subscribers.set(descriptor.name, arr); + } +}