feat: add dispatchMany for bulk job dispatch

Author: RomainLanz

.changelog/push-many.md

@@ -0,0 +1,53 @@
+# Bulk job dispatch
+
+## New feature
+
+### dispatchMany
+
+Jobs can now be dispatched in batches using `Job.dispatchMany()`. This is more efficient than calling `dispatch()` multiple times as it uses optimized batch operations (Redis MULTI/EXEC transaction, SQL batch insert).
+
+```typescript
+// Dispatch multiple jobs at once
+const { jobIds } = await SendEmailJob.dispatchMany([
+  { to: 'user1@example.com', subject: 'Newsletter' },
+  { to: 'user2@example.com', subject: 'Newsletter' },
+  { to: 'user3@example.com', subject: 'Newsletter' },
+])
+  .group('newsletter-jan-2025')
+  .toQueue('emails')
+  .priority(3)
+  .run()
+
+console.log(`Dispatched ${jobIds.length} jobs`)
+```
+
+### Use cases
+
+- **Newsletters**: Send thousands of emails in a single batch operation
+- **Bulk exports**: Create export jobs for multiple users
+- **Data migrations**: Queue many transformation jobs at once
+- **Notifications**: Dispatch notifications to many recipients
+
+### API
+
+The `JobBatchDispatcher` supports the same fluent API as `JobDispatcher`:
+
+```typescript
+await SendEmailJob.dispatchMany(payloads)
+  .toQueue('emails')      // Target queue
+  .group('batch-123')     // Group all jobs together
+  .priority(1)            // Set priority for all jobs
+  .with('redis')          // Use specific adapter
+  .run()
+```
+
+### Adapter API
+
+For low-level access, adapters now support `pushMany()` and `pushManyOn()`:
+
+```typescript
+await adapter.pushManyOn('emails', [
+  { id: 'uuid1', name: 'SendEmailJob', payload: {...}, attempts: 0 },
+  { id: 'uuid2', name: 'SendEmailJob', payload: {...}, attempts: 0 },
+])
+```

index.ts

@@ -4,7 +4,8 @@ export { QueueManager } from './src/queue_manager.js'
 export { Locator } from './src/locator.js'
 export { Schedule } from './src/schedule.js'
 export { ScheduleBuilder } from './src/schedule_builder.js'
-export type { AdapterFactory, JobFactory } from './src/types/main.js'
+export { JobBatchDispatcher } from './src/job_batch_dispatcher.js'
+export type { AdapterFactory, JobFactory, DispatchManyResult } from './src/types/main.js'
 export {
   customBackoff,
   linearBackoff,

src/contracts/adapter.ts

@@ -147,6 +147,27 @@ export interface Adapter {
    */
   pushLaterOn(queue: string, jobData: JobData, delay: number): Promise<void>
 
+  /**
+   * Push multiple jobs to the default queue for immediate processing.
+   *
+   * This is more efficient than calling push() multiple times as it
+   * batches the operations (e.g., Redis pipeline, SQL batch insert).
+   *
+   * @param jobs - Array of job data to push
+   */
+  pushMany(jobs: JobData[]): Promise<void>
+
+  /**
+   * Push multiple jobs to a specific queue for immediate processing.
+   *
+   * This is more efficient than calling pushOn() multiple times as it
+   * batches the operations (e.g., Redis pipeline, SQL batch insert).
+   *
+   * @param queue - The queue name to push to
+   * @param jobs - Array of job data to push
+   */
+  pushManyOn(queue: string, jobs: JobData[]): Promise<void>
+
   /**
    * Get the number of pending jobs in the default queue.
    *

src/drivers/knex_adapter.ts

@@ -445,6 +445,27 @@ export class KnexAdapter implements Adapter {
     })
   }
 
+  async pushMany(jobs: JobData[]): Promise<void> {
+    return this.pushManyOn('default', jobs)
+  }
+
+  async pushManyOn(queue: string, jobs: JobData[]): Promise<void> {
+    if (jobs.length === 0) return
+
+    await this.#ensureTables()
+
+    const now = Date.now()
+    const rows = jobs.map((job) => ({
+      id: job.id,
+      queue,
+      status: 'pending' as const,
+      data: JSON.stringify(job),
+      score: calculateScore(job.priority ?? DEFAULT_PRIORITY, now),
+    }))
+
+    await this.#connection(this.#jobsTable).insert(rows)
+  }
+
   async size(): Promise<number> {
     return this.sizeOf('default')
   }

src/drivers/redis_adapter.ts

@@ -648,6 +648,27 @@ export class RedisAdapter implements Adapter {
     )
   }
 
+  pushMany(jobs: JobData[]): Promise<void> {
+    return this.pushManyOn('default', jobs)
+  }
+
+  async pushManyOn(queue: string, jobs: JobData[]): Promise<void> {
+    if (jobs.length === 0) return
+
+    const keys = this.#getKeys(queue)
+    const now = Date.now()
+    const multi = this.#connection.multi()
+
+    for (const job of jobs) {
+      const priority = job.priority ?? DEFAULT_PRIORITY
+      const score = calculateScore(priority, now)
+      multi.hset(keys.data, job.id, JSON.stringify(job))
+      multi.zadd(keys.pending, score, job.id)
+    }
+
+    await multi.exec()
+  }
+
   size(): Promise<number> {
     return this.sizeOf('default')
   }

src/drivers/sync_adapter.ts

@@ -45,6 +45,16 @@ export class SyncAdapter implements Adapter {
     return Promise.resolve()
   }
 
+  pushMany(jobs: JobData[]): Promise<void> {
+    return this.pushManyOn('default', jobs)
+  }
+
+  async pushManyOn(queue: string, jobs: JobData[]): Promise<void> {
+    for (const job of jobs) {
+      await this.pushOn(queue, job)
+    }
+  }
+
   size(): Promise<number> {
     return this.sizeOf('default')
   }

src/job.ts

@@ -1,4 +1,5 @@
 import { JobDispatcher } from './job_dispatcher.js'
+import { JobBatchDispatcher } from './job_batch_dispatcher.js'
 import { ScheduleBuilder } from './schedule_builder.js'
 import type { JobContext, JobOptions } from './types/main.js'
 
@@ -193,6 +194,57 @@ export abstract class Job<Payload = any> {
     return dispatcher
   }
 
+  /**
+   * Dispatch multiple jobs to the queue in a single batch.
+   *
+   * Returns a JobBatchDispatcher for fluent configuration before dispatching.
+   * The jobs are not actually dispatched until `.run()` is called or the
+   * dispatcher is awaited.
+   *
+   * This is more efficient than calling `dispatch()` multiple times as it
+   * uses batched operations (e.g., Redis pipeline, SQL batch insert).
+   *
+   * @param payloads - Array of data to pass to each job
+   * @returns A JobBatchDispatcher for fluent configuration
+   *
+   * @example
+   * ```typescript
+   * // Batch dispatch for newsletter
+   * const { jobIds } = await SendEmailJob.dispatchMany([
+   *   { to: 'user1@example.com', subject: 'Newsletter' },
+   *   { to: 'user2@example.com', subject: 'Newsletter' },
+   * ])
+   *   .group('newsletter-jan-2025')
+   *   .toQueue('emails')
+   *   .run()
+   *
+   * console.log(`Dispatched ${jobIds.length} jobs`)
+   * ```
+   */
+  static dispatchMany<T extends Job>(
+    this: new (...args: any[]) => T,
+    payloads: (T extends Job<infer P> ? P : never)[]
+  ): JobBatchDispatcher<T extends Job<infer P> ? P : never> {
+    const options = (this as any).options || {}
+    const jobName = options.name || this.name
+
+    const dispatcher = new JobBatchDispatcher<T extends Job<infer P> ? P : never>(jobName, payloads)
+
+    if (options.queue) {
+      dispatcher.toQueue(options.queue)
+    }
+
+    if (options.adapter) {
+      dispatcher.with(options.adapter)
+    }
+
+    if (options.priority !== undefined) {
+      dispatcher.priority(options.priority)
+    }
+
+    return dispatcher
+  }
+
   /**
    * Create a schedule for this job.
    *