feat: add groupId support for job grouping

Author: RomainLanz

.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'
+```

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 },
   }
 

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) {

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
 }
 
 /**

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)
+  })
+})