Kafka consumer schema input support (#310)

* Adding to support for zod inpit on Kafka consumer

* Release prepare

* lint fix

Author: CarlosGamero

packages/kafka/lib/handler-container/KafkaHandlerConfig.ts

@@ -1,6 +1,6 @@
 import type { CommonLogger } from '@lokalise/node-core'
 import type { Message } from '@platformatic/kafka'
-import type { ZodSchema } from 'zod'
+import type { ZodSchema, ZodTypeDef } from 'zod'
 
 export interface RequestContext {
   logger: CommonLogger
@@ -14,11 +14,11 @@ export type KafkaHandler<MessageValue extends object, ExecutionContext> = (
 ) => Promise<void> | void
 
 export class KafkaHandlerConfig<MessageValue extends object, ExecutionContext> {
-  public readonly schema: ZodSchema<MessageValue>
+  public readonly schema: ZodSchema<MessageValue, ZodTypeDef, unknown>
   public readonly handler: KafkaHandler<MessageValue, ExecutionContext>
 
   constructor(
-    schema: ZodSchema<MessageValue>,
+    schema: ZodSchema<MessageValue, ZodTypeDef, unknown>,
     handler: KafkaHandler<MessageValue, ExecutionContext>,
   ) {
     this.schema = schema

packages/kafka/lib/handler-container/KafkaHandlerContainer.spec.ts

@@ -4,7 +4,10 @@ import { KafkaHandlerConfig } from './KafkaHandlerConfig.ts'
 import { KafkaHandlerContainer } from './KafkaHandlerContainer.ts'
 import type { KafkaHandlerRouting } from './KafkaHandlerRoutingBuilder.ts'
 
-const CREATE_SCHEMA = z.object({ type: z.literal('create') })
+const CREATE_SCHEMA = z.object({
+  type: z.literal('create'),
+  prop: z.string().transform((val) => Number(val)),
+})
 const UPDATE_SCHEMA = z.object({ type: z.literal('update') })
 const EMPTY_SCHEMA = z.object({})
 
@@ -69,8 +72,12 @@ describe('KafkaHandlerContainer', () => {
       expect(container.resolveHandler('all', { type: 'non-existing' })?.schema).toBe(EMPTY_SCHEMA)
       expect(container.resolveHandler('all', {})?.schema).toBe(EMPTY_SCHEMA)
 
-      expect(container.resolveHandler('create', { type: 'create' })?.schema).toBe(CREATE_SCHEMA)
-      expect(container.resolveHandler('create', { type: 'update' as any })?.schema).toBe(undefined)
+      expect(container.resolveHandler('create', { type: 'create', prop: 1 })?.schema).toBe(
+        CREATE_SCHEMA,
+      )
+      expect(container.resolveHandler('create', { type: 'update' as any, prop: 1 })?.schema).toBe(
+        undefined,
+      )
       expect(container.resolveHandler('create', {} as any)?.schema).toBe(undefined)
 
       expect(container.resolveHandler('empty', {} as any)?.schema).toBe(EMPTY_SCHEMA)
@@ -88,8 +95,10 @@ describe('KafkaHandlerContainer', () => {
       const container = new KafkaHandlerContainer(topicHandlers)
 
       // Then
-      expect(container.resolveHandler('create', { type: 'create' })?.schema).toBe(CREATE_SCHEMA)
-      expect(container.resolveHandler('create', { type: 'update' as any })?.schema).toBe(
+      expect(container.resolveHandler('create', { type: 'create', prop: 1 })?.schema).toBe(
+        CREATE_SCHEMA,
+      )
+      expect(container.resolveHandler('create', { type: 'update' as any, prop: 1 })?.schema).toBe(
         CREATE_SCHEMA,
       )
       expect(container.resolveHandler('create', {} as any)?.schema).toBe(CREATE_SCHEMA)

packages/kafka/package.json

@@ -1,6 +1,6 @@
 {
     "name": "@message-queue-toolkit/kafka",
-    "version": "0.2.2",
+    "version": "0.2.3",
     "engines": {
         "node": ">= 22.14.0"
     },