stronger coupling to standard schema

Author: dillonstreator

README.md

@@ -130,7 +130,7 @@ const client = new pg.Client({
 await client.connect();
 
 const processor = new EventProcessor({
-  client: createProcessorClient({ querier: client }),
+  client: createProcessorClient({ querier: client, eventSchemas }),
   handlerMap: {
     UserCreated: {
       // Handlers are processed concurrently and independently with retries
@@ -354,18 +354,20 @@ const incrementCounter: TxOBEventHandler = async (event) => {
 
 #### Typed payloads with Standard Schema (Zod/ArkType/Valibot/etc.)
 
-txob supports schema-driven event payload typing through the Standard Schema interface.
+txob uses schema-driven event payload typing through the Standard Schema interface.
+`eventSchemas` is required by both `createProcessorClient(...)` and `createEventProcessor(...)`.
+Use any validator that implements Standard Schema (for example Zod, ArkType, or Valibot).
 
 ```typescript
 import { z } from "zod";
 import {
-  defineTxOBEventHandlerMap,
-  defineTxOBEventSchemas,
-  EventProcessor,
-  type TxOBEventDataMapFromSchemas,
+  createEventHandlerMap,
+  createEventProcessor,
+  type TxOBEventSchemaMap,
 } from "txob";
+import { createProcessorClient } from "txob/pg";
 
-const eventSchemas = defineTxOBEventSchemas({
+const eventSchemas = {
   UserCreated: z.object({
     userId: z.string().uuid(),
     email: z.string().email(),
@@ -374,31 +376,32 @@ const eventSchemas = defineTxOBEventSchemas({
     orderId: z.string().uuid(),
     amount: z.number().positive(),
   }),
-});
+} satisfies TxOBEventSchemaMap<"UserCreated" | "OrderPlaced">;
 
-type EventType = keyof typeof eventSchemas;
-type EventDataMap = TxOBEventDataMapFromSchemas<typeof eventSchemas>;
-
-const handlers = defineTxOBEventHandlerMap(eventSchemas, {
-  UserCreated: {
-    sendWelcomeEmail: async (event) => {
-      await emailService.send(event.data.email); // typed as string
+const handlerMap = createEventHandlerMap({
+  eventSchemas,
+  handlerMap: {
+    UserCreated: {
+      sendWelcomeEmail: async (event) => {
+        await emailService.send(event.data.email); // typed as string
+      },
     },
-  },
-  OrderPlaced: {
-    sendReceipt: async (event) => {
-      await receipts.send(event.data.orderId, event.data.amount); // strongly typed
+    OrderPlaced: {
+      sendReceipt: async (event) => {
+        await receipts.send(event.data.orderId, event.data.amount); // strongly typed
+      },
     },
   },
 });
 
-const processor = new EventProcessor<EventType, EventDataMap>({
-  client: createProcessorClient<EventType, EventDataMap>({ querier: client }),
-  handlerMap: handlers,
+const processor = createEventProcessor({
+  eventSchemas,
+  client: createProcessorClient({ querier: client, eventSchemas }),
+  handlerMap,
 });
 ```
 
-`defineTxOBEventSchemas(...)` is optional. It is a convenience helper for cleaner type inference. You can also define schema maps without it using explicit types.
+Schema-first inference via `eventSchemas` is the standard and required approach.
 
 ### Handler Results
 
@@ -843,7 +846,7 @@ await client.connect();
 
 // 3. Create and start the processor
 const processor = new EventProcessor<EventType>({
-  client: createProcessorClient<EventType>({ querier: client }),
+  client: createProcessorClient({ querier: client, eventSchemas }),
   handlerMap: {
     UserCreated: {
       sendEmail: async (event, { signal }) => {
@@ -963,7 +966,7 @@ gracefulShutdown(server, {
 
 ```typescript
 const processor = new EventProcessor({
-  client: createProcessorClient({ querier: client }),
+  client: createProcessorClient({ querier: client, eventSchemas }),
   handlerMap: {
     UserCreated: {
       sendWelcomeEmail: async (event) => {
@@ -1061,7 +1064,7 @@ const producer = kafka.producer();
 await producer.connect();
 
 const processor = new EventProcessor({
-  client: createProcessorClient({ querier: client }),
+  client: createProcessorClient({ querier: client, eventSchemas }),
   handlerMap: {
     UserCreated: {
       // Publish to Kafka with guaranteed consistency
@@ -1118,7 +1121,7 @@ const client = new pg.Client({
 await client.connect();
 
 const processor = new EventProcessor({
-  client: createProcessorClient({ querier: client }),
+  client: createProcessorClient({ querier: client, eventSchemas }),
   handlerMap: {
     // All your handlers...
   },
@@ -1191,11 +1194,12 @@ Creates a PostgreSQL processor client.
 ```typescript
 import { createProcessorClient } from "txob/pg";
 
-createProcessorClient<EventType>(opts: {
+createProcessorClient(opts: {
   querier: pg.Client;
+  eventSchemas: Record<string, StandardSchemaV1<unknown, unknown>>;
   table?: string;    // Default: "events"
   limit?: number;   // Default: 100
-}): TxOBProcessorClient<EventType>
+}): TxOBProcessorClient<...inferred from eventSchemas...>
 ```
 
 ### `createProcessorClient` (MongoDB)
@@ -1205,12 +1209,13 @@ Creates a MongoDB processor client.
 ```typescript
 import { createProcessorClient } from "txob/mongodb";
 
-createProcessorClient<EventType>(opts: {
+createProcessorClient(opts: {
   mongo: mongodb.MongoClient;
   db: string;               // Database name
+  eventSchemas: Record<string, StandardSchemaV1<unknown, unknown>>;
   collection?: string;      // Default: "events"
   limit?: number;           // Default: 100
-}): TxOBProcessorClient<EventType>
+}): TxOBProcessorClient<...inferred from eventSchemas...>
 ```
 
 ### `TxOBError`
@@ -1317,10 +1322,12 @@ type TxOBEventHandlerMap<
   [TType in EventType]: Record<string, TxOBEventHandler<TType, EventDataMap[TType]>>;
 };
 
-// Standard Schema helpers
-type TxOBEventDataMapFromSchemas<TSchemas> = {
-  [TType in keyof TSchemas & string]: /* schema output for TType */;
-};
+// Schema-first convenience API
+createEventProcessor({
+  eventSchemas,
+  client: createProcessorClient({ querier, eventSchemas }),
+  handlerMap,
+});
 
 // Handler result tracking
 type TxOBEventHandlerResult = {
@@ -1460,7 +1467,7 @@ If using `FOR UPDATE SKIP LOCKED` properly (which txob does), stuck events are n
 - Lower `maxEventConcurrency`
 - Profile handlers for memory leaks
 - Archive old events
-- Reduce `limit` in `createProcessorClient({ querier: client, table, limit })`
+- Reduce `limit` in `createProcessorClient({ querier: client, eventSchemas, table, limit })`
 
 ### Duplicate handler executions
 
@@ -1765,7 +1772,7 @@ txob can emit spans and metrics without depending on a specific telemetry SDK. I
 import { metrics, trace } from "@opentelemetry/api";
 
 const processor = new EventProcessor({
-  client: createProcessorClient({ querier: client }),
+  client: createProcessorClient({ querier: client, eventSchemas }),
   handlerMap: handlers,
   telemetry: {
     tracer: trace.getTracer("txob"),
@@ -1837,35 +1844,37 @@ Yes! txob is written in TypeScript and provides full type safety, including type
 ```typescript
 import { z } from "zod";
 import {
-  defineTxOBEventHandlerMap,
-  defineTxOBEventSchemas,
-  type TxOBEventDataMapFromSchemas,
+  createEventHandlerMap,
+  createEventProcessor,
+  type TxOBEventSchemaMap,
 } from "txob";
+import { createProcessorClient } from "txob/pg";
 
-const eventSchemas = defineTxOBEventSchemas({
+const eventSchemas = {
   UserCreated: z.object({ userId: z.string().uuid(), email: z.string().email() }),
   OrderPlaced: z.object({ orderId: z.string().uuid(), amount: z.number() }),
-});
-
-type EventType = keyof typeof eventSchemas;
-type EventDataMap = TxOBEventDataMapFromSchemas<typeof eventSchemas>;
+} satisfies TxOBEventSchemaMap<"UserCreated" | "OrderPlaced">;
 
-const handlers = defineTxOBEventHandlerMap(eventSchemas, {
-  UserCreated: {
-    sendWelcomeEmail: async (event) => {
-      event.data.email; // string
+const handlers = createEventHandlerMap({
+  eventSchemas,
+  handlerMap: {
+    UserCreated: {
+      sendWelcomeEmail: async (event) => {
+        event.data.email; // string
+      },
     },
-  },
-  OrderPlaced: {
-    sendReceipt: async (event) => {
-      event.data.amount; // number
+    OrderPlaced: {
+      sendReceipt: async (event) => {
+        event.data.amount; // number
+      },
     },
+    // Missing an event type? TypeScript error
   },
-  // Missing an event type? TypeScript error
 });
 
-const processor = new EventProcessor<EventType, EventDataMap>({
-  client: createProcessorClient<EventType, EventDataMap>({ querier: client }),
+const processor = createEventProcessor({
+  eventSchemas,
+  client: createProcessorClient({ querier: client, eventSchemas }),
   handlerMap: handlers,
 });
 ```

examples/pg/events.ts

@@ -1,25 +1,21 @@
 import { z } from "zod";
-import {
-  defineTxOBEventSchemas,
-  type TxOBEventDataMapFromSchemas,
-} from "../../src/index.js";
+import type { TxOBEventSchemaMap } from "../../src/index.js";
 
 export const eventTypes = {
   ResourceSaved: "ResourceSaved",
   EventMaxErrorsReached: "EventMaxErrorsReached",
 } as const;
 
-export const eventSchemas = defineTxOBEventSchemas({
+export type EventType = keyof typeof eventTypes;
+
+export const eventSchemas = {
   [eventTypes.ResourceSaved]: z.object({
     type: z.literal("activity"),
-    id: z.string().uuid(),
+    id: z.uuid(),
   }),
   [eventTypes.EventMaxErrorsReached]: z.object({
-    failedEventId: z.string().uuid(),
+    failedEventId: z.uuid(),
     failedEventType: z.string(),
-    failedEventCorrelationId: z.string().uuid(),
+    failedEventCorrelationId: z.uuid(),
   }),
-});
-
-export type EventType = keyof typeof eventSchemas;
-export type EventDataMap = TxOBEventDataMapFromSchemas<typeof eventSchemas>;
+} satisfies TxOBEventSchemaMap<EventType>;