update tests

Author: dillonstreator

.DS_Store

@@ -0,0 +1,18 @@
+import { describe, it, expect } from "vitest";
+import { getDate } from "./date.js";
+
+describe("getDate", () => {
+  it("should return a Date object", () => {
+    const result = getDate();
+    expect(result).toBeInstanceOf(Date);
+  });
+
+  it("should return current time", () => {
+    const before = Date.now();
+    const result = getDate();
+    const after = Date.now();
+
+    expect(result.getTime()).toBeGreaterThanOrEqual(before);
+    expect(result.getTime()).toBeLessThanOrEqual(after);
+  });
+});

src/date.test.ts

@@ -159,4 +159,83 @@ describe("transaction", () => {
       );
     });
   });
+
+  describe("createEvent", () => {
+    it("should execute the correct query", async () => {
+      const pgClient = {
+        query: vi.fn<any>(() => Promise.resolve()),
+      } as any;
+      const event = {
+        id: "1",
+        timestamp: new Date(),
+        type: "test_event",
+        data: {
+          thing1: "something",
+        },
+        correlation_id: "abc123",
+        handler_results: {},
+        errors: 0,
+      };
+      const client = createProcessorClient(pgClient);
+      await client.transaction(async (txClient) => {
+        await txClient.createEvent(event);
+      });
+
+      expect(pgClient.query).toHaveBeenCalledTimes(3);
+      expect(pgClient.query).toHaveBeenCalledWith(
+        'INSERT INTO "events" (id, timestamp, type, data, correlation_id, handler_results, errors) VALUES ($1, $2, $3, $4, $5, $6, $7)',
+        [
+          event.id,
+          event.timestamp,
+          event.type,
+          event.data,
+          event.correlation_id,
+          event.handler_results,
+          event.errors,
+        ],
+      );
+    });
+  });
+
+  describe("rollback failure", () => {
+    it("should throw combined error when both transaction and rollback fail", async () => {
+      const transactionError = new Error("transaction failed");
+      const rollbackError = new Error("rollback failed");
+      let queryCount = 0;
+
+      const pgClient = {
+        query: vi.fn<any>((sql: string) => {
+          queryCount++;
+          if (sql === "BEGIN") {
+            // BEGIN succeeds
+            return Promise.resolve();
+          } else if (sql === "ROLLBACK") {
+            // ROLLBACK fails
+            return Promise.reject(rollbackError);
+          } else {
+            // COMMIT or other operations
+            return Promise.resolve();
+          }
+        }),
+      } as any;
+
+      const client = createProcessorClient(pgClient);
+
+      try {
+        await client.transaction(async () => {
+          throw transactionError;
+        });
+        expect.fail("should have thrown an error");
+      } catch (error: any) {
+        expect(error.message).toBe(
+          "Transaction failed: transaction failed (rollback also failed: rollback failed)",
+        );
+        expect(error.cause).toBe(transactionError);
+      }
+
+      expect(pgClient.query).toHaveBeenCalledTimes(2);
+      expect(pgClient.query).toHaveBeenNthCalledWith(1, "BEGIN");
+      expect(pgClient.query).toHaveBeenNthCalledWith(2, "ROLLBACK");
+    });
+  });
 });

src/pg/client.test.ts

@@ -633,6 +633,67 @@ describe("processEvents", () => {
       type: "evtType1",
     });
   });
+
+  it("should throw and log when onEventMaxErrorsReached hook throws an error", async () => {
+    const hookError = new Error("hook error");
+    const logger = {
+      debug: vi.fn(),
+      info: vi.fn(),
+      warn: vi.fn(),
+      error: vi.fn(),
+    };
+    const opts = {
+      maxErrors: 5,
+      backoff: vi.fn(),
+      onEventMaxErrorsReached: vi.fn(() => {
+        throw hookError;
+      }),
+      logger,
+    };
+    const errUnprocessable = new ErrorUnprocessableEventHandler(
+      new Error("err1"),
+    );
+    const handlerMap = {
+      evtType1: {
+        handler1: vi.fn(() => Promise.reject(errUnprocessable)),
+      },
+    };
+    const evt1: TxOBEvent<keyof typeof handlerMap> = {
+      type: "evtType1",
+      id: "1",
+      timestamp: now,
+      data: {},
+      correlation_id: "abc123",
+      handler_results: {},
+      errors: 0,
+    };
+    const events = [evt1];
+    mockClient.getEventsToProcess.mockImplementation(() => events);
+    mockTxClient.getEventByIdForUpdateSkipLocked.mockImplementation((id) => {
+      return events.find((e) => e.id === id);
+    });
+    mockTxClient.updateEvent.mockImplementation(() => {
+      return Promise.resolve();
+    });
+
+    await processEvents(mockClient, handlerMap, opts);
+
+    expect(opts.onEventMaxErrorsReached).toHaveBeenCalledOnce();
+    expect(logger.error).toHaveBeenCalledWith(
+      {
+        eventId: "1",
+        error: hookError,
+      },
+      "error in onEventMaxErrorsReached hook",
+    );
+    expect(logger.error).toHaveBeenCalledWith(
+      {
+        eventId: "1",
+        error: hookError,
+      },
+      "error processing event",
+    );
+  });
 });
 
 describe("defaultBackoff", () => {
@@ -644,6 +705,16 @@ describe("defaultBackoff", () => {
 
     expect(diff).lessThanOrEqual(1);
   });
+
+  it("should cap backoff at maxDelayMs for large error counts", () => {
+    const maxDelayMs = 1000 * 60; // 60 seconds
+    const backoff = defaultBackoff(20); // Large error count that would exceed max
+    const actual = backoff.getTime();
+    const expected = Date.now() + maxDelayMs;
+    const diff = Math.abs(actual - expected);
+
+    expect(diff).lessThanOrEqual(1);
+  });
 });
 
 describe("Processor", () => {
@@ -682,6 +753,91 @@ describe("Processor", () => {
     const diff = Date.now() - start;
     expect(diff).toBeLessThan(50);
   });
+  it("should warn when stopping a processor that is not started", async () => {
+    const logger = {
+      debug: vi.fn(),
+      info: vi.fn(),
+      warn: vi.fn(),
+      error: vi.fn(),
+    };
+    const processor = Processor(() => sleep(1), { sleepTimeMs: 0, logger });
+
+    await processor.stop();
+
+    expect(logger.warn).toHaveBeenCalledWith(
+      "cannot stop processor from 'stopped'",
+    );
+  });
+  it("should handle shutdown when processor completes before stop is called", async () => {
+    let resolve: (() => void) | null = null;
+    const promise = new Promise<void>((r) => {
+      resolve = r;
+    });
+
+    const processor = Processor(
+      () => {
+        return promise;
+      },
+      { sleepTimeMs: 0 },
+    );
+    processor.start();
+
+    // Complete the processor's work
+    // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+    resolve!();
+
+    // Wait a bit for the processor to emit shutdownComplete
+    await sleep(10);
+
+    // Now stop should handle the already-completed case
+    await processor.stop();
+  });
+  it("should warn when starting a processor that is already started", () => {
+    const logger = {
+      debug: vi.fn(),
+      info: vi.fn(),
+      warn: vi.fn(),
+      error: vi.fn(),
+    };
+    const processor = Processor(() => sleep(1000), { sleepTimeMs: 0, logger });
+
+    processor.start();
+    processor.start(); // Try to start again
+
+    expect(logger.warn).toHaveBeenCalledWith(
+      "cannot start processor from 'started'",
+    );
+  });
+  it("should handle non-abort errors and continue processing", async () => {
+    let calls = 0;
+    const logger = {
+      debug: vi.fn(),
+      info: vi.fn(),
+      warn: vi.fn(),
+      error: vi.fn(),
+    };
+    const error = new Error("processing error");
+
+    const processor = Processor(
+      () => {
+        calls++;
+        if (calls === 1) {
+          throw error;
+        }
+        return Promise.resolve();
+      },
+      { sleepTimeMs: 0, logger },
+    );
+    processor.start();
+
+    // Wait for the error to be logged and processing to continue
+    await sleep(1100);
+
+    await processor.stop();
+
+    expect(logger.error).toHaveBeenCalledWith(error);
+    expect(calls).toBeGreaterThan(1); // Should continue processing after error
+  });
 });
 
 describe("EventProcessor", () => {

src/processor.test.ts

@@ -0,0 +1,58 @@
+import { describe, it, expect, vi } from "vitest";
+import { retryable } from "./retry.js";
+
+describe("retryable", () => {
+  it("should retry and succeed on second attempt", async () => {
+    let attempts = 0;
+    const action = vi.fn(async () => {
+      attempts++;
+      if (attempts < 2) {
+        throw new Error("temporary error");
+      }
+      return "success";
+    });
+
+    const result = await retryable(action, {
+      retries: 3,
+      minTimeout: 1,
+      maxTimeout: 10,
+      factor: 1,
+    });
+
+    expect(result).toBe("success");
+    expect(action).toHaveBeenCalledTimes(2);
+  });
+
+  it("should exhaust retries and reject", async () => {
+    const error = new Error("persistent error");
+    const action = vi.fn(async () => {
+      throw error;
+    });
+
+    await expect(
+      retryable(action, {
+        retries: 2,
+        minTimeout: 1,
+        maxTimeout: 10,
+        factor: 1,
+      }),
+    ).rejects.toThrow("persistent error");
+
+    expect(action).toHaveBeenCalledTimes(3); // initial + 2 retries
+  });
+
+  it("should succeed on first attempt", async () => {
+    const action = vi.fn(async () => {
+      return "success";
+    });
+
+    const result = await retryable(action, {
+      retries: 3,
+      minTimeout: 1,
+      maxTimeout: 10,
+    });
+
+    expect(result).toBe("success");
+    expect(action).toHaveBeenCalledTimes(1);
+  });
+});