DLQ tests

Author: matt-aitken

internal-packages/redis-worker/src/queue.test.ts

@@ -185,68 +185,123 @@ describe("SimpleQueue", () => {
       await queue.close();
     }
   });
-});
 
-redisTest("dequeue multiple items", { timeout: 20_000 }, async ({ redisContainer }) => {
-  const queue = new SimpleQueue({
-    name: "test-1",
-    schema: {
-      test: z.object({
-        value: z.number(),
-      }),
-    },
-    redisOptions: {
-      host: redisContainer.getHost(),
-      port: redisContainer.getPort(),
-      password: redisContainer.getPassword(),
-    },
-    logger: new Logger("test", "log"),
+  redisTest("dequeue multiple items", { timeout: 20_000 }, async ({ redisContainer }) => {
+    const queue = new SimpleQueue({
+      name: "test-1",
+      schema: {
+        test: z.object({
+          value: z.number(),
+        }),
+      },
+      redisOptions: {
+        host: redisContainer.getHost(),
+        port: redisContainer.getPort(),
+        password: redisContainer.getPassword(),
+      },
+      logger: new Logger("test", "log"),
+    });
+
+    try {
+      await queue.enqueue({ id: "1", job: "test", item: { value: 1 }, visibilityTimeoutMs: 2000 });
+      await queue.enqueue({ id: "2", job: "test", item: { value: 2 }, visibilityTimeoutMs: 2000 });
+      await queue.enqueue({ id: "3", job: "test", item: { value: 3 }, visibilityTimeoutMs: 2000 });
+
+      expect(await queue.size()).toBe(3);
+
+      const dequeued = await queue.dequeue(2);
+      expect(dequeued).toHaveLength(2);
+      expect(dequeued[0]).toEqual({
+        id: "1",
+        job: "test",
+        item: { value: 1 },
+        visibilityTimeoutMs: 2000,
+        attempt: 0,
+      });
+      expect(dequeued[1]).toEqual({
+        id: "2",
+        job: "test",
+        item: { value: 2 },
+        visibilityTimeoutMs: 2000,
+        attempt: 0,
+      });
+
+      expect(await queue.size()).toBe(1);
+      expect(await queue.size({ includeFuture: true })).toBe(3);
+
+      await queue.ack(dequeued[0].id);
+      await queue.ack(dequeued[1].id);
+
+      expect(await queue.size({ includeFuture: true })).toBe(1);
+
+      const [last] = await queue.dequeue(1);
+      expect(last).toEqual({
+        id: "3",
+        job: "test",
+        item: { value: 3 },
+        visibilityTimeoutMs: 2000,
+        attempt: 0,
+      });
+
+      await queue.ack(last.id);
+      expect(await queue.size({ includeFuture: true })).toBe(0);
+    } finally {
+      await queue.close();
+    }
   });
 
-  try {
-    await queue.enqueue({ id: "1", job: "test", item: { value: 1 }, visibilityTimeoutMs: 2000 });
-    await queue.enqueue({ id: "2", job: "test", item: { value: 2 }, visibilityTimeoutMs: 2000 });
-    await queue.enqueue({ id: "3", job: "test", item: { value: 3 }, visibilityTimeoutMs: 2000 });
-
-    expect(await queue.size()).toBe(3);
-
-    const dequeued = await queue.dequeue(2);
-    expect(dequeued).toHaveLength(2);
-    expect(dequeued[0]).toEqual({
-      id: "1",
-      job: "test",
-      item: { value: 1 },
-      visibilityTimeoutMs: 2000,
-      attempt: 0,
-    });
-    expect(dequeued[1]).toEqual({
-      id: "2",
-      job: "test",
-      item: { value: 2 },
-      visibilityTimeoutMs: 2000,
-      attempt: 0,
+  redisTest("Dead Letter Queue", { timeout: 20_000 }, async ({ redisContainer }) => {
+    const queue = new SimpleQueue({
+      name: "test-dlq",
+      schema: {
+        test: z.object({
+          value: z.number(),
+        }),
+      },
+      redisOptions: {
+        host: redisContainer.getHost(),
+        port: redisContainer.getPort(),
+        password: redisContainer.getPassword(),
+      },
+      logger: new Logger("test", "log"),
     });
 
-    expect(await queue.size()).toBe(1);
-    expect(await queue.size({ includeFuture: true })).toBe(3);
+    try {
+      // Enqueue an item
+      await queue.enqueue({ id: "1", job: "test", item: { value: 1 }, visibilityTimeoutMs: 2000 });
+      expect(await queue.size()).toBe(1);
+      expect(await queue.sizeOfDeadLetterQueue()).toBe(0);
 
-    await queue.ack(dequeued[0].id);
-    await queue.ack(dequeued[1].id);
+      // Move item to DLQ
+      await queue.moveToDeadLetterQueue("1", "Test error message");
+      expect(await queue.size()).toBe(0);
+      expect(await queue.sizeOfDeadLetterQueue()).toBe(1);
 
-    expect(await queue.size({ includeFuture: true })).toBe(1);
+      // Attempt to dequeue from the main queue should return empty
+      const dequeued = await queue.dequeue(1);
+      expect(dequeued).toEqual([]);
 
-    const [last] = await queue.dequeue(1);
-    expect(last).toEqual({
-      id: "3",
-      job: "test",
-      item: { value: 3 },
-      visibilityTimeoutMs: 2000,
-      attempt: 0,
-    });
+      // Redrive item from DLQ
+      await queue.redriveFromDeadLetterQueue("1");
+      expect(await queue.size()).toBe(1);
+      expect(await queue.sizeOfDeadLetterQueue()).toBe(0);
 
-    await queue.ack(last.id);
-    expect(await queue.size({ includeFuture: true })).toBe(0);
-  } finally {
-    await queue.close();
-  }
+      // Dequeue the redriven item
+      const [redrivenItem] = await queue.dequeue(1);
+      expect(redrivenItem).toEqual({
+        id: "1",
+        job: "test",
+        item: { value: 1 },
+        visibilityTimeoutMs: 2000,
+        attempt: 0,
+      });
+
+      // Acknowledge the item
+      await queue.ack(redrivenItem.id);
+      expect(await queue.size()).toBe(0);
+      expect(await queue.sizeOfDeadLetterQueue()).toBe(0);
+    } finally {
+      await queue.close();
+    }
+  });
 });