From 84767126204c901b0b437e41402acea26d93ab69 Mon Sep 17 00:00:00 2001 From: Danijel Martinek Date: Fri, 10 Jul 2026 16:30:53 +0200 Subject: [PATCH] fix(generators): event buses deliver the zod-parsed payload MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit InMemoryEventBus.publish and PayloadJobsEventBus.publish called descriptor.schema.parse(payload) but discarded the result, so handlers and enqueued jobs received the raw input — zod defaults, catches, transforms and strips never applied. Both bus templates now fan out the parsed value; regression tests added to both test templates and the events snapshot hashes regenerated (template-tree sha over hbs-stripped paths — only the four touched files differ). Co-Authored-By: Claude Fable 5 --- .../core-package/events.snapshot.json | 8 ++++---- .../events/src/in-memory-event-bus.test.ts.hbs | 15 +++++++++++++++ .../events/src/in-memory-event-bus.ts.hbs | 6 ++---- .../events/src/payload-jobs-event-bus.test.ts.hbs | 15 +++++++++++++++ .../events/src/payload-jobs-event-bus.ts.hbs | 4 ++-- 5 files changed, 38 insertions(+), 10 deletions(-) diff --git a/turbo/generators/__snapshots__/core-package/events.snapshot.json b/turbo/generators/__snapshots__/core-package/events.snapshot.json index 6375f42..145f3be 100644 --- a/turbo/generators/__snapshots__/core-package/events.snapshot.json +++ b/turbo/generators/__snapshots__/core-package/events.snapshot.json @@ -29,11 +29,11 @@ }, { "path": "src/in-memory-event-bus.test.ts", - "sha256": "56f704de153f7676bf243d6350f4bebe28cb9ffec1eb3afef8f5f2bb08561c21" + "sha256": "d75a7694cbf08f6afdc0b5ffe80c37e2c61e00b94f61231200e1c26becb9f26f" }, { "path": "src/in-memory-event-bus.ts", - "sha256": "d7de709049bd864bdcd8f02316d981cf0407efbc5b6f13fd2ab86094ecaddb8a" + "sha256": "c629e0733ab23ef808be7cba04ccd4d31792846dc81d02c3722a210a47aa5035" }, { "path": "src/index.ts", @@ -41,11 +41,11 @@ }, { "path": "src/payload-jobs-event-bus.test.ts", - "sha256": "cda493839ed9af7d48a4c240392933af7167805a94ebaf30efd64f8a374b7441" + "sha256": "dc091425978d4e02881ccdcb2a46b716273d848ebdd87f72131abdf419a40161" }, { "path": "src/payload-jobs-event-bus.ts", - "sha256": "5fbeb611af7f4ba4302828de6a0262a74d9b4bb6a11cbbb8ed40712aae731ded" + "sha256": "2aae98fcc2b3d4c77d7b0f80e1aedb56bbce737e296b9681b663fd22380ae3ab" }, { "path": "src/symbols.ts", diff --git a/turbo/generators/templates/core-package/events/src/in-memory-event-bus.test.ts.hbs b/turbo/generators/templates/core-package/events/src/in-memory-event-bus.test.ts.hbs index 1ce7f4f..2ef2784 100644 --- a/turbo/generators/templates/core-package/events/src/in-memory-event-bus.test.ts.hbs +++ b/turbo/generators/templates/core-package/events/src/in-memory-event-bus.test.ts.hbs @@ -46,4 +46,19 @@ describe("InMemoryEventBus", () => { const bus = new InMemoryEventBus(); await expect(bus.publish(evt, { id: "x" })).resolves.toBeUndefined(); }); + + it("delivers the zod-parsed payload, not the raw input", async () => { + const evtNormalized = defineEvent( + "test.normalized", + z.object({ id: z.string(), mode: z.string().catch("auto") }), + ); + const bus = new InMemoryEventBus(); + const handler = vi.fn(); + bus.subscribe(evtNormalized, "consumer", handler); + await bus.publish(evtNormalized, { + id: "x", + mode: 42 as unknown as string, + }); + expect(handler).toHaveBeenCalledWith({ id: "x", mode: "auto" }); + }); }); diff --git a/turbo/generators/templates/core-package/events/src/in-memory-event-bus.ts.hbs b/turbo/generators/templates/core-package/events/src/in-memory-event-bus.ts.hbs index 6ac36e8..f1741c4 100644 --- a/turbo/generators/templates/core-package/events/src/in-memory-event-bus.ts.hbs +++ b/turbo/generators/templates/core-package/events/src/in-memory-event-bus.ts.hbs @@ -16,12 +16,10 @@ export class InMemoryEventBus implements IEventBus { descriptor: EventDescriptor>, payload: T, ): Promise { - descriptor.schema.parse(payload); + const parsed = 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)), - ); + const settled = await Promise.allSettled(subscribers.map((h) => h(parsed))); if (this.options.failFast) { const failure = settled.find((s) => s.status === "rejected"); // Only the first rejection is rethrown. Other failures are intentionally diff --git a/turbo/generators/templates/core-package/events/src/payload-jobs-event-bus.test.ts.hbs b/turbo/generators/templates/core-package/events/src/payload-jobs-event-bus.test.ts.hbs index 87b5a06..f5cd4bc 100644 --- a/turbo/generators/templates/core-package/events/src/payload-jobs-event-bus.test.ts.hbs +++ b/turbo/generators/templates/core-package/events/src/payload-jobs-event-bus.test.ts.hbs @@ -42,6 +42,21 @@ describe("PayloadJobsEventBus", () => { expect(queue.enqueued[0]!.input).toEqual({ userId: "u1" }); }); + it("enqueues the zod-parsed payload, not the raw input", async () => { + const evtNormalized = defineEvent( + "test.normalized", + z.object({ userId: z.string(), mode: z.string().catch("auto") }), + ); + const queue = recordingQueue(); + const bus = new PayloadJobsEventBus(queue); + bus.subscribe(evtNormalized, "marketing-pages", vi.fn()); + await bus.publish(evtNormalized, { + userId: "u1", + mode: 42 as unknown as string, + }); + expect(queue.enqueued[0]!.input).toEqual({ userId: "u1", mode: "auto" }); + }); + it("enqueues nothing when no subscribers are registered", async () => { const queue = recordingQueue(); const bus = new PayloadJobsEventBus(queue); diff --git a/turbo/generators/templates/core-package/events/src/payload-jobs-event-bus.ts.hbs b/turbo/generators/templates/core-package/events/src/payload-jobs-event-bus.ts.hbs index 20227bb..ff03012 100644 --- a/turbo/generators/templates/core-package/events/src/payload-jobs-event-bus.ts.hbs +++ b/turbo/generators/templates/core-package/events/src/payload-jobs-event-bus.ts.hbs @@ -19,13 +19,13 @@ export class PayloadJobsEventBus implements IEventBus { descriptor: EventDescriptor>, payload: T, ): Promise { - descriptor.schema.parse(payload); + const parsed = 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, + parsed, ), ), );