pg-boss-registry

Typed queue registry and lifecycle helpers for pg-boss.

The library keeps the real PgBoss instance visible. It adds typed helpers for queue names, payloads, workers, schedules, and registration order, so framework plugins for Fastify, Hono, Express, NestJS, or custom workers can share the same core definitions.

Install

npm install pg-boss-registry pg-boss

Quick Start

import { PgBoss } from 'pg-boss'
import {
  asTypedPgBoss,
  definePgBossQueues,
  queue,
  setupPgBoss,
  type PgBossQueuesFromRegistry,
} from 'pg-boss-registry'

type EmailJob = {
  userId: string
}

// Define every known queue in one registry. `queue<T>()` carries the payload
// type, while `create: true` tells setupPgBoss to create it in pg-boss.
const queues = definePgBossQueues({
  'email/send': queue<EmailJob>({ create: true }),
})

// You can also derive the queue map for asTypedPgBoss.
type Queues = PgBossQueuesFromRegistry<typeof queues>

// You keep ownership of the real PgBoss instance and its connection settings.
const boss = new PgBoss(process.env.POSTGRES_URL!)

// Bind workers through the registry so handler payloads match the queue.
const workers = [
  queues.worker('email/send', {
    name: 'email-worker',
    schedule: {
      cron: '0 8 * * *',
      data: {
        userId: 'daily',
      },
    },
    async handler(jobs) {
      for (const job of jobs) {
        job.data.userId
      }
    },
  }),
]

// Register queue definitions, schedules declared on workers, and workers.
// Set start: true only when this setup call should start pg-boss for you.
const setup = await setupPgBoss(boss, {
  start: true,
  queueRegistry: queues,
  workers,
})

// setup.boss is the same PgBoss instance, typed from the queue registry.
await setup.boss.send('email/send', {
  userId: 'user_123',
})

// If setup happens elsewhere, asTypedPgBoss gives the same instance a typed
// queue API without wrapping it.
const typedBoss = asTypedPgBoss<Queues>(boss)

await typedBoss.send('email/send', {
  userId: 'user_456',
})

Framework Integration Shape

Pass framework state through context. Worker factories receive the same context, but the registry itself stays independent from any framework.

const queues = definePgBossQueues<AppContext>()({
  'email/send': queue<EmailJob>({ create: true }),
})

const worker = queues.worker('email/send', (app) => ({
  name: 'email-worker',
  async handler(jobs) {
    app.log.info({ count: jobs.length }, 'processing email jobs')
  },
}))

const setup = await setupPgBoss(boss, {
  context: app,
  start: true,
  stopOnClose: true,
  queueRegistry: queues,
  workers: [worker],
})

app.onClose(setup.close)

Worker Error Handling

Workers can declare onError to observe handler failures with the original error and the same jobs passed to handler. The original error is rethrown after onError finishes, so pg-boss still applies its normal retry and failure behavior.

const worker = queues.worker('email/send', {
  name: 'email-worker',
  async handler(jobs) {
    for (const job of jobs) {
      await sendEmail(job.data.userId)
    }
  },
  onError(error, jobs) {
    console.error({ error, jobIds: jobs.map((job) => job.id) }, 'email worker failed')
  },
})

Typed pg-boss API

asTypedPgBoss<Queues>(boss) returns the same pg-boss object with queue-related methods narrowed to the known registry keys and payload types. The underlying runtime object is not wrapped.

When setupPgBoss receives a typed queueRegistry, the returned setup.boss is already typed from that registry.

Typed methods include send, sendAfter, sendThrottled, sendDebounced, insert, fetch, work, job commands, queue commands, queue getters, schedules, and spies.

const typedBoss = setup.boss

await typedBoss.send('email/send', { userId: 'user_123' })
await typedBoss.fetch('email/send')
await typedBoss.schedule('email/send', '0 8 * * *', { userId: 'daily' })

Unknown queue names or wrong payload shapes fail at compile time.

API Reference

queue<Data>(definition?)

Declares one queue in a registry and carries the queue payload type.

const queues = definePgBossQueues({
  email: queue<{ userId: string }>({
    create: true,
    options: {
      retryLimit: 3,
    },
  }),
  heartbeat: queue<undefined>(),
})

Options:

| Option | Type | Description | | --- | --- | --- | | create | boolean | When true, setupPgBoss creates the queue through boss.createQueue. Defaults to not creating it. | | options | Omit<Queue, 'name'> | Queue options passed to boss.createQueue(name, options) when create is true. |

queue<undefined>() and queue<null>() make send(name) valid without a data argument. Other payload types require data.

definePgBossQueues(registry)

Creates a typed queue registry. Registry keys become the allowed queue names for workers and typed boss methods.

const queues = definePgBossQueues({
  email: queue<{ userId: string }>({ create: true }),
})

type Queues = PgBossQueuesFromRegistry<typeof queues>

Use the curried form when worker factories need a framework context type:

const queues = definePgBossQueues<AppContext>()({
  email: queue<{ userId: string }>({ create: true }),
})

Returned properties:

| Property | Description | | --- | --- | | queues | The original registry object. | | definitions | Queue definitions generated from entries with create: true. | | worker(name, definition) | Binds a worker definition to a known queue name and payload type. |

setupPgBoss(boss, options?)

Registers queue definitions and workers against an existing PgBoss instance. When queueRegistry is provided, the returned setup.boss is typed from that registry.

const setup = await setupPgBoss(boss, {
  context: app,
  start: true,
  stopOnClose: true,
  stopOptions: { close: true },
  queueRegistry: queues,
  workers: [worker],
})

Options:

| Option | Type | Description | | --- | --- | --- | | context | Context | Value passed to worker factories. Required at runtime when any worker is a factory function. | | start | boolean | Starts pg-boss with boss.start() before registration. Defaults to false. | | stopOnClose | boolean | Calls boss.stop(stopOptions) from setup.close(). Defaults to false. | | stopOptions | StopOptions | Options passed to boss.stop when stopOnClose is true. | | queueRegistry | PgBossDefinedQueueRegistry | Creates queues from registry entries with create: true and types setup.boss. | | workers | readonly PgBossWorkerRegistration[] | Worker definitions or worker factories registered after queue creation. |

Return value:

| Property | Description | | --- | --- | | boss | The same PgBoss instance. Typed as TypedPgBoss<PgBossQueuesFromRegistry<typeof queueRegistry>> when a typed registry is provided. | | workers | Resolved worker definitions after applying worker factories. | | close() | Calls offWork for registered workers unless disabled, then optionally stops pg-boss. Returns Promise<void>. |

Worker Definitions

Workers can be registered directly or through queues.worker(name, definition). Using queues.worker is recommended because it ties the worker to a known queue name and payload type.

const worker = queues.worker('email', {
  name: 'email-worker',
  enabled: true,
  includeMetadata: false,
  options: {
    pollingIntervalSeconds: 1,
  },
  schedule: '0 8 * * *',
  async handler(jobs) {
    jobs[0]?.data.userId
  },
  async onError(error, jobs) {
    console.error(error, jobs.length)
  },
})

Options:

| Option | Type | Description | | --- | --- | --- | | name | string | Human-readable worker name. Also used as the queue name for plain workers that are not created with queues.worker. | | enabled | boolean | When false, registration and offWork are skipped. Defaults to enabled. | | includeMetadata | boolean | When true, the handler receives JobWithMetadata<Data>[]; otherwise it receives Job<Data>[]. | | handler | WorkHandler or WorkWithMetadataHandler | Worker handler passed to boss.work. Payload data is typed from the queue. | | onError | (error, jobs) => void \| Promise<void> | Called when handler throws. The original error is rethrown after this hook. | | options | WorkOptions | Options passed to boss.work. Put includeMetadata on the worker itself, not inside options. | | schedule | PgBossWorkerScheduleDefinition<Data> | Optional schedule created before registering the worker. | | offWorkOnClose | boolean | When false, setup.close() will not call boss.offWork for this worker. Defaults to true. | | offWorkOptions | OffWorkOptions | Options passed to boss.offWork from setup.close(). |

Worker factories receive setupPgBoss context:

const worker = queues.worker('email', (app) => ({
  name: 'email-worker',
  async handler(jobs) {
    app.log.info({ count: jobs.length })
  },
}))

Worker Schedules

worker.schedule can be a cron string or an object.

const worker = queues.worker('email', {
  name: 'email-worker',
  schedule: {
    cron: '0 8 * * *',
    data: { userId: 'daily' },
    enabled: true,
    key: 'daily-email',
    name: 'email',
    options: {},
    tz: 'UTC',
  },
  async handler() {},
})

Options:

| Option | Type | Description | | --- | --- | --- | | cron | string | Cron expression passed to boss.schedule. Required for object schedules. | | data | Data | Scheduled job payload. Typed from the queue payload. undefined becomes null for pg-boss. | | enabled | boolean | When false, the schedule is skipped. Defaults to enabled. | | key | string | Schedule key. Merged into schedule options as options.key. | | name | string | Queue name to schedule. Defaults to the worker queue. | | options | ScheduleOptions | Options passed to boss.schedule. | | tz | string | Shortcut for options.tz. |

Plain Schedule Definitions

Use registerPgBossSchedule or registerPgBossSchedules when you want to register schedules outside worker definitions.

await registerPgBossSchedule(boss, {
  name: 'email',
  cron: '0 8 * * *',
  data: { userId: 'daily' },
  enabled: true,
  key: 'daily-email',
  options: {},
})

Options:

| Option | Type | Description | | --- | --- | --- | | name | string | Queue name passed to boss.schedule. | | cron | string | Cron expression passed to boss.schedule. | | data | Data | Scheduled job payload. undefined becomes null for pg-boss. | | enabled | boolean | When false, registration is skipped. Defaults to enabled. | | key | string | Schedule key. Merged into schedule options as options.key. | | options | ScheduleOptions | Options passed to boss.schedule. |

Typed Boss Methods

TypedPgBoss<Queues> narrows queue names and payloads for these pg-boss APIs:

| Group | Methods | | --- | --- | | Send jobs | send, sendAfter, sendThrottled, sendDebounced, insert | | Read jobs | fetch, findJobs, getJobById | | Work jobs | work | | Job state | cancel, resume, retry, deleteJob, complete, fail, touch | | Queues | createQueue, updateQueue, deleteQueue, getQueue, getQueues, getQueueStats, getBlockedKeys | | Delete jobs | deleteQueuedJobs, deleteStoredJobs, deleteAllJobs | | Schedules and spies | schedule, unschedule, getSchedules, getSpy |

Use asTypedPgBoss<Queues>(boss) when you need a typed view without calling setupPgBoss with a registry:

const typedBoss = asTypedPgBoss<Queues>(boss)

Registration Helpers

Lower-level helpers are exported for framework adapters or custom lifecycles:

| Helper | Description | | --- | --- | | registerPgBossQueue(boss, queue) | Creates one queue from a string or Queue definition. | | registerPgBossQueues(boss, queues) | Creates many queues in order. | | registerPgBossSchedule(boss, schedule) | Registers one schedule unless enabled is false. | | registerPgBossSchedules(boss, schedules) | Registers many schedules in order. | | registerPgBossWorker(boss, worker) | Registers one worker and its inline schedule unless enabled is false. | | registerPgBossWorkers(boss, workers) | Registers many workers in order. | | closePgBossWorkers(boss, workers) | Calls offWork for workers unless disabled. | | getPgBossWorkerSchedule(worker) | Converts a worker schedule into a plain schedule definition. | | resolvePgBossWorkerDefinition(context, worker) | Resolves a worker definition or factory. |

Development

npm install
npm run db:up
npm test