Don’t have a default visibility timeout in the queue

Author: matt-aitken

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

@@ -10,12 +10,9 @@ describe("SimpleQueue", () => {
     const queue = new SimpleQueue({
       name: "test-1",
       schema: {
-        test: {
-          schema: z.object({
-            value: z.number(),
-          }),
-          defaultVisibilityTimeoutMs: 2000,
-        },
+        test: z.object({
+          value: z.number(),
+        }),
       },
       redisOptions: {
         host: redisContainer.getHost(),
@@ -26,10 +23,10 @@ describe("SimpleQueue", () => {
     });
 
     try {
-      await queue.enqueue({ id: "1", job: "test", item: { value: 1 } });
+      await queue.enqueue({ id: "1", job: "test", item: { value: 1 }, visibilityTimeoutMs: 2000 });
       expect(await queue.size()).toBe(1);
 
-      await queue.enqueue({ id: "2", job: "test", item: { value: 2 } });
+      await queue.enqueue({ id: "2", job: "test", item: { value: 2 }, visibilityTimeoutMs: 2000 });
       expect(await queue.size()).toBe(2);
 
       const [first] = await queue.dequeue(1);
@@ -64,12 +61,9 @@ describe("SimpleQueue", () => {
     const queue = new SimpleQueue({
       name: "test-1",
       schema: {
-        test: {
-          schema: z.object({
-            value: z.number(),
-          }),
-          defaultVisibilityTimeoutMs: 2000,
-        },
+        test: z.object({
+          value: z.number(),
+        }),
       },
       redisOptions: {
         host: redisContainer.getHost(),
@@ -103,12 +97,9 @@ describe("SimpleQueue", () => {
     const queue = new SimpleQueue({
       name: "test-1",
       schema: {
-        test: {
-          schema: z.object({
-            value: z.number(),
-          }),
-          defaultVisibilityTimeoutMs: 2000,
-        },
+        test: z.object({
+          value: z.number(),
+        }),
       },
       redisOptions: {
         host: redisContainer.getHost(),
@@ -124,6 +115,7 @@ describe("SimpleQueue", () => {
         job: "test",
         item: { value: 1 },
         availableAt: new Date(Date.now() + 50),
+        visibilityTimeoutMs: 2000,
       });
 
       const miss = await queue.dequeue(1);
@@ -147,12 +139,9 @@ describe("SimpleQueue", () => {
     const queue = new SimpleQueue({
       name: "test-1",
       schema: {
-        test: {
-          schema: z.object({
-            value: z.number(),
-          }),
-          defaultVisibilityTimeoutMs: 2000,
-        },
+        test: z.object({
+          value: z.number(),
+        }),
       },
       redisOptions: {
         host: redisContainer.getHost(),
@@ -193,14 +182,11 @@ describe("SimpleQueue", () => {
 
 redisTest("dequeue multiple items", { timeout: 20_000 }, async ({ redisContainer }) => {
   const queue = new SimpleQueue({
-    name: "test-multi",
+    name: "test-1",
     schema: {
-      test: {
-        schema: z.object({
-          value: z.number(),
-        }),
-        defaultVisibilityTimeoutMs: 2000,
-      },
+      test: z.object({
+        value: z.number(),
+      }),
     },
     redisOptions: {
       host: redisContainer.getHost(),
@@ -211,9 +197,9 @@ redisTest("dequeue multiple items", { timeout: 20_000 }, async ({ redisContainer
   });
 
   try {
-    await queue.enqueue({ id: "1", job: "test", item: { value: 1 } });
-    await queue.enqueue({ id: "2", job: "test", item: { value: 2 } });
-    await queue.enqueue({ id: "3", job: "test", item: { value: 3 } });
+    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);
 

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

@@ -4,17 +4,14 @@ import { nanoid } from "nanoid";
 import { z } from "zod";
 
 export interface MessageCatalogSchema {
-  [key: string]: {
-    schema: z.ZodFirstPartySchemaTypes | z.ZodDiscriminatedUnion<any, any>;
-    defaultVisibilityTimeoutMs: number;
-  };
+  [key: string]: z.ZodFirstPartySchemaTypes | z.ZodDiscriminatedUnion<any, any>;
 }
 
 export type MessageCatalogKey<TMessageCatalog extends MessageCatalogSchema> = keyof TMessageCatalog;
 export type MessageCatalogValue<
   TMessageCatalog extends MessageCatalogSchema,
   TKey extends MessageCatalogKey<TMessageCatalog>,
-> = z.infer<TMessageCatalog[TKey]["schema"]>;
+> = z.infer<TMessageCatalog[TKey]>;
 
 export class SimpleQueue<TMessageCatalog extends MessageCatalogSchema> {
   name: string;
@@ -77,16 +74,14 @@ export class SimpleQueue<TMessageCatalog extends MessageCatalogSchema> {
     job: MessageCatalogKey<TMessageCatalog>;
     item: MessageCatalogValue<TMessageCatalog, MessageCatalogKey<TMessageCatalog>>;
     availableAt?: Date;
-    visibilityTimeoutMs?: number;
+    visibilityTimeoutMs: number;
   }): Promise<void> {
     try {
       const score = availableAt ? availableAt.getTime() : Date.now();
-      const jobSchema = this.schema[job];
-      const actualVisibilityTimeoutMs = visibilityTimeoutMs ?? jobSchema.defaultVisibilityTimeoutMs;
       const serializedItem = JSON.stringify({
         job,
         item,
-        visibilityTimeoutMs: actualVisibilityTimeoutMs,
+        visibilityTimeoutMs,
       });
 
       const result = await this.redis.enqueueItem(
@@ -148,7 +143,7 @@ export class SimpleQueue<TMessageCatalog extends MessageCatalogSchema> {
           continue;
         }
 
-        const validatedItem = schema.schema.safeParse(parsedItem.item);
+        const validatedItem = schema.safeParse(parsedItem.item);
 
         if (!validatedItem.success) {
           this.logger.error("Invalid item in queue", {
@@ -160,8 +155,7 @@ export class SimpleQueue<TMessageCatalog extends MessageCatalogSchema> {
           continue;
         }
 
-        const visibilityTimeoutMs =
-          (parsedItem.visibilityTimeoutMs as number) || schema.defaultVisibilityTimeoutMs;
+        const visibilityTimeoutMs = parsedItem.visibilityTimeoutMs as number;
         const invisibleUntil = now + visibilityTimeoutMs;
 
         await this.redis.zadd(`queue`, invisibleUntil, id);