@pearl-framework/queue

BullMQ-powered background job queue with retry, backoff, and worker management.

Installation

npm install @pearl-framework/queue @pearl-framework/core bullmq ioredis

Requires a running Redis instance. The default connection is localhost:6379.

Define a Job

Important: Job payload must be plain public properties, not constructor arguments. The worker reconstructs jobs using new JobClass() then restores data via Object.assign. Constructor arguments are lost after serialization.

import { Job } from '@pearl-framework/queue'

export class SendWelcomeEmail extends Job {
  readonly queue   = 'mail'
  get tries()      { return 3 }
  get retryDelay() { return 2_000 }

  userId!: number 

  async handle(): Promise<void> {
    const user = await User.find(db, this.userId)
    if (!user) return
    await mailer.send(new WelcomeEmail(user))
  }

  async failed(error: Error): Promise<void> {
    // Called when all retry attempts are exhausted
    console.error(`SendWelcomeEmail failed for user ${this.userId}:`, error.message)
    await alertSlack(`Job failed: ${error.message}`)
  }
}

Dispatch Jobs

const queue = app.make(QueueManager)

// Dispatch immediately
const job = new SendWelcomeEmail()
job.userId = user.id
await queue.dispatch(job)

// Dispatch with a delay
await queue.dispatchAfter(job, 5_000)  // runs in 5 seconds

// Dispatch multiple
await queue.dispatchBulk([
  Object.assign(new SendWelcomeEmail(), { userId: 1 }),
  Object.assign(new SendWelcomeEmail(), { userId: 2 }),
  Object.assign(new SendWelcomeEmail(), { userId: 3 }),
])

QueueServiceProvider

import { QueueServiceProvider } from '@pearl-framework/queue'

export class AppQueueServiceProvider extends QueueServiceProvider {
  protected config = {
    connection: {
      host: process.env.REDIS_HOST ?? 'localhost',
      port: Number(process.env.REDIS_PORT ?? 6379),
    },
    workers: [
      { queue: 'default', concurrency: 5 },
      { queue: 'mail',    concurrency: 2 },
      { queue: 'reports', concurrency: 1 },
    ],
  }
}

app.register(AppQueueServiceProvider)

Custom Job Options

Override jobOptions for full control over BullMQ options on a per-job basis:

import type { JobsOptions } from 'bullmq'

export class ProcessReport extends Job {
  readonly queue = 'reports'
  reportId!: number

  // Override for full BullMQ options
  get jobOptions(): JobsOptions {
    return {
      attempts:         5,
      backoff:          { type: 'fixed', delay: 30_000 },  // fixed 30s between retries
      removeOnComplete: true,
      removeOnFail:     false,   // keep failed jobs for inspection
      priority:         10,
    }
  }

  async handle(): Promise<void> {
    await generateReport(this.reportId)
  }
}

Multiple Queues

Use readonly queue to route different job types to dedicated workers:

// High-priority jobs — 10 concurrent workers
export class ProcessPayment extends Job {
  readonly queue = 'critical'
  paymentId!: number
  async handle() { /* ... */ }
}

// Low-priority jobs — 1 worker
export class GenerateMonthlyReport extends Job {
  readonly queue = 'reports'
  month!: string
  async handle() { /* ... */ }
}

// QueueServiceProvider config
workers: [
  { queue: 'default',  concurrency: 5 },
  { queue: 'critical', concurrency: 10 },
  { queue: 'reports',  concurrency: 1 },
]

API Reference

`Job` (base class)

Property / Method	Default	Description
`queue`	`'default'`	Queue name to dispatch to
`tries`	`3`	Max attempts before marking failed
`retryDelay`	`1000`	Initial retry delay in ms (doubles each retry)
`delay`	`0`	Delay before first execution in ms
`jobOptions`	—	Override for full `JobsOptions` control
`handle()`	—	Required. The job's main logic
`failed(error)`	—	Called when all retry attempts are exhausted
`serialize()`	—	Serialises job payload for BullMQ — override for custom behaviour

`QueueManager`

Method	Description
`dispatch(job)`	Add a job to its queue
`dispatchAfter(job, delayMs)`	Add a job with a delay before it runs
`dispatchBulk(jobs[])`	Add multiple jobs in a single operation

@pearl-framework/queue

Package Exports

Readme

@pearl-framework/queue

Installation

Define a Job

Dispatch Jobs

QueueServiceProvider

Custom Job Options

Multiple Queues

API Reference

Job (base class)

QueueManager

`Job` (base class)

`QueueManager`