fix: resolve consumer message schemas to the schema output type (#472)

CarlosGamero · claude · web-flow · commit 547ea16db64a · 2026-06-12T16:16:29.000+02:00
* fix: resolve consumer message schemas to the schema output type

Consumers receive messages that have already been parsed by the consumer
schema, so ConsumerMessageSchema and AllConsumerMessageSchemas should
resolve to z.output rather than z.input.

For schemas without transforms both types are identical, so this changes
nothing. They diverge once a schema uses transforms or preprocess (e.g. a
field that tolerantly drops unknown enum values): there z.input degrades
to unknown and breaks typing in message handlers, while the handler
actually receives the parsed output.

Publisher-side types intentionally stay z.input, since publishers pass
the raw payload that the schema parses on emit.

Co-Authored-By: Claude Opus 4.8 (1M context) &lt;noreply@anthropic.com&gt;

* Adding type check tests to message type utils

* Fixing one more typing issue

* Adding type tests

---------

Co-authored-by: Claude Opus 4.8 (1M context) &lt;noreply@anthropic.com&gt;
diff --git a/packages/schemas/lib/events/eventTypes.ts b/packages/schemas/lib/events/eventTypes.ts
@@ -32,7 +32,8 @@ export type CommonEventDefinition = {
   tags?: readonly string[] // Free-form tags for the event
 }
 
-export type CommonEventDefinitionConsumerSchemaType<T extends CommonEventDefinition> = z.input<
+// Consumers receive messages already parsed by the consumer schema, hence the output type.
+export type CommonEventDefinitionConsumerSchemaType<T extends CommonEventDefinition> = z.output<
   T['consumerSchema']
 >
 
diff --git a/packages/schemas/lib/events/eventTypes.types.spec.ts b/packages/schemas/lib/events/eventTypes.types.spec.ts
@@ -0,0 +1,75 @@
+import { expectTypeOf } from 'vitest'
+import { z } from 'zod/v4'
+import { enrichMessageSchemaWithBase } from '../messages/baseMessageSchemas.ts'
+import type {
+  AnyEventHandler,
+  CommonEventDefinition,
+  CommonEventDefinitionConsumerSchemaType,
+  CommonEventDefinitionPublisherSchemaType,
+  SingleEventHandler,
+} from './eventTypes.ts'
+
+const myEvents = {
+  plainEvent: {
+    ...enrichMessageSchemaWithBase('entity.created', z.object({ name: z.string() })),
+  },
+  transformingEvent: {
+    ...enrichMessageSchemaWithBase(
+      'entity.updated',
+      z.object({
+        // Forward-compatible field: unknown values are dropped instead of failing validation
+        mode: z.preprocess(
+          (value) => (value === 'live' ? value : undefined),
+          z.literal('live').optional(),
+        ),
+      }),
+    ),
+  },
+} as const satisfies Record<string, CommonEventDefinition>
+
+describe('eventTypes', () => {
+  describe('CommonEventDefinitionConsumerSchemaType', () => {
+    it('resolves transformed fields to their output type, not their input type', () => {
+      type ConsumerMessage = CommonEventDefinitionConsumerSchemaType<
+        typeof myEvents.transformingEvent
+      >
+
+      // DomainEventEmitter hands handlers the event parsed by the schema
+      expectTypeOf<ConsumerMessage['payload']['mode']>().toEqualTypeOf<'live' | undefined>()
+    })
+  })
+
+  describe('CommonEventDefinitionPublisherSchemaType', () => {
+    it('keeps the input type for transformed fields', () => {
+      type PublisherMessage = CommonEventDefinitionPublisherSchemaType<
+        typeof myEvents.transformingEvent
+      >
+
+      // Publishers pass the raw payload that the schema parses on emit
+      expectTypeOf<PublisherMessage['payload']['mode']>().toBeUnknown()
+    })
+  })
+
+  describe('SingleEventHandler', () => {
+    it('receives the parsed event', () => {
+      type Handler = SingleEventHandler<[typeof myEvents.transformingEvent], 'entity.updated'>
+      type HandledEvent = Parameters<Handler['handleEvent']>[0]
+
+      expectTypeOf<HandledEvent['payload']['mode']>().toEqualTypeOf<'live' | undefined>()
+    })
+  })
+
+  describe('AnyEventHandler', () => {
+    it('receives the parsed event union', () => {
+      type Handler = AnyEventHandler<
+        [typeof myEvents.plainEvent, typeof myEvents.transformingEvent]
+      >
+      type HandledEvent = Parameters<Handler['handleEvent']>[0]
+
+      expectTypeOf<HandledEvent['type']>().toEqualTypeOf<'entity.created' | 'entity.updated'>()
+      expectTypeOf<
+        Extract<HandledEvent, { type: 'entity.updated' }>['payload']['mode']
+      >().toEqualTypeOf<'live' | undefined>()
+    })
+  })
+})
diff --git a/packages/schemas/lib/utils/messageTypeUtils.ts b/packages/schemas/lib/utils/messageTypeUtils.ts
@@ -3,9 +3,10 @@ import type { z } from 'zod/v4'
 import type { CommonEventDefinition } from '../events/eventTypes.ts'
 
 /**
- * Resolves schema of a consumer message for a given event definition
+ * Resolves schema of a consumer message for a given event definition.
+ * Consumers receive messages already parsed by the consumer schema, hence the output type.
  */
-export type ConsumerMessageSchema<MessageDefinitionType extends CommonEventDefinition> = z.input<
+export type ConsumerMessageSchema<MessageDefinitionType extends CommonEventDefinition> = z.output<
   MessageDefinitionType['consumerSchema']
 >
 
@@ -23,7 +24,8 @@ export type AllPublisherMessageSchemas<MessageDefinitionTypes extends CommonEven
   z.input<MessageDefinitionTypes[number]['publisherSchema']>
 
 /**
- * Resolves schema of all possible consumer messages for a given list of event definitions
+ * Resolves schema of all possible consumer messages for a given list of event definitions.
+ * Consumers receive messages already parsed by the consumer schema, hence the output type.
  */
 export type AllConsumerMessageSchemas<MessageDefinitionTypes extends CommonEventDefinition[]> =
-  z.input<MessageDefinitionTypes[number]['consumerSchema']>
+  z.output<MessageDefinitionTypes[number]['consumerSchema']>
diff --git a/packages/schemas/lib/utils/messageTypeUtils.types.spec.ts b/packages/schemas/lib/utils/messageTypeUtils.types.spec.ts
@@ -0,0 +1,71 @@
+import { expectTypeOf } from 'vitest'
+import { z } from 'zod/v4'
+import type { CommonEventDefinition } from '../events/eventTypes.ts'
+import { enrichMessageSchemaWithBase } from '../messages/baseMessageSchemas.ts'
+import type {
+  AllConsumerMessageSchemas,
+  ConsumerMessageSchema,
+  PublisherMessageSchema,
+} from './messageTypeUtils.ts'
+
+const myEvents = {
+  plainEvent: {
+    ...enrichMessageSchemaWithBase('entity.created', z.object({ name: z.string() })),
+  },
+  transformingEvent: {
+    ...enrichMessageSchemaWithBase(
+      'entity.updated',
+      z.object({
+        // Forward-compatible field: unknown values are dropped instead of failing validation
+        mode: z.preprocess(
+          (value) => (value === 'live' ? value : undefined),
+          z.literal('live').optional(),
+        ),
+      }),
+    ),
+  },
+} as const satisfies Record<string, CommonEventDefinition>
+
+describe('messageTypeUtils', () => {
+  describe('ConsumerMessageSchema', () => {
+    it('resolves to the parsed message type for transform-free schemas', () => {
+      type ConsumerMessage = ConsumerMessageSchema<typeof myEvents.plainEvent>
+
+      expectTypeOf<ConsumerMessage['type']>().toEqualTypeOf<'entity.created'>()
+      expectTypeOf<ConsumerMessage['payload']['name']>().toEqualTypeOf<string>()
+    })
+
+    it('resolves transformed fields to their output type, not their input type', () => {
+      type ConsumerMessage = ConsumerMessageSchema<typeof myEvents.transformingEvent>
+
+      // Consumers receive messages already parsed by the consumer schema, so the
+      // field is the preprocess output, not unknown (the input of any preprocess)
+      expectTypeOf<ConsumerMessage['payload']['mode']>().toEqualTypeOf<'live' | undefined>()
+    })
+  })
+
+  describe('PublisherMessageSchema', () => {
+    it('resolves to the raw (pre-parse) message type', () => {
+      type PublisherMessage = PublisherMessageSchema<typeof myEvents.plainEvent>
+
+      expectTypeOf<PublisherMessage['payload']['name']>().toEqualTypeOf<string>()
+    })
+
+    it('keeps the input type for transformed fields', () => {
+      type PublisherMessage = PublisherMessageSchema<typeof myEvents.transformingEvent>
+
+      // Publishers pass the raw payload that the schema parses on emit
+      expectTypeOf<PublisherMessage['payload']['mode']>().toBeUnknown()
+    })
+  })
+
+  describe('AllConsumerMessageSchemas', () => {
+    it('resolves to the union of parsed message types', () => {
+      type SupportedMessages = AllConsumerMessageSchemas<
+        [typeof myEvents.plainEvent, typeof myEvents.transformingEvent]
+      >
+
+      expectTypeOf<SupportedMessages['type']>().toEqualTypeOf<'entity.created' | 'entity.updated'>()
+    })
+  })
+})
diff --git a/packages/schemas/vitest.config.ts b/packages/schemas/vitest.config.ts
@@ -7,6 +7,10 @@ export default defineConfig({
     watch: false,
     mockReset: true,
     pool: 'threads',
+    typecheck: {
+      enabled: true,
+      include: ['**/*.types.spec.ts'],
+    },
     coverage: {
       provider: 'v8',
       include: ['lib/**/*.ts'],

Original file line number	Diff line number	Diff line change
`@@ -32,7 +32,8 @@ export type CommonEventDefinition = {`
`32`	`32`	`tags?: readonly string[] // Free-form tags for the event`
`33`	`33`	`}`
`34`	`34`
`35`		`-export type CommonEventDefinitionConsumerSchemaType<T extends CommonEventDefinition> = z.input<`
	`35`	`+// Consumers receive messages already parsed by the consumer schema, hence the output type.`
	`36`	`+export type CommonEventDefinitionConsumerSchemaType<T extends CommonEventDefinition> = z.output<`
`36`	`37`	`T['consumerSchema']`
`37`	`38`	`>`
`38`	`39`