feat: add groupId support for job grouping

RomainLanz · RomainLanz · commit b2594d2e7de3 · 2026-01-14T22:45:18.000+01:00
diff --git a/.changelog/group-id.md b/.changelog/group-id.md
@@ -0,0 +1,36 @@
+# Job grouping
+
+## New feature
+
+### Group ID
+
+Jobs can now be assigned to a group using the `groupId` option. This allows organizing related jobs together for easier monitoring and filtering in UIs.
+
+```typescript
+// Assign jobs to a group
+await SendEmailJob.dispatch({ to: 'user@example.com' })
+  .group('newsletter-jan-2025')
+  .run()
+
+// Combine with other options
+await ExportJob.dispatch({ userId: 1 })
+  .group('batch-export-123')
+  .toQueue('exports')
+  .priority(2)
+  .run()
+```
+
+### Use cases
+
+- **Batch operations**: Group all jobs from a newsletter send, bulk export, or data migration
+- **Monitoring**: Filter and view related jobs together in queue UIs
+- **Debugging**: Easily find all jobs related to a specific operation
+
+### API
+
+The `groupId` is stored with the job data and can be accessed via:
+
+```typescript
+const record = await adapter.getJob('job-id', 'queue-name')
+console.log(record.data.groupId) // 'newsletter-jan-2025'
+```
diff --git a/examples/jobs/metrics_job.ts b/examples/jobs/metrics_job.ts
@@ -19,7 +19,7 @@ interface MetricsJobPayload {
 export default class MetricsJob extends Job<MetricsJobPayload> {
   static options: JobOptions = {
     queue: 'metrics',
-    removeOnComplete: true,
+    removeOnComplete: { count: 10 },
     removeOnFail: { count: 100 },
   }
 
diff --git a/src/job_dispatcher.ts b/src/job_dispatcher.ts
@@ -44,6 +44,7 @@ export class JobDispatcher<T> {
   #adapter?: string | (() => Adapter)
   #delay?: Duration
   #priority?: number
+  #groupId?: string
 
   /**
    * Create a new job dispatcher.
@@ -121,6 +122,30 @@ export class JobDispatcher<T> {
     return this
   }
 
+  /**
+   * Assign this job to a group.
+   *
+   * Jobs with the same groupId can be filtered and displayed together
+   * in monitoring UIs. Useful for batch operations like newsletters
+   * or bulk exports.
+   *
+   * @param groupId - Group identifier
+   * @returns This dispatcher for chaining
+   *
+   * @example
+   * ```typescript
+   * // Group newsletter jobs together
+   * await SendEmailJob.dispatch({ to: 'user@example.com' })
+   *   .group('newsletter-jan-2025')
+   *   .run()
+   * ```
+   */
+  group(groupId: string): this {
+    this.#groupId = groupId
+
+    return this
+  }
+
   /**
    * Use a specific adapter for this job.
    *
@@ -166,6 +191,7 @@ export class JobDispatcher<T> {
       payload: this.#payload,
       attempts: 0,
       priority: this.#priority,
+      groupId: this.#groupId,
     }
 
     if (this.#delay) {
diff --git a/src/types/main.ts b/src/types/main.ts
@@ -90,6 +90,22 @@ export interface JobData {
    * Number of times this job was recovered from stalled state.
    */
   stalledCount?: number
+
+  /**
+   * Optional group identifier for organizing related jobs.
+   *
+   * Jobs with the same groupId can be filtered and displayed together
+   * in monitoring UIs. Useful for batch operations like newsletters
+   * or bulk exports.
+   *
+   * @example
+   * ```typescript
+   * await SendEmailJob.dispatch({ to: 'user@example.com' })
+   *   .group('newsletter-jan-2025')
+   *   .run()
+   * ```
+   */
+  groupId?: string
 }
 
 /**
diff --git a/tests/job_dispatcher.spec.ts b/tests/job_dispatcher.spec.ts
@@ -222,3 +222,77 @@ test.group('JobDispatcher | DispatchResult', () => {
     assert.property(result, 'jobId')
   })
 })
+
+test.group('JobDispatcher | groupId', () => {
+  test('should dispatch job with groupId', async ({ assert }) => {
+    const sharedAdapter = memory()()
+
+    const localConfig = {
+      default: 'memory',
+      adapters: { memory: () => sharedAdapter },
+    }
+
+    await QueueManager.init(localConfig)
+
+    const dispatcher = new JobDispatcher('NewsletterJob', { userId: 1 })
+
+    const { jobId } = await dispatcher.group('newsletter-jan-2025').run()
+
+    assert.isString(jobId)
+
+    const job = await sharedAdapter.pop()
+
+    assert.isNotNull(job)
+    assert.equal(job!.id, jobId)
+    assert.equal(job!.groupId, 'newsletter-jan-2025')
+  })
+
+  test('should dispatch multiple jobs with same groupId', async ({ assert }) => {
+    const sharedAdapter = memory()()
+
+    const localConfig = {
+      default: 'memory',
+      adapters: { memory: () => sharedAdapter },
+    }
+
+    await QueueManager.init(localConfig)
+
+    const groupId = 'batch-export-123'
+
+    await new JobDispatcher('ExportJob', { userId: 1 }).group(groupId).run()
+    await new JobDispatcher('ExportJob', { userId: 2 }).group(groupId).run()
+    await new JobDispatcher('ExportJob', { userId: 3 }).group(groupId).run()
+
+    const job1 = await sharedAdapter.pop()
+    const job2 = await sharedAdapter.pop()
+    const job3 = await sharedAdapter.pop()
+
+    assert.equal(job1!.groupId, groupId)
+    assert.equal(job2!.groupId, groupId)
+    assert.equal(job3!.groupId, groupId)
+  })
+
+  test('should work with other options like priority and queue', async ({ assert }) => {
+    const sharedAdapter = memory()()
+
+    const localConfig = {
+      default: 'memory',
+      adapters: { memory: () => sharedAdapter },
+    }
+
+    await QueueManager.init(localConfig)
+
+    const { jobId } = await new JobDispatcher('ImportJob', { file: 'data.csv' })
+      .group('import-batch-456')
+      .toQueue('imports')
+      .priority(2)
+      .run()
+
+    const job = await sharedAdapter.popFrom('imports')
+
+    assert.isNotNull(job)
+    assert.equal(job!.id, jobId)
+    assert.equal(job!.groupId, 'import-batch-456')
+    assert.equal(job!.priority, 2)
+  })
+})

Original file line number	Diff line number	Diff line change
`@@ -19,7 +19,7 @@ interface MetricsJobPayload {`
`19`	`19`	`export default class MetricsJob extends Job<MetricsJobPayload> {`
`20`	`20`	`static options: JobOptions = {`
`21`	`21`	`queue: 'metrics',`
`22`		`- removeOnComplete: true,`
	`22`	`+ removeOnComplete: { count: 10 },`
`23`	`23`	`removeOnFail: { count: 100 },`
`24`	`24`	`}`
`25`	`25`