Merge pull request #58 from dillonstreator/decouple-polling-and-processing

Decouple polling and processing

Author: dillonstreator

README.md

@@ -80,7 +80,7 @@ await db.query("COMMIT");
 - ✅ **Configurable error handling** - Exponential backoff, max retries, and custom error hooks
 - ✅ **TypeScript-first** - Full type safety and autocompletion
 - ✅ **Handler result tracking** - Track the execution status of each handler independently
-- ✅ **Minimal dependencies** - Only `p-limit` (plus your database driver)
+- ✅ **Minimal dependencies** - Only `p-limit` and `p-queue` (plus your database driver)
 
 ## Quick Start
 

examples/pg/processor.ts

@@ -7,9 +7,10 @@ import {
 import { createProcessorClient } from "../../src/pg/client.js";
 import { migrate, type EventType, eventTypes } from "./server.js";
 import dotenv from "dotenv";
+import { sleep } from "../../src/sleep.js";
 dotenv.config();
 
-let processor: ReturnType<typeof EventProcessor> | undefined = undefined;
+let processor: EventProcessor<EventType> | undefined = undefined;
 
 (async () => {
   const client = new pg.Client({
@@ -21,9 +22,9 @@ let processor: ReturnType<typeof EventProcessor> | undefined = undefined;
   await client.connect();
   await migrate(client);
 
-  processor = EventProcessor(
-    createProcessorClient<EventType>(client),
-    {
+  processor = new EventProcessor<EventType>({
+    client: createProcessorClient<EventType>(client),
+    handlerMap: {
       ResourceSaved: {
         thing1: async (event) => {
           console.log(`${event.id} thing1 ${event.correlation_id}`);
@@ -40,6 +41,7 @@ let processor: ReturnType<typeof EventProcessor> | undefined = undefined;
           return;
         },
         thing3: async (event) => {
+          await sleep(Math.random() * 10_000);
           console.log(`${event.id} thing3 ${event.correlation_id}`);
           if (Math.random() > 0.75) throw new Error("some issue");
 
@@ -51,41 +53,39 @@ let processor: ReturnType<typeof EventProcessor> | undefined = undefined;
         // For example, you might want to send alerts or log to external systems
       },
     },
-    {
-      sleepTimeMs: 5000,
-      logger: console,
-      onEventMaxErrorsReached: async ({ event, txClient, signal }) => {
-        // Transactionally persist an 'event max errors reached' event
-        // This hook is called when:
-        // - Maximum allowed errors are reached
-        // - An unprocessable error is encountered
-        // - Event handler map is missing for the event type
+    pollingIntervalMs: 5000,
+    logger: console,
+    onEventMaxErrorsReached: async ({ event, txClient, signal }) => {
+      // Transactionally persist an 'event max errors reached' event
+      // This hook is called when:
+      // - Maximum allowed errors are reached
+      // - An unprocessable error is encountered
+      // - Event handler map is missing for the event type
 
-        // Use the abort signal for cleanup during graceful shutdown
-        if (signal?.aborted) {
-          return;
-        }
-
-        await txClient.createEvent({
-          id: randomUUID(),
-          timestamp: new Date(),
-          type: eventTypes.EventMaxErrorsReached,
-          data: {
-            failedEventId: event.id,
-            failedEventType: event.type,
-            failedEventCorrelationId: event.correlation_id,
-          },
-          correlation_id: event.correlation_id,
-          handler_results: {},
-          errors: 0,
-        });
+      // Use the abort signal for cleanup during graceful shutdown
+      if (signal?.aborted) {
+        return;
+      }
 
-        console.log("Event max errors reached event created", {
+      await txClient.createEvent({
+        id: randomUUID(),
+        timestamp: new Date(),
+        type: eventTypes.EventMaxErrorsReached,
+        data: {
           failedEventId: event.id,
-        });
-      },
+          failedEventType: event.type,
+          failedEventCorrelationId: event.correlation_id,
+        },
+        correlation_id: event.correlation_id,
+        handler_results: {},
+        errors: 0,
+      });
+
+      console.log("Event max errors reached event created", {
+        failedEventId: event.id,
+      });
     },
-  );
+  });
   processor.start();
 })();
 

package.json

@@ -58,7 +58,8 @@
     "format": "prettier -w ."
   },
   "dependencies": {
-    "p-limit": "^7.2.0"
+    "p-limit": "^7.2.0",
+    "p-queue": "^9.0.1"
   },
   "peerDependencies": {
     "mongodb": "^7.0.0",

src/mongodb/client.ts

@@ -37,6 +37,7 @@ export const createProcessorClient = <EventType extends string>(
       .find(filter)
       .project({ id: 1, errors: 1 })
       .limit(limit)
+      .sort('timestamp', 'asc')
       .toArray()) as Pick<TxOBEvent<EventType>, "id" | "errors">[];
 
     return events;

src/pg/client.test.ts

@@ -29,7 +29,7 @@ describe("getEventsToProcess", () => {
     const result = await client.getEventsToProcess(opts);
     expect(pgClient.query).toHaveBeenCalledOnce();
     expect(pgClient.query).toHaveBeenCalledWith(
-      'SELECT id, errors FROM "events" WHERE processed_at IS NULL AND (backoff_until IS NULL OR backoff_until < NOW()) AND errors < $1 LIMIT 100',
+      'SELECT id, errors FROM "events" WHERE processed_at IS NULL AND (backoff_until IS NULL OR backoff_until < NOW()) AND errors < $1 ORDER BY timestamp ASC LIMIT 100',
       [opts.maxErrors],
     );
     expect(result).toBe(rows);

src/pg/client.ts

@@ -24,7 +24,7 @@ export const createProcessorClient = <EventType extends string>(
     const events = await querier.query<
       Pick<TxOBEvent<EventType>, "id" | "errors">
     >(
-      `SELECT id, errors FROM ${escapeIdentifier(table)} WHERE processed_at IS NULL AND (backoff_until IS NULL OR backoff_until < NOW()) AND errors < $1 LIMIT ${limit}`,
+      `SELECT id, errors FROM ${escapeIdentifier(table)} WHERE processed_at IS NULL AND (backoff_until IS NULL OR backoff_until < NOW()) AND errors < $1 ORDER BY timestamp ASC LIMIT ${limit}`,
       [opts.maxErrors],
     );
     return events.rows;