wip: idempotency key

Author: muniter

ARCHITECTURE.md

@@ -7,8 +7,10 @@ A TypeScript queue system inspired by Yii2-Queue architecture, providing a clean
 - **Driver-based architecture**: Swap between DB, SQS, and File drivers seamlessly
 - **Event-based jobs**: Register handlers for job types without complex classes
 - **Type-safe API**: Full TypeScript support with driver-specific option validation
-- **Multiple backends**: Database, Amazon SQS, and File storage drivers
+- **Multiple backends**: Database, Amazon SQS, File, Redis, and In-Memory drivers
+- **Idempotency**: Built-in deduplication with idempotency keys
 - **Event system**: Hook into queue lifecycle events
+- **Plugin system**: Extend functionality with plugins (ECS task protection, metrics, etc.)
 
 ## Installation
 
@@ -210,6 +212,62 @@ await sqsQueue.addJob('send-email', {
 });
 ```
 
+## Idempotency
+
+Prevent duplicate job processing by using idempotency keys. When a job with the same `idempotencyKey` is added while an existing job with that key is still `waiting`, `delayed`, or `reserved`, the existing job ID is returned instead of creating a duplicate.
+
+```typescript
+// First attempt - creates new job
+const id1 = await queue.addJob('send-email', {
+  payload: { to: 'user@example.com', subject: 'Welcome', body: 'Hello!' },
+  idempotencyKey: 'welcome-email-user123'
+});
+
+// Second attempt with same key - returns existing job ID
+const id2 = await queue.addJob('send-email', {
+  payload: { to: 'user@example.com', subject: 'Welcome', body: 'Hello!' },
+  idempotencyKey: 'welcome-email-user123'
+});
+
+console.log(id1 === id2); // true - no duplicate job created
+```
+
+**Idempotency Key Lifecycle:**
+- Keys are automatically cleaned up when jobs complete (successfully or with failure)
+- After completion/failure, the same key can be reused for new jobs
+- Deduplication lasts as long as the job remains in the queue
+
+**Driver Support:**
+
+| Driver | Idempotency Support | Implementation |
+|--------|-------------------|----------------|
+| InMemoryQueue | ✅ Yes | In-memory Map |
+| FileQueue | ✅ Yes | JSON index file |
+| SQLiteQueue | ✅ Yes | Unique partial index |
+| RedisQueue | ✅ Yes | Redis hash |
+| MongooseQueue | ✅ Yes | MongoDB partial index |
+| SqsQueue | ✅ FIFO queues only | AWS MessageDeduplicationId |
+| DbQueue | ✅ Yes* | *Depends on adapter implementation |
+
+**SQS FIFO Queue Example:**
+```typescript
+// SQS idempotency only works with FIFO queues (URLs ending in .fifo)
+const sqsQueue = new SqsQueue<MyJobs>(
+  sqsClient,
+  'https://sqs.us-east-1.amazonaws.com/123456789/my-queue.fifo' // Note: .fifo suffix
+);
+
+await sqsQueue.addJob('send-email', {
+  payload: { to: 'user@example.com', subject: 'Test', body: 'Hello' },
+  idempotencyKey: 'unique-key-123' // Uses AWS MessageDeduplicationId
+});
+```
+
+**Best Practices:**
+- Use descriptive keys that capture the uniqueness: `email-${userId}-${emailType}`, `invoice-${orderId}`
+- Include relevant identifiers in the key to prevent unintended deduplication
+- Don't reuse keys for semantically different operations
+
 ## Event Handling
 
 ```typescript

README.md

@@ -129,6 +129,7 @@ export abstract class Queue<
       delaySeconds: (options as any).delaySeconds ?? 0,
       priority: (options as any).priority ?? 0,
       pushedAt: new Date(),
+      idempotencyKey: options.idempotencyKey,
     };
 
     const event: QueueEvent = {

src/core/queue.ts

@@ -17,6 +17,7 @@ interface IndexData {
   waiting: Array<[id: string, ttr: number]>;
   delayed: Array<[id: string, ttr: number, time: number]>;
   reserved: Array<[id: string, ttr: number, attempt: number, time: number]>;
+  idempotency?: Record<string, string>; // idempotency key -> job ID
 }
 
 interface ReservedInfo {
@@ -63,15 +64,43 @@ export class FileQueue<TJobMap = Record<string, any>> extends Queue<TJobMap, Fil
   }
 
   protected async pushMessage(payload: unknown, meta: JobMeta): Promise<string> {
-    const id = await this.touchIndex(async (data) => {
+    const id = await this.touchIndex(async (data): Promise<string> => {
+      // Initialize idempotency map if it doesn't exist
+      if (!data.idempotency) {
+        data.idempotency = {};
+      }
+
+      // Check for idempotency
+      if (meta.idempotencyKey) {
+        const existingJobId = data.idempotency[meta.idempotencyKey];
+        if (existingJobId) {
+          // Check if job still exists in waiting, delayed, or reserved
+          const stillExists =
+            data.waiting.some(item => item[0] === existingJobId) ||
+            data.delayed.some(item => item[0] === existingJobId) ||
+            data.reserved.some(item => item[0] === existingJobId);
+
+          if (stillExists) {
+            return existingJobId;
+          }
+          // Job completed/failed, clean up the idempotency mapping
+          delete data.idempotency[meta.idempotencyKey];
+        }
+      }
+
       const jobId = String(++data.lastId);
       const jobPath = path.join(this.path, `job${jobId}.data`);
-      
+
       await fs.writeFile(jobPath, JSON.stringify({ payload, meta }), 'utf8');
       if (this.fileMode !== undefined) {
         await fs.chmod(jobPath, this.fileMode);
       }
 
+      // Store idempotency key mapping if provided
+      if (meta.idempotencyKey) {
+        data.idempotency[meta.idempotencyKey] = jobId;
+      }
+
       const ttr = meta.ttr ?? 300;
       const delay = meta.delaySeconds ?? 0;
 
@@ -173,6 +202,16 @@ export class FileQueue<TJobMap = Record<string, any>> extends Queue<TJobMap, Fil
       if (index !== -1) {
         data.reserved.splice(index, 1);
       }
+
+      // Clean up idempotency key mapping
+      if (data.idempotency) {
+        const idempotencyKey = Object.keys(data.idempotency).find(
+          key => data.idempotency![key] === id
+        );
+        if (idempotencyKey) {
+          delete data.idempotency[idempotencyKey];
+        }
+      }
     });
 
     const jobPath = path.join(this.path, `job${id}.data`);
@@ -227,6 +266,7 @@ export class FileQueue<TJobMap = Record<string, any>> extends Queue<TJobMap, Fil
       data.waiting = [];
       data.delayed = [];
       data.reserved = [];
+      data.idempotency = {};
     });
 
     const files = await fs.readdir(this.path);

src/drivers/file.ts

@@ -78,6 +78,7 @@ export class InMemoryQueue<TJobMap = Record<string, any>> extends Queue<TJobMap,
   private reservedJobs = new Set<string>();
   private delayedJobs = new Map<string, NodeJS.Timeout>(); // job ID -> timeout handle
   private ttrTimeouts = new Map<string, NodeJS.Timeout>(); // job ID -> TTR timeout handle
+  private idempotencyKeys = new Map<string, string>(); // idempotency key -> job ID
   private nextJobId = 1;
   private maxJobs: number;
 
@@ -87,9 +88,25 @@ export class InMemoryQueue<TJobMap = Record<string, any>> extends Queue<TJobMap,
   }
 
   protected async pushMessage(payload: unknown, meta: JobMeta): Promise<string> {
+    // Check for idempotency
+    if (meta.idempotencyKey) {
+      const existingJobId = this.idempotencyKeys.get(meta.idempotencyKey);
+      if (existingJobId) {
+        const existingJob = this.jobs.get(existingJobId);
+        // Only deduplicate if job is still waiting, delayed, or reserved
+        if (existingJob && (existingJob.status === 'waiting' || existingJob.status === 'reserved')) {
+          return existingJobId;
+        }
+        // If job is done or failed, remove the idempotency key mapping
+        if (existingJob && (existingJob.status === 'done' || existingJob.status === 'failed')) {
+          this.idempotencyKeys.delete(meta.idempotencyKey);
+        }
+      }
+    }
+
     const id = (this.nextJobId++).toString();
     const now = new Date();
-    
+
     const job: InMemoryJobRecord = {
       id,
       payload,
@@ -100,24 +117,29 @@ export class InMemoryQueue<TJobMap = Record<string, any>> extends Queue<TJobMap,
 
     this.jobs.set(id, job);
 
+    // Store idempotency key mapping if provided
+    if (meta.idempotencyKey) {
+      this.idempotencyKeys.set(meta.idempotencyKey, id);
+    }
+
     // Handle delay
     if (meta.delaySeconds && meta.delaySeconds > 0) {
       job.delayTime = Date.now() + (meta.delaySeconds * 1000);
       job.status = 'waiting';
-      
+
       // Schedule job to become available after delay
       const timeout = setTimeout(() => {
         this.delayedJobs.delete(id);
         this.addToWaitingQueue(id);
       }, meta.delaySeconds * 1000);
-      
+
       this.delayedJobs.set(id, timeout);
     } else {
       // Add immediately to waiting queue
       this.addToWaitingQueue(id);
     }
     this.cleanupOldJobs();
-    
+
     return id;
   }
 
@@ -167,9 +189,14 @@ export class InMemoryQueue<TJobMap = Record<string, any>> extends Queue<TJobMap,
 
     job.status = 'done';
     job.doneAt = new Date();
-    
+
     this.reservedJobs.delete(message.id);
     this.clearTtrTimeout(message.id);
+
+    // Clean up idempotency key mapping
+    if (job.meta.idempotencyKey) {
+      this.idempotencyKeys.delete(job.meta.idempotencyKey);
+    }
   }
 
   protected async failJob(message: QueueMessage, error: unknown): Promise<void> {
@@ -179,9 +206,14 @@ export class InMemoryQueue<TJobMap = Record<string, any>> extends Queue<TJobMap,
     job.status = 'failed';
     job.doneAt = new Date();
     job.error = error instanceof Error ? error.message : String(error);
-    
+
     this.reservedJobs.delete(message.id);
     this.clearTtrTimeout(message.id);
+
+    // Clean up idempotency key mapping to allow re-enqueueing after failure
+    if (job.meta.idempotencyKey) {
+      this.idempotencyKeys.delete(job.meta.idempotencyKey);
+    }
   }
 
   async status(id: string): Promise<JobStatus> {
@@ -259,6 +291,7 @@ export class InMemoryQueue<TJobMap = Record<string, any>> extends Queue<TJobMap,
     this.reservedJobs.clear();
     this.delayedJobs.clear();
     this.ttrTimeouts.clear();
+    this.idempotencyKeys.clear();
     this.nextJobId = 1;
   }
 

src/drivers/memory.ts

@@ -37,6 +37,7 @@ export interface IQueueJobDocument {
   status: "waiting" | "reserved" | "done" | "failed";
   attempt: number;
   errorMessage?: string;
+  idempotencyKey?: string;
 }
 
 // Mongoose document interface
@@ -65,6 +66,7 @@ export const QueueJobSchema = new Schema<IQueueJob>(
     },
     attempt: { type: Number, required: true, default: 0 },
     errorMessage: { type: String },
+    idempotencyKey: { type: String, sparse: true },
   },
   {
     collection: "queue_jobs",
@@ -76,12 +78,32 @@ export const QueueJobSchema = new Schema<IQueueJob>(
 QueueJobSchema.index({ status: 1, delayTime: 1, priority: -1, pushTime: 1 });
 QueueJobSchema.index({ status: 1, expireTime: 1 });
 QueueJobSchema.index({ _id: 1, status: 1 });
+// Unique partial index for idempotency on active jobs
+QueueJobSchema.index(
+  { idempotencyKey: 1 },
+  { unique: true, sparse: true, partialFilterExpression: { status: { $in: ["waiting", "reserved"] } } }
+);
 
 // Mongoose database adapter implementing DatabaseAdapter interface
 export class MongooseDatabaseAdapter implements DatabaseAdapter {
   constructor(private model: Model<IQueueJob>) {}
 
   async insertJob(payload: unknown, meta: JobMeta): Promise<string> {
+    // Check for idempotency
+    if (meta.idempotencyKey) {
+      const existingJob = await this.model
+        .findOne({
+          idempotencyKey: meta.idempotencyKey,
+          status: { $in: ["waiting", "reserved"] },
+        })
+        .select("_id")
+        .exec();
+
+      if (existingJob) {
+        return existingJob._id.toHexString();
+      }
+    }
+
     const now = new Date();
 
     const doc = await this.model
@@ -97,6 +119,7 @@ export class MongooseDatabaseAdapter implements DatabaseAdapter {
           : null,
         status: "waiting",
         attempt: 0,
+        idempotencyKey: meta.idempotencyKey,
       })
       .catch((err) => {
         if (err instanceof Error && "code" in err) {