feat: add .id() method for job deduplication

Author: OhSo-Dublin

README.md

@@ -26,6 +26,7 @@ npm install @boringnode/queue
 - **Priority Queues**: Process high-priority jobs first
 - **Bulk Dispatch**: Efficiently dispatch thousands of jobs at once
 - **Job Grouping**: Organize related jobs for monitoring
+- **Job Deduplication**: Prevent duplicate jobs with custom IDs
 - **Retry with Backoff**: Exponential, linear, or fixed backoff strategies
 - **Job Timeout**: Fail or retry jobs that exceed a time limit
 - **Job History**: Retain completed/failed jobs for debugging
@@ -131,6 +132,37 @@ await SendEmailJob.dispatchMany(recipients).group('newsletter-jan-2025')
 
 The `groupId` is stored with job data and accessible via `job.data.groupId`.
 
+## Job Deduplication
+
+Prevent the same job from being pushed to the queue twice using custom job IDs:
+
+```typescript
+// First dispatch - job is created
+await SendInvoiceJob.dispatch({ orderId: 123 }).id('order-123').run()
+
+// Second dispatch with same ID - silently skipped
+await SendInvoiceJob.dispatch({ orderId: 123 }).id('order-123').run()
+```
+
+The custom ID is automatically prefixed with the job name, so different job types can use the same ID without conflicts:
+
+```typescript
+// These are two different jobs, no conflict
+await SendInvoiceJob.dispatch({ orderId: 123 }).id('order-123').run()
+await SendReceiptJob.dispatch({ orderId: 123 }).id('order-123').run()
+```
+
+Deduplication is atomic and race-condition-free across all adapters:
+
+- **Redis**: Uses `HSETNX` (set-if-not-exists)
+- **Knex**: Uses `INSERT ... ON CONFLICT DO NOTHING`
+
+> [!NOTE]
+> Without `.id()`, jobs use auto-generated UUIDs and are never deduplicated. The `.id()` method is only available on single dispatch, not `dispatchMany`.
+
+> [!TIP]
+> When job retention is enabled (`removeOnComplete: false`), completed jobs remain in storage. A re-dispatch with the same custom ID will be silently skipped since the record still exists. With the default retention (`true`), completed jobs are removed immediately, so re-dispatch with the same ID succeeds normally.
+
 ## Job History & Retention
 
 Keep completed and failed jobs for debugging:

src/drivers/fake_adapter.ts

@@ -160,6 +160,11 @@ export class FakeAdapter implements Adapter {
   }
 
   async pushOn(queue: string, jobData: JobData): Promise<void> {
+    if (jobData.unique) {
+      const jobs = this.#queues.get(queue)
+      if (jobs?.some((j) => j.id === jobData.id)) return
+    }
+
     this.#recordPush(queue, jobData)
     this.#enqueue(queue, jobData)
   }
@@ -169,6 +174,14 @@ export class FakeAdapter implements Adapter {
   }
 
   pushLaterOn(queue: string, jobData: JobData, delay: number): Promise<void> {
+    if (jobData.unique) {
+      const jobs = this.#queues.get(queue)
+      if (jobs?.some((j) => j.id === jobData.id)) return Promise.resolve()
+
+      const delayed = this.#delayedJobs.get(queue)
+      if (delayed?.has(jobData.id)) return Promise.resolve()
+    }
+
     this.#recordPush(queue, jobData, delay)
     this.#schedulePush(queue, jobData, delay)
 

src/drivers/knex_adapter.ts

@@ -370,13 +370,19 @@ export class KnexAdapter implements Adapter {
     const timestamp = Date.now()
     const score = calculateScore(priority, timestamp)
 
-    await this.#connection(this.#jobsTable).insert({
+    const query = this.#connection(this.#jobsTable).insert({
       id: jobData.id,
       queue,
       status: 'pending',
       data: JSON.stringify(jobData),
       score,
     })
+
+    if (jobData.unique) {
+      await query.onConflict(['id', 'queue']).ignore()
+    } else {
+      await query
+    }
   }
 
   async pushLater(jobData: JobData, delay: number): Promise<void> {
@@ -386,13 +392,19 @@ export class KnexAdapter implements Adapter {
   async pushLaterOn(queue: string, jobData: JobData, delay: number): Promise<void> {
     const executeAt = Date.now() + delay
 
-    await this.#connection(this.#jobsTable).insert({
+    const query = this.#connection(this.#jobsTable).insert({
       id: jobData.id,
       queue,
       status: 'delayed',
       data: JSON.stringify(jobData),
       execute_at: executeAt,
     })
+
+    if (jobData.unique) {
+      await query.onConflict(['id', 'queue']).ignore()
+    } else {
+      await query
+    }
   }
 
   async pushMany(jobs: JobData[]): Promise<void> {

src/drivers/redis_adapter.ts

@@ -35,6 +35,26 @@ const PUSH_JOB_SCRIPT = `
   return 1
 `
 
+/**
+ * Lua script for pushing a unique job.
+ * Uses HSETNX to only store data if the job doesn't already exist.
+ * Only adds to pending ZSET if the job was newly created.
+ */
+const PUSH_UNIQUE_JOB_SCRIPT = `
+  local data_key = KEYS[1]
+  local pending_key = KEYS[2]
+  local job_id = ARGV[1]
+  local job_data = ARGV[2]
+  local score = tonumber(ARGV[3])
+
+  local added = redis.call('HSETNX', data_key, job_id, job_data)
+  if added == 1 then
+    redis.call('ZADD', pending_key, score, job_id)
+  end
+
+  return added
+`
+
 /**
  * Lua script for pushing a delayed job.
  * Stores job data in the central hash and adds jobId to delayed ZSET.
@@ -52,6 +72,26 @@ const PUSH_DELAYED_JOB_SCRIPT = `
   return 1
 `
 
+/**
+ * Lua script for pushing a unique delayed job.
+ * Uses HSETNX to only store data if the job doesn't already exist.
+ * Only adds to delayed ZSET if the job was newly created.
+ */
+const PUSH_UNIQUE_DELAYED_JOB_SCRIPT = `
+  local data_key = KEYS[1]
+  local delayed_key = KEYS[2]
+  local job_id = ARGV[1]
+  local job_data = ARGV[2]
+  local execute_at = tonumber(ARGV[3])
+
+  local added = redis.call('HSETNX', data_key, job_id, job_data)
+  if added == 1 then
+    redis.call('ZADD', delayed_key, execute_at, job_id)
+  end
+
+  return added
+`
+
 /**
  * Lua script for atomic job acquisition.
  * 1. Check and process delayed jobs
@@ -620,8 +660,10 @@ export class RedisAdapter implements Adapter {
     const keys = this.#getKeys(queue)
     const executeAt = Date.now() + delay
 
+    const script = jobData.unique ? PUSH_UNIQUE_DELAYED_JOB_SCRIPT : PUSH_DELAYED_JOB_SCRIPT
+
     await this.#connection.eval(
-      PUSH_DELAYED_JOB_SCRIPT,
+      script,
       2,
       keys.data,
       keys.delayed,
@@ -637,8 +679,10 @@ export class RedisAdapter implements Adapter {
     const timestamp = Date.now()
     const score = calculateScore(priority, timestamp)
 
+    const script = jobData.unique ? PUSH_UNIQUE_JOB_SCRIPT : PUSH_JOB_SCRIPT
+
     await this.#connection.eval(
-      PUSH_JOB_SCRIPT,
+      script,
       2,
       keys.data,
       keys.pending,

src/job_dispatcher.ts

@@ -47,6 +47,7 @@ export class JobDispatcher<T> {
   #delay?: Duration
   #priority?: number
   #groupId?: string
+  #id?: string
 
   /**
    * Create a new job dispatcher.
@@ -148,6 +149,40 @@ export class JobDispatcher<T> {
     return this
   }
 
+  /**
+   * Set a custom job ID for deduplication.
+   *
+   * When a custom ID is provided, the adapter will silently skip
+   * the job if one with the same ID already exists in the queue.
+   * The ID is automatically prefixed with the job name to prevent
+   * collisions between different job types.
+   *
+   * @param jobId - Custom identifier for this job
+   * @returns This dispatcher for chaining
+   *
+   * @example
+   * ```typescript
+   * // Prevent duplicate invoice jobs for the same order
+   * await SendInvoiceJob.dispatch({ orderId: 123 })
+   *   .id('order-123')
+   *   .run()
+   *
+   * // Second dispatch with same ID is silently skipped
+   * await SendInvoiceJob.dispatch({ orderId: 123 })
+   *   .id('order-123')
+   *   .run()
+   * ```
+   */
+  id(jobId: string): this {
+    if (!jobId) {
+      throw new Error('Job ID must be a non-empty string')
+    }
+
+    this.#id = jobId
+
+    return this
+  }
+
   /**
    * Use a specific adapter for this job.
    *
@@ -181,7 +216,7 @@ export class JobDispatcher<T> {
    * ```
    */
   async run(): Promise<DispatchResult> {
-    const id = randomUUID()
+    const id = this.#id ? `${this.#name}::${this.#id}` : randomUUID()
 
     debug('dispatching job %s with id %s using payload %s', this.#name, id, this.#payload)
 
@@ -197,6 +232,7 @@ export class JobDispatcher<T> {
       priority: this.#priority,
       groupId: this.#groupId,
       createdAt: Date.now(),
+      ...(this.#id ? { unique: true } : {}),
     }
 
     const message: JobDispatchMessage = { jobs: [jobData], queue: this.#queue, delay: parsedDelay }

src/types/main.ts

@@ -133,6 +133,13 @@ export interface JobData {
    * Injected by OTel plugin at dispatch time.
    */
   traceContext?: Record<string, string>
+
+  /**
+   * When true, adapters use atomic insert-if-not-exists semantics
+   * to silently skip duplicate jobs with the same ID.
+   * Set automatically when a custom job ID is provided via `.id()`.
+   */
+  unique?: boolean
 }
 
 /**

tests/_mocks/memory_adapter.ts

@@ -51,6 +51,11 @@ export class MemoryAdapter implements Adapter {
   }
 
   async pushOn(queue: string, jobData: JobData): Promise<void> {
+    if (jobData.unique) {
+      const jobs = this.#queues.get(queue)
+      if (jobs?.some((j) => j.id === jobData.id)) return
+    }
+
     if (!this.#queues.has(queue)) {
       this.#queues.set(queue, [])
     }
@@ -63,6 +68,14 @@ export class MemoryAdapter implements Adapter {
   }
 
   pushLaterOn(queue: string, jobData: JobData, delay: number): Promise<void> {
+    if (jobData.unique) {
+      const jobs = this.#queues.get(queue)
+      if (jobs?.some((j) => j.id === jobData.id)) return Promise.resolve()
+
+      const delayed = this.#delayedJobs.get(queue)
+      if (delayed?.has(jobData.id)) return Promise.resolve()
+    }
+
     if (!this.#delayedJobs.has(queue)) {
       this.#delayedJobs.set(queue, new Map())
     }