boringnode/queue

The Kysely adapter supports PostgreSQL, MySQL, and SQLite. Pass your application-owned Kysely instance to the adapter factory. The adapter will not destroy your instance. If you use custom table names in your migrations, specify them using tableName and schedulesTableName in the options.

import {
  kysely,
  KyselyQueueSchemaService,
  type QueueDatabase,
} from '@boringnode/queue/drivers/kysely_adapter'

interface Database extends QueueDatabase {
  orders: OrderTable
}

const adapter = kysely<Database>(db, { dialect: 'postgres' })

// Schema setup
const schema = new KyselyQueueSchemaService(db, { dialect: 'postgres' })
await schema.createJobsTable()
await schema.createSchedulesTable()

Install the package using npm to start using the queue system in your Node.js application.

npm install @boringnode/queue

The sync adapter executes jobs immediately in the same process. It is intended for tests and lightweight local development only.

Warning: When using the sync adapter, retries and backoff delays happen inline in the caller. If a job has a high maxRetries and a long backoff, the await job.dispatch().run() call will block the caller for the entire duration of the retry cycle. Use a worker-backed adapter in production to avoid blocking your main execution flow.

import { sync } from '@boringnode/queue/drivers/sync_adapter'

const adapter = sync() // Jobs execute immediately

// Note: This will block until all retries are exhausted
await MyJob.dispatch(payload).run()

To reload job implementations without restarting workers, use hotReload: true in QueueManager.init(). This requires the hot-hook package and an environment where Hot Hook is initialized (like an AdonisJS server with --hmr).

Note: Only jobs discovered via locations or Locator.registerFromGlob() support hot reloading. Manual registrations via Locator.register() do not.

npm install --save-dev hot-hook

await QueueManager.init({
  default: 'redis',
  adapters: {
    redis: redis({ host: 'localhost', port: 6379 }),
  },
  locations: ['./app/jobs/**/*.ts'],
  hotReload: process.env.NODE_ENV === 'development',
})

The library provides built-in OpenTelemetry instrumentation that creates PRODUCER spans for job dispatch and CONSUMER spans for job execution.

Note: This API is experimental and subject to change.

To enable instrumentation, use QueueInstrumentation and call manuallyRegister(boringqueue). This patches QueueManager.init() to automatically inject wrappers without requiring changes to your existing queue configuration.

import { QueueInstrumentation } from '@boringnode/queue/otel'
import * as boringqueue from '@boringnode/queue'

const instrumentation = new QueueInstrumentation({
  messagingSystem: 'boringqueue', // default
  executionSpanLinkMode: 'link', // or 'parent'
})

instrumentation.enable()
instrumentation.manuallyRegister(boringqueue)

The Knex adapter allows you to use relational databases. It requires tables to be created before use via the QueueSchemaService.

Database Setup: You can use QueueSchemaService to create the necessary jobs and schedules tables. You can also extend the jobs table with custom columns during creation.

import { knex } from '@boringnode/queue/drivers/knex_adapter'
import { QueueSchemaService } from '@boringnode/queue'
import Knex from 'knex'

// Initialize adapter
const connection = Knex({ client: 'pg', connection: '...' })
const adapter = knex(connection)

// Setup schema
const schemaService = new QueueSchemaService(connection)
await schemaService.createJobsTable()
await schemaService.createSchedulesTable()

// Custom columns example
await schemaService.createJobsTable('queue_jobs', (table) => {
  table.string('tenant_id', 255).nullable()
})

The @boringnode/queue package provides the following primary building blocks for managing asynchronous task execution:

• Job: The fundamental unit of work to be executed.
• Worker: The consumer that processes jobs from the queue.
• QueueManager: The central orchestrator for managing queues, jobs, and workers.
• Locator: Used for finding or identifying specific queue resources.
• Schedule & ScheduleBuilder: Tools for defining when jobs should run (e.g., cron-like or delayed execution).
• JobBatchDispatcher: Specialized component for dispatching multiple jobs at once.
• QueueSchemaService: Service for managing and validating queue schemas.
• backoff_strategies: Functions to define retry behaviors (customBackoff, linearBackoff, exponentialBackoff, fixedBackoff).