update functions to accept arguments as object

Author: dillonstreator

README.md

@@ -129,7 +129,7 @@ const client = new pg.Client({
 });
 await client.connect();
 
-const processor = EventProcessor(createProcessorClient(client), {
+const processor = EventProcessor(createProcessorClient({ querier: client }), {
   UserCreated: {
     // Handlers are processed concurrently and independently with retries
     // If one handler fails, others continue processing
@@ -191,15 +191,19 @@ For better performance, you can set up wakeup signals to reduce polling frequenc
 // PostgreSQL: Use Postgres NOTIFY
 import { createWakeupEmitter } from "txob/pg";
 
-const wakeupEmitter = await createWakeupEmitter(clientConfig, {
+const wakeupEmitter = await createWakeupEmitter({
+  listenClientConfig: clientConfig,
   createTrigger: true,
   querier: client,
 });
 
 // MongoDB: Use Change Streams
 import { createWakeupEmitter } from "txob/mongodb";
 
-const wakeupEmitter = await createWakeupEmitter(mongoClient, "myapp");
+const wakeupEmitter = await createWakeupEmitter({
+  mongo: mongoClient,
+  db: "myapp",
+});
 
 // Use with EventProcessor
 const processor = new EventProcessor({
@@ -492,11 +496,11 @@ const client = new pg.Client({
 });
 await client.connect();
 
-const processorClient = createProcessorClient(
-  client,
-  "events", // Optional: table name (default: "events")
-  100, // Optional: max events per poll (default: 100)
-);
+const processorClient = createProcessorClient({
+  querier: client,
+  table: "events", // Optional: table name (default: "events")
+  limit: 100, // Optional: max events per poll (default: 100)
+});
 ```
 
 **4. (Optional) Set up wakeup signals to reduce polling:**
@@ -506,7 +510,8 @@ import { createWakeupEmitter } from "txob/pg";
 
 // Create a wakeup emitter using Postgres NOTIFY
 // This will automatically create a trigger that sends NOTIFY on INSERT
-const wakeupEmitter = await createWakeupEmitter(clientConfig, {
+const wakeupEmitter = await createWakeupEmitter({
+  listenClientConfig: clientConfig,
   createTrigger: true,
   querier: client,
   table: "events", // Optional: table name (default: "events")
@@ -552,12 +557,12 @@ await eventsCollection.createIndex({ correlation_id: 1 });
 ```typescript
 import { createProcessorClient } from "txob/mongodb";
 
-const processorClient = createProcessorClient(
-  mongoClient,
-  "myapp", // Database name
-  "events", // Optional: collection name (default: "events")
-  100, // Optional: max events per poll (default: 100)
-);
+const processorClient = createProcessorClient({
+  mongo: mongoClient,
+  db: "myapp", // Database name
+  collection: "events", // Optional: collection name (default: "events")
+  limit: 100, // Optional: max events per poll (default: 100)
+});
 ```
 
 **3. (Optional) Set up wakeup signals to reduce polling:**
@@ -567,7 +572,9 @@ import { createWakeupEmitter } from "txob/mongodb";
 
 // Create a wakeup emitter using MongoDB Change Streams
 // Note: Requires a replica set or sharded cluster
-const wakeupEmitter = await createWakeupEmitter(mongoClient, "myapp", {
+const wakeupEmitter = await createWakeupEmitter({
+  mongo: mongoClient,
+  db: "myapp",
   collection: "events", // Optional: collection name (default: "events")
 });
 
@@ -712,7 +719,7 @@ await client.connect();
 
 // 3. Create and start the processor
 const processor = EventProcessor(
-  createProcessorClient<EventType>(client),
+  createProcessorClient<EventType>({ querier: client }),
   {
     UserCreated: {
       sendEmail: async (event, { signal }) => {
@@ -833,7 +840,7 @@ gracefulShutdown(server, {
 ### Multiple Event Types
 
 ```typescript
-const processor = EventProcessor(createProcessorClient(client), {
+const processor = EventProcessor(createProcessorClient({ querier: client }), {
   UserCreated: {
     sendWelcomeEmail: async (event) => {
       /* ... */
@@ -926,7 +933,7 @@ const kafka = new Kafka({ brokers: ["localhost:9092"] });
 const producer = kafka.producer();
 await producer.connect();
 
-const processor = EventProcessor(createProcessorClient(client), {
+const processor = EventProcessor(createProcessorClient({ querier: client }), {
   UserCreated: {
     // Publish to Kafka with guaranteed consistency
     publishToKafka: async (event) => {
@@ -980,7 +987,7 @@ const client = new pg.Client({
 });
 await client.connect();
 
-const processor = EventProcessor(createProcessorClient(client), {
+const processor = EventProcessor(createProcessorClient({ querier: client }), {
   // All your handlers...
 });
 
@@ -1050,11 +1057,11 @@ Creates a PostgreSQL processor client.
 ```typescript
 import { createProcessorClient } from "txob/pg";
 
-createProcessorClient<EventType>(
-  client: pg.Client,
-  tableName?: string,    // Default: "events"
-  limit?: number         // Default: 100
-): TxOBProcessorClient<EventType>
+createProcessorClient<EventType>(opts: {
+  querier: pg.Client;
+  table?: string;    // Default: "events"
+  limit?: number;   // Default: 100
+}): TxOBProcessorClient<EventType>
 ```
 
 ### `createProcessorClient` (MongoDB)
@@ -1064,12 +1071,12 @@ Creates a MongoDB processor client.
 ```typescript
 import { createProcessorClient } from "txob/mongodb";
 
-createProcessorClient<EventType>(
-  mongo: mongodb.MongoClient,
-  db: string,               // Database name
-  collectionName?: string,   // Default: "events"
-  limit?: number            // Default: 100
-): TxOBProcessorClient<EventType>
+createProcessorClient<EventType>(opts: {
+  mongo: mongodb.MongoClient;
+  db: string;               // Database name
+  collection?: string;      // Default: "events"
+  limit?: number;           // Default: 100
+}): TxOBProcessorClient<EventType>
 ```
 
 ### `TxOBError`
@@ -1110,7 +1117,8 @@ Creates a Postgres NOTIFY-based wakeup emitter to reduce polling frequency.
 ```typescript
 import { createWakeupEmitter } from "txob/pg";
 
-const wakeupEmitter = await createWakeupEmitter(clientConfig, {
+const wakeupEmitter = await createWakeupEmitter({
+  listenClientConfig: clientConfig,
   createTrigger: true, // Automatically create database trigger
   querier: client, // Required if createTrigger is true
   table: "events", // Optional: table name (default: "events")
@@ -1127,7 +1135,9 @@ Creates a MongoDB Change Stream-based wakeup emitter to reduce polling frequency
 ```typescript
 import { createWakeupEmitter } from "txob/mongodb";
 
-const wakeupEmitter = await createWakeupEmitter(mongoClient, "myapp", {
+const wakeupEmitter = await createWakeupEmitter({
+  mongo: mongoClient,
+  db: "myapp",
   collection: "events", // Optional: collection name (default: "events")
 });
 ```
@@ -1299,7 +1309,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(client, table, limit)`
+- Reduce `limit` in `createProcessorClient({ querier: client, table, limit })`
 
 ### Duplicate handler executions
 
@@ -1578,7 +1588,7 @@ type EventType = keyof typeof eventTypes;
 
 // TypeScript will enforce all event types have handlers
 const processor = EventProcessor<EventType>(
-  createProcessorClient<EventType>(client),
+  createProcessorClient<EventType>({ querier: client }),
   {
     UserCreated: {
       /* handlers */

examples/pg/processor.ts

@@ -28,13 +28,14 @@ let wakeupEmitter: WakeupEmitter | undefined = undefined;
   await client.connect();
   await migrate(client);
 
-  wakeupEmitter = await createWakeupEmitter(clientConfig, {
+  wakeupEmitter = await createWakeupEmitter({
+    listenClientConfig: clientConfig,
     createTrigger: true,
     querier: client,
   });
 
   processor = new EventProcessor<EventType>({
-    client: createProcessorClient<EventType>(client),
+    client: createProcessorClient<EventType>({ querier: client }),
     wakeupEmitter,
     handlerMap: {
       ResourceSaved: {

src/mongodb/client.ts

@@ -22,12 +22,17 @@ const createReadyToProcessFilter = (maxErrors: number) => ({
   errors: { $lt: maxErrors },
 });
 
+export type CreateProcessorClientOpts<EventType extends string> = {
+  mongo: MongoClient;
+  db: string;
+  collection?: string;
+  limit?: number;
+};
+
 export const createProcessorClient = <EventType extends string>(
-  mongo: MongoClient,
-  db: string,
-  collection: string = "events",
-  limit: number = 100,
+  opts: CreateProcessorClientOpts<EventType>,
 ): TxOBProcessorClient<EventType> => {
+  const { mongo, db, collection = "events", limit = 100 } = opts;
   const getEventsToProcess = async (
     opts: TxOBProcessorClientOpts,
   ): Promise<Pick<TxOBEvent<EventType>, "id" | "errors">[]> => {
@@ -146,6 +151,8 @@ export const createProcessorClient = <EventType extends string>(
 };
 
 type CreateWakeupEmitterOpts = {
+  mongo: MongoClient;
+  db: string;
   collection?: string;
 };
 
@@ -163,19 +170,15 @@ type CreateWakeupEmitterOpts = {
  *
  * See: https://www.mongodb.com/docs/manual/changeStreams/
  *
- * @param mongo - MongoDB client instance
- * @param db - Database name
  * @param opts - Options for the wakeup emitter
  * @returns A WakeupEmitter that emits 'wakeup' events when new events are inserted.
  *          Errors (including replica set requirement failures) are emitted via the 'error' event.
  * @throws Does not throw synchronously. Errors are emitted via the 'error' event.
  */
 export const createWakeupEmitter = async (
-  mongo: MongoClient,
-  db: string,
-  opts?: CreateWakeupEmitterOpts,
+  opts: CreateWakeupEmitterOpts,
 ): Promise<WakeupEmitter & { close: () => Promise<void> }> => {
-  const collection = opts?.collection ?? "events";
+  const { mongo, db, collection = "events" } = opts;
   const emitter = new EventEmitter();
 
   // Get the collection to watch

src/pg/client.test.ts

@@ -6,7 +6,7 @@ describe("createProcessorClient", () => {
     const pgClient = {
       query: vi.fn(),
     } as any;
-    const client = createProcessorClient(pgClient);
+    const client = createProcessorClient({ querier: pgClient });
     expect(typeof client.getEventsToProcess).toBe("function");
     expect(typeof client.transaction).toBe("function");
   });
@@ -25,7 +25,7 @@ describe("getEventsToProcess", () => {
     const opts = {
       maxErrors: 10,
     };
-    const client = createProcessorClient(pgClient);
+    const client = createProcessorClient({ querier: pgClient });
     const result = await client.getEventsToProcess(opts);
     expect(pgClient.query).toHaveBeenCalledOnce();
     expect(pgClient.query).toHaveBeenCalledWith(
@@ -41,7 +41,7 @@ describe("transaction", () => {
     const pgClient = {
       query: vi.fn<any>(() => Promise.resolve()),
     } as any;
-    const client = createProcessorClient(pgClient);
+    const client = createProcessorClient({ querier: pgClient });
     await client.transaction(async () => {});
     expect(pgClient.query).toHaveBeenCalledTimes(2);
     expect(pgClient.query).toHaveBeenNthCalledWith(1, "BEGIN");
@@ -51,7 +51,7 @@ describe("transaction", () => {
     const pgClient = {
       query: vi.fn<any>(() => Promise.resolve()),
     } as any;
-    const client = createProcessorClient(pgClient);
+    const client = createProcessorClient({ querier: pgClient });
     await client
       .transaction(async () => {
         throw new Error("error");
@@ -74,7 +74,7 @@ describe("transaction", () => {
         ),
       } as any;
       const eventId = "123";
-      const client = createProcessorClient(pgClient);
+      const client = createProcessorClient({ querier: pgClient });
       let result: any;
       await client.transaction(async (txClient) => {
         result = await txClient.getEventByIdForUpdateSkipLocked(eventId, {
@@ -101,7 +101,7 @@ describe("transaction", () => {
         ),
       } as any;
       const eventId = "123";
-      const client = createProcessorClient(pgClient);
+      const client = createProcessorClient({ querier: pgClient });
       let result: any;
       await client.transaction(async (txClient) => {
         result = await txClient.getEventByIdForUpdateSkipLocked(eventId, {
@@ -141,7 +141,7 @@ describe("transaction", () => {
         },
         correlation_id: "abc123",
       };
-      const client = createProcessorClient(pgClient);
+      const client = createProcessorClient({ querier: pgClient });
       await client.transaction(async (txClient) => {
         await txClient.updateEvent(event);
       });
@@ -176,7 +176,7 @@ describe("transaction", () => {
         handler_results: {},
         errors: 0,
       };
-      const client = createProcessorClient(pgClient);
+      const client = createProcessorClient({ querier: pgClient });
       await client.transaction(async (txClient) => {
         await txClient.createEvent(event);
       });
@@ -219,7 +219,7 @@ describe("transaction", () => {
         }),
       } as any;
 
-      const client = createProcessorClient(pgClient);
+      const client = createProcessorClient({ querier: pgClient });
 
       try {
         await client.transaction(async () => {