feat: implement unified Lua script for atomic job creation with idempotency support in Redis driver

Author: muniter

src/drivers/redis.ts

@@ -14,30 +14,33 @@ export interface RedisJobRequest<TPayload> extends BaseJobOptions, WithPriority,
 export interface RedisClient {
   // Counter operations
   incr(key: string): Promise<number>;
-  
+
   // Hash operations
   hSet(key: string, field: string, value: string): Promise<number>;
   hGet(key: string, field: string): Promise<string | null>;
   hDel(key: string, fields: string[]): Promise<number>;
   hIncrBy(key: string, field: string, increment: number): Promise<number>;
-  
+
   // Sorted set operations
   zAdd(key: string, members: { score: number; value: string }): Promise<number>;
   zRem(key: string, members: string[]): Promise<number>;
   zRangeByScore(
-    key: string, 
-    min: number | string, 
-    max: number | string, 
+    key: string,
+    min: number | string,
+    max: number | string,
     options?: { REV?: boolean; LIMIT?: { offset: number; count: number } }
   ): Promise<string[]>;
   zRange(
-    key: string, 
-    start: number, 
-    stop: number, 
+    key: string,
+    start: number,
+    stop: number,
     options?: { REV?: boolean }
   ): Promise<string[]>;
   zScore(key: string, member: string): Promise<number | null>;
-  
+
+  // Script operations
+  eval(script: string, options: { keys: string[]; arguments: string[] }): Promise<any>;
+
   // General operations
   del(keys: string[]): Promise<number>;
 }
@@ -72,46 +75,86 @@ export class RedisQueue<TJobMap = Record<string, any>> extends Queue<TJobMap, Re
   }
 
   protected async pushMessage(payload: unknown, meta: JobMeta): Promise<string> {
-    // Check for idempotency
-    if (meta.idempotencyKey) {
-      const existingJobId = await this.redis.hGet(this.idempotencyKey, meta.idempotencyKey);
-      if (existingJobId) {
-        // Check if the job still exists
-        const existingMessage = await this.redis.hGet(this.messagesKey, existingJobId);
-        if (existingMessage) {
-          // Job is still waiting, delayed, or reserved - return existing ID
-          return existingJobId;
-        }
-        // Job completed/failed, clean up the idempotency mapping
-        await this.redis.hDel(this.idempotencyKey, [meta.idempotencyKey]);
-      }
-    }
-
-    const id = (await this.redis.incr(this.idKey)).toString();
     const now = Math.floor(Date.now() / 1000);
 
+    // Unified Lua script for atomic job creation (with or without idempotency)
+    const script = `
+      local idempotencyKey = KEYS[1]
+      local messagesKey = KEYS[2]
+      local idKey = KEYS[3]
+      local queueKey = KEYS[4]
+
+      local idempotencyField = ARGV[1]
+      local message = ARGV[2]
+      local queueScore = ARGV[3]
+
+      -- Check if idempotency key is provided (non-empty string)
+      if idempotencyField ~= "" then
+        -- Check if idempotency key exists
+        local existingJobId = redis.call('HGET', idempotencyKey, idempotencyField)
+        if existingJobId then
+          -- Check if the job still exists
+          local existingMessage = redis.call('HGET', messagesKey, existingJobId)
+          if existingMessage then
+            -- Job is still active, return existing ID
+            return existingJobId
+          else
+            -- Job completed/failed, clean up the idempotency mapping
+            redis.call('HDEL', idempotencyKey, idempotencyField)
+          end
+        end
+      end
+
+      -- Create new job
+      local newId = tostring(redis.call('INCR', idKey))
+
+      -- Store job message
+      redis.call('HSET', messagesKey, newId, message)
+
+      -- Store idempotency mapping if key is provided
+      if idempotencyField ~= "" then
+        redis.call('HSET', idempotencyKey, idempotencyField, newId)
+      end
+
+      -- Add to appropriate queue
+      redis.call('ZADD', queueKey, queueScore, newId)
+
+      return newId
+    `;
+
     const message = JSON.stringify({ payload, meta });
-    await this.redis.hSet(this.messagesKey, id, message);
+    const priority = meta.priority || 0;
+    const delaySeconds = meta.delaySeconds || 0;
 
-    // Store idempotency key mapping if provided
-    if (meta.idempotencyKey) {
-      await this.redis.hSet(this.idempotencyKey, meta.idempotencyKey, id);
-    }
+    let queueKey: string;
+    let queueScore: string;
 
-    if (meta.delaySeconds && meta.delaySeconds > 0) {
-      // Add to delayed set with execution time as score
-      const executeAt = now + meta.delaySeconds;
-      await this.redis.zAdd(this.delayedKey, { score: executeAt, value: id });
+    if (delaySeconds > 0) {
+      // Delayed job
+      queueKey = this.delayedKey;
+      queueScore = (now + delaySeconds).toString();
     } else {
-      // Add to waiting queue with priority support
-      // Use sorted set for priority, with higher priority = higher score
-      // For same priority, use job ID for FIFO (lower ID = added earlier = higher priority within same priority)
-      // Score = priority * 1000000000 + (1000000000 - jobId) for FIFO within same priority
-      const priority = meta.priority || 0;
+      // Waiting job with priority
+      queueKey = this.waitingKey;
+      // Note: we can't calculate the exact score without knowing the ID first
+      // So we'll use a simpler approach: priority-based score only
+      // The timePart will be handled separately after getting the ID
+      queueScore = (priority * 1000000000).toString();
+    }
+
+    const result = await this.redis.eval(script, {
+      keys: [this.idempotencyKey, this.messagesKey, this.idKey, queueKey],
+      arguments: [meta.idempotencyKey || '', message, queueScore]
+    });
+
+    const id = result.toString();
+
+    // For non-delayed jobs, we need to update the score to include the FIFO timePart
+    if (delaySeconds === 0) {
       const jobIdNum = parseInt(id);
-      const timePart = 1000000000 - jobIdNum; // Invert job ID for FIFO within same priority
-      const score = priority * 1000000000 + timePart;
-      await this.redis.zAdd(this.waitingKey, { score, value: id });
+      const timePart = 1000000000 - jobIdNum;
+      const finalScore = priority * 1000000000 + timePart;
+      await this.redis.zAdd(this.waitingKey, { score: finalScore, value: id });
     }
 
     return id;