refactor: replace .id() with .dedup() API for extensible job deduplication

Author: jigs1996

.gitignore

@@ -15,6 +15,9 @@ coverage
 npm-debug.log
 yarn-error.log
 
+# OS specific
+.DS_Store
+
 # Editors specific
 .fleet
 .idea

README.md

@@ -134,22 +134,22 @@ 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:
+Prevent the same job from being pushed to the queue twice using `.dedup()`:
 
 ```typescript
 // First dispatch - job is created
-await SendInvoiceJob.dispatch({ orderId: 123 }).id('order-123').run()
+await SendInvoiceJob.dispatch({ orderId: 123 }).dedup({ id: 'order-123' }).run()
 
-// Second dispatch with same ID - silently skipped
-await SendInvoiceJob.dispatch({ orderId: 123 }).id('order-123').run()
+// Second dispatch with same dedup ID - silently skipped
+await SendInvoiceJob.dispatch({ orderId: 123 }).dedup({ 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:
+The dedup 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()
+await SendInvoiceJob.dispatch({ orderId: 123 }).dedup({ id: 'order-123' }).run()
+await SendReceiptJob.dispatch({ orderId: 123 }).dedup({ id: 'order-123' }).run()
 ```
 
 Deduplication is atomic and race-condition-free for adapters that support storage-level uniqueness checks:
@@ -159,10 +159,10 @@ Deduplication is atomic and race-condition-free for adapters that support storag
 - **SyncAdapter**: Executes jobs inline and does not support deduplication
 
 > [!NOTE]
-> Without `.id()`, jobs use auto-generated UUIDs and are never deduplicated. The `.id()` method is only available on single dispatch, not `dispatchMany`.
+> Without `.dedup()`, jobs use auto-generated UUIDs and are never deduplicated. The `.dedup()` 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.
+> When job retention is enabled (`removeOnComplete: false`), completed jobs remain in storage. A re-dispatch with the same dedup 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
 

src/drivers/fake_adapter.ts

@@ -160,7 +160,7 @@ export class FakeAdapter implements Adapter {
   }
 
   async pushOn(queue: string, jobData: JobData): Promise<void> {
-    if (jobData.unique) {
+    if (jobData.dedup) {
       const existing = await this.getJob(jobData.id, queue)
       if (existing) return
     }
@@ -174,7 +174,7 @@ export class FakeAdapter implements Adapter {
   }
 
   async pushLaterOn(queue: string, jobData: JobData, delay: number): Promise<void> {
-    if (jobData.unique) {
+    if (jobData.dedup) {
       const existing = await this.getJob(jobData.id, queue)
       if (existing) return
     }

src/drivers/knex_adapter.ts

@@ -378,7 +378,7 @@ export class KnexAdapter implements Adapter {
       score,
     })
 
-    if (jobData.unique) {
+    if (jobData.dedup) {
       await query.onConflict(['id', 'queue']).ignore()
     } else {
       await query
@@ -400,7 +400,7 @@ export class KnexAdapter implements Adapter {
       execute_at: executeAt,
     })
 
-    if (jobData.unique) {
+    if (jobData.dedup) {
       await query.onConflict(['id', 'queue']).ignore()
     } else {
       await query

src/drivers/redis_adapter.ts

@@ -36,11 +36,11 @@ const PUSH_JOB_SCRIPT = `
 `
 
 /**
- * Lua script for pushing a unique job.
+ * Lua script for pushing a dedup 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 = `
+const PUSH_DEDUP_JOB_SCRIPT = `
   local data_key = KEYS[1]
   local pending_key = KEYS[2]
   local job_id = ARGV[1]
@@ -73,11 +73,11 @@ const PUSH_DELAYED_JOB_SCRIPT = `
 `
 
 /**
- * Lua script for pushing a unique delayed job.
+ * Lua script for pushing a dedup 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 = `
+const PUSH_DEDUP_DELAYED_JOB_SCRIPT = `
   local data_key = KEYS[1]
   local delayed_key = KEYS[2]
   local job_id = ARGV[1]
@@ -660,7 +660,7 @@ 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
+    const script = jobData.dedup ? PUSH_DEDUP_DELAYED_JOB_SCRIPT : PUSH_DELAYED_JOB_SCRIPT
 
     await this.#connection.eval(
       script,
@@ -679,7 +679,7 @@ 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
+    const script = jobData.dedup ? PUSH_DEDUP_JOB_SCRIPT : PUSH_JOB_SCRIPT
 
     await this.#connection.eval(
       script,

src/job_dispatcher.ts

@@ -15,11 +15,12 @@ import { parse } from './utils.js'
  *
  * ```
  * Job.dispatch(payload)
- *     .toQueue('emails')     // optional: target queue
- *     .priority(1)           // optional: 1-10, lower = higher priority
- *     .in('5m')              // optional: delay before processing
- *     .with('redis')         // optional: specific adapter
- *     .run()                 // dispatch the job
+ *     .toQueue('emails')              // optional: target queue
+ *     .priority(1)                    // optional: 1-10, lower = higher priority
+ *     .in('5m')                       // optional: delay before processing
+ *     .dedup({ id: 'order-123' })     // optional: deduplication
+ *     .with('redis')                  // optional: specific adapter
+ *     .run()                          // dispatch the job
  * ```
  *
  * @typeParam T - The payload type for this job
@@ -47,7 +48,7 @@ export class JobDispatcher<T> {
   #delay?: Duration
   #priority?: number
   #groupId?: string
-  #id?: string
+  #dedup?: { id: string }
 
   /**
    * Create a new job dispatcher.
@@ -150,35 +151,36 @@ export class JobDispatcher<T> {
   }
 
   /**
-   * Set a custom job ID for deduplication.
+   * Configure deduplication for this job.
    *
-   * When a custom ID is provided, the adapter will silently skip
-   * the job if one with the same ID already exists in the queue.
+   * When deduplication is configured, the adapter will silently skip
+   * the job if one with the same dedup 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
+   * @param options - Deduplication options
+   * @param options.id - Unique deduplication key
    * @returns This dispatcher for chaining
    *
    * @example
    * ```typescript
    * // Prevent duplicate invoice jobs for the same order
    * await SendInvoiceJob.dispatch({ orderId: 123 })
-   *   .id('order-123')
+   *   .dedup({ id: 'order-123' })
    *   .run()
    *
-   * // Second dispatch with same ID is silently skipped
+   * // Second dispatch with same dedup ID is silently skipped
    * await SendInvoiceJob.dispatch({ orderId: 123 })
-   *   .id('order-123')
+   *   .dedup({ id: 'order-123' })
    *   .run()
    * ```
    */
-  id(jobId: string): this {
-    if (!jobId) {
-      throw new Error('Job ID must be a non-empty string')
+  dedup(options: { id: string }): this {
+    if (!options.id) {
+      throw new Error('Dedup ID must be a non-empty string')
     }
 
-    this.#id = jobId
+    this.#dedup = options
 
     return this
   }
@@ -216,7 +218,7 @@ export class JobDispatcher<T> {
    * ```
    */
   async run(): Promise<DispatchResult> {
-    const id = this.#id ? `${this.#name}::${this.#id}` : randomUUID()
+    const id = this.#dedup ? `${this.#name}::${this.#dedup.id}` : randomUUID()
 
     debug('dispatching job %s with id %s using payload %s', this.#name, id, this.#payload)
 
@@ -232,7 +234,7 @@ export class JobDispatcher<T> {
       priority: this.#priority,
       groupId: this.#groupId,
       createdAt: Date.now(),
-      ...(this.#id ? { unique: true } : {}),
+      ...(this.#dedup ? { dedup: { id: this.#dedup.id } } : {}),
     }
 
     const message: JobDispatchMessage = { jobs: [jobData], queue: this.#queue, delay: parsedDelay }

src/types/main.ts

@@ -135,11 +135,15 @@ export interface JobData {
   traceContext?: Record<string, string>
 
   /**
-   * When true, adapters use atomic insert-if-not-exists semantics
+   * Deduplication configuration for this job.
+   * When set, 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()`.
+   * Set automatically when `.dedup()` is called on the dispatcher.
    */
-  unique?: boolean
+  dedup?: {
+    /** The original dedup key provided by the caller (before name-prefixing). */
+    id: string
+  }
 }
 
 /**

tests/_mocks/memory_adapter.ts

@@ -51,7 +51,7 @@ export class MemoryAdapter implements Adapter {
   }
 
   async pushOn(queue: string, jobData: JobData): Promise<void> {
-    if (jobData.unique) {
+    if (jobData.dedup) {
       const existing = await this.getJob(jobData.id, queue)
       if (existing) return
     }
@@ -68,7 +68,7 @@ export class MemoryAdapter implements Adapter {
   }
 
   async pushLaterOn(queue: string, jobData: JobData, delay: number): Promise<void> {
-    if (jobData.unique) {
+    if (jobData.dedup) {
       const existing = await this.getJob(jobData.id, queue)
       if (existing) return
     }

tests/_utils/register_driver_test_suite.ts

@@ -1650,7 +1650,7 @@ export function registerDriverTestSuite(options: DriverTestSuiteOptions) {
     assert.equal(third!.id, 'low')
   })
 
-  test('pushOn with unique flag should skip duplicate job', async ({ assert }) => {
+  test('pushOn with dedup should skip duplicate job', async ({ assert }) => {
     const adapter = await options.createAdapter()
     adapter.setWorkerId('worker-1')
 
@@ -1659,15 +1659,15 @@ export function registerDriverTestSuite(options: DriverTestSuiteOptions) {
       name: 'TestJob',
       payload: { attempt: 1 },
       attempts: 0,
-      unique: true,
+      dedup: { id: 'order-1' },
     })
 
     await adapter.pushOn('test-queue', {
       id: 'TestJob::order-1',
       name: 'TestJob',
       payload: { attempt: 2 },
       attempts: 0,
-      unique: true,
+      dedup: { id: 'order-1' },
     })
 
     const size = await adapter.sizeOf('test-queue')
@@ -1677,7 +1677,7 @@ export function registerDriverTestSuite(options: DriverTestSuiteOptions) {
     assert.deepEqual(job!.payload, { attempt: 1 })
   })
 
-  test('pushOn without unique flag should insert normally', async ({ assert }) => {
+  test('pushOn without dedup should insert normally', async ({ assert }) => {
     const adapter = await options.createAdapter()
     adapter.setWorkerId('worker-1')
 
@@ -1699,7 +1699,7 @@ export function registerDriverTestSuite(options: DriverTestSuiteOptions) {
     assert.equal(size, 2)
   })
 
-  test('pushLaterOn with unique flag should skip duplicate delayed job', async ({ assert }) => {
+  test('pushLaterOn with dedup should skip duplicate delayed job', async ({ assert }) => {
     const adapter = await options.createAdapter()
     adapter.setWorkerId('worker-1')
 
@@ -1710,7 +1710,7 @@ export function registerDriverTestSuite(options: DriverTestSuiteOptions) {
         name: 'TestJob',
         payload: { attempt: 1 },
         attempts: 0,
-        unique: true,
+        dedup: { id: 'delayed-1' },
       },
       60_000
     )
@@ -1722,7 +1722,7 @@ export function registerDriverTestSuite(options: DriverTestSuiteOptions) {
         name: 'TestJob',
         payload: { attempt: 2 },
         attempts: 0,
-        unique: true,
+        dedup: { id: 'delayed-1' },
       },
       60_000
     )
@@ -1732,7 +1732,7 @@ export function registerDriverTestSuite(options: DriverTestSuiteOptions) {
     assert.deepEqual(job!.data.payload, { attempt: 1 })
   })
 
-  test('pushOn with unique flag should allow same id on different queues', async ({ assert }) => {
+  test('pushOn with dedup should allow same id on different queues', async ({ assert }) => {
     const adapter = await options.createAdapter()
     adapter.setWorkerId('worker-1')
 
@@ -1741,15 +1741,15 @@ export function registerDriverTestSuite(options: DriverTestSuiteOptions) {
       name: 'TestJob',
       payload: { queue: 'a' },
       attempts: 0,
-      unique: true,
+      dedup: { id: 'shared-id' },
     })
 
     await adapter.pushOn('queue-b', {
       id: 'TestJob::shared-id',
       name: 'TestJob',
       payload: { queue: 'b' },
       attempts: 0,
-      unique: true,
+      dedup: { id: 'shared-id' },
     })
 
     const sizeA = await adapter.sizeOf('queue-a')