`@octabits-io/queue`

Durable background jobs on the Postgres you already run. Payloads are Zod-validated when they are enqueued and again when a worker picks them up. Each job in a batch is acked on its own. Failures retry into a dead-letter queue. A declarative definition turns a name, a schema and a handler factory into an enqueuer, a worker and a dead-letter handler. A job can be enqueued on the caller’s own transaction, so it exists only if the write that produced it commits. And there is a dead-letter audit sink: a contract with an in-memory reference, a plain-SQL store and a Drizzle adapter. pg-boss is the runtime, behind its own subpath; the logger, the tracer and the meter are structural seams.

npm i @octabits-io/queue zod pg-boss     # + drizzle-orm for ./drizzle

Subpath	Runs on	Deps	Contents
`.`	any runtime	`zod`	the contract: `JobContext`, `JobHandler`, `QueueDomain`, `EnqueueOptions`, the error types, `SCHEMA_BASE_JOB_PAYLOAD` / `SCHEMA_SYSTEM_JOB_PAYLOAD` / `SCHEMA_SCOPED_JOB_PAYLOAD`, `JOB_TRACE_KEY`, `QueueScope` / `QueueScopeFactory`, the monitoring types and error factories, the structural `Logger` / `Tracer` / `Meter` seams, and the dead-letter audit contract (`JobAuditRecord`, `JobAuditStore` with `record` / `list`, `JobAuditPage` / `JobAuditEntry`, `createJobAuditStore`, `createInMemoryJobAuditStore`, `DlqAuditRecord` / `DlqAuditSink`)
`./pg-boss`	Node, Bun	`pg-boss` (optional peer)	`createBossManager` (lifecycle + monitoring, with a `role` per process: `full`, `consumer`, `producer`, `migrator`), `installPgBossSchema` + `pgBossGrantsSql` (the owner’s migration step of a two-role deployment), `createQueueDomain` (queue/worker/DLQ trio), `defineQueue` (the declarative layer)
`./postgres`	any runtime	none (structural)	`createPostgresJobAuditStore` over any `query(text, params)` executor (`pg`, PGlite, postgres.js), `jobAuditDdl()`
`./drizzle`	Node, Bun	`drizzle-orm` (optional peer), `@octabits-io/postgres` (peer)	`jobAuditColumns` (timestamps read back as ISO-8601 in UTC) + `createDrizzleJobAuditStore` — the same audit store for a host that already runs Drizzle

The only hard dependency is @octabits-io/result. A boundary lint confines pg-boss to ./pg-boss and drizzle-orm to ./drizzle, and refuses @octabits-io/telemetry and every sibling package. A type check with no Node typings keeps the root and ./postgres runtime-agnostic.

When not to use it

Live updates to a browser. A job runs once, later, on one worker; it does not reach the people looking at the page. Telling subscribers that something changed is an event — @octabits-io/events, emitted in the same transaction.
A multi-step workflow. Steps that depend on each other, durable sleeps, signals, compensation: that is a workflow engine (octaflow), which runs on the same pg-boss.
Throughput beyond what your Postgres takes. Jobs are rows. Thousands a second is fine; a firehose that would dominate the database’s write load wants a broker.
Strict ordering or exactly-once side effects. Delivery is at-least-once with retries; a handler must be idempotent, and jobs are not processed in enqueue order across workers.
A database that is not Postgres. The contract is neutral, but the only runtime is pg-boss, and pg-boss is Postgres. That is the point, not an accident.

The shape

 producer                          Postgres (pg-boss schema)            worker process
 ────────                          ─────────────────────────            ──────────────
 enqueue(payload, { db?, parent?, key? })
   │ validate against the schema
   │ ensure queue + DLQ (pool, once)
   └─ INSERT job ──────────────────▶ <queue> ──── fetch batch ──▶ validate again
      (on `db` when given:                                        handler(job) → Result
       the caller's transaction;       │ failed, retries left ◀── err / throw
       with `key`: one waiting
       job per key, upserted)
                                       │ retries exhausted
                                       ▼
                                     <queue>-dlq ─── DLQ handler ─▶ onDlq(scope, job)
                                                                   onDlqAudit(scope, record) ─▶ JobAuditStore

A queue in forty lines

import { z } from 'zod';
import { ok, err } from '@octabits-io/result';
import { SCHEMA_SCOPED_JOB_PAYLOAD } from '@octabits-io/queue';
import { createBossManager, defineQueue } from '@octabits-io/queue/pg-boss';
import { createPostgresJobAuditStore } from '@octabits-io/queue/postgres';

const SCHEMA_EXPORT_JOB = SCHEMA_SCOPED_JOB_PAYLOAD.extend({ requestId: z.uuid() });
type ExportJob = z.infer<typeof SCHEMA_EXPORT_JOB>;

const audit = createPostgresJobAuditStore({ db: pool, scope: { column: 'org_id' } });

export const exportQueue = defineQueue<ExportJob, AppServices>({
  name: 'export',                                         // the DLQ is `export-dlq`
  schema: SCHEMA_EXPORT_JOB,
  config: { retryLimit: 3, retryDelay: 30, expireInSeconds: 300 },
  resolveScopeKey: (data) => data.scopeKey,               // binds the DLQ scope + the audit row's scope
  createHandler: ({ createSystemScope, logger }) => async (job) => {
    const scope = await createSystemScope(job.data.scopeKey);
    try {
      const done = await scope.resolve('exports').run(job.data.requestId);
      return done.ok ? ok(undefined) : err({ key: 'job_failed', message: done.error.message, jobId: job.id });
    } finally {
      await scope.dispose();
    }
  },
  onDlqAudit: audit.onDlqAudit,
});

const manager = createBossManager({ connectionString: process.env.DATABASE_URL!, logger });
await manager.start();
const boss = manager.getBoss();

await exportQueue.createWorker({ boss, logger }).startWorker({ createSystemScope });
await exportQueue.createDlqHandler({ boss, createSystemScope, logger }).start();
const { enqueue } = exportQueue.createEnqueuer({ boss });
await enqueue({ scopeKey: 'org_1', requestId: crypto.randomUUID() });

createSystemScope is any (scopeKey?) => Promise<{ resolve(key), dispose() }> — a DI container’s child scope, or a plain object in a test. logger is any object with debug / info / warn / error(message, error?, attributes?) / child. createQueueDomain is the layer underneath defineQueue when you want the trio without the declarative wrapper.

What the queue guarantees, so a handler does not have to:

Validation both ways. A bad payload on enqueue returns a PayloadValidationError with the Zod issues. A bad payload that reaches a worker anyway — a foreign producer, a corrupt row — goes straight to the DLQ on its first attempt, because a retry cannot fix it.
Per-job acking. One failing job in a batch neither fails nor re-runs its batch-mates, and job.retryCount is the real count.
Errors are values. Every method returns a Result; a handler returns err(...) to fail a job. A throw is caught and treated the same.
Permanent failures skip the retries. A handler that returns job_failed with permanent: true — a refused credential, an address the provider rejects — goes to the DLQ on that attempt instead of spending retryLimit on something a retry cannot fix.
Queues exist before use. The queue and its DLQ are created and their settings synced on first use, memoized per domain; a failed ensure is retried, not cached.

Enqueue on the caller’s transaction

A job for a write belongs to that write: if the order insert rolls back, the confirmation mail must not go out. Pass the transaction’s connection as db:

import { fromDrizzle } from 'pg-boss';
import { sql } from 'drizzle-orm';

await db.transaction(async (tx) => {
  await tx.insert(orders).values(order);
  const queued = await enqueue({ scopeKey, orderId: order.id }, { db: fromDrizzle(tx, sql) });
  if (!queued.ok) throw new Error(queued.error.message);   // rolls the order back too
});

db is structurally pg-boss’s Db — one method, executeSql(text, values) → { rows } — so pg-boss’s own adapters (fromDrizzle, fromKysely, fromKnex, fromPrisma) fit as-is, and so does a two-line wrapper over a pg client inside BEGIN:

const client = await pool.connect();
await client.query('BEGIN');
await enqueue(payload, { db: { executeSql: (text, values) => client.query(text, values) } });
await client.query('COMMIT');

Only the job’s INSERT runs on that connection. The queue and its DLQ are ensured first, on the queue’s own pool and outside the transaction: creating a queue is DDL, a rollback would undo it, and its locks could outlive the send. The ensure is memoized, so a transactional producer pays it once per process. The integration suite proves it against a real Postgres:

A rolled-back transaction leaves no job.
A committed one leaves exactly one.
A rollback on a queue’s first-ever enqueue leaves the queue itself in place.

On a database with one connection (PGlite), call ensureQueue() at boot, before any request can open a transaction and enqueue inside it. The ensure also reads the queue into pg-boss’s queue cache, so the send inside the transaction only runs its INSERT. A send that has to look the queue up would wait on the one connection the transaction holds. pglite-transaction.test.ts holds this.

ensureQueue() stays public for a producer that writes jobs some other way.

One job per record

The same work asked for twice while it is still waiting — two requests for one record, a retried HTTP call, a restart that re-sends its start-up work — should stay one job. Pass a key the producer owns:

const queued = await enqueue({ scopeKey, checkId }, { key: `check-${checkId}` });
// queued.value: { jobId, queue, deduplicated }

With a key the send is pg-boss’s upsert by singletonKey, one statement pair that is atomic under concurrent producers:

The newest job with that key is…	What happens	`deduplicated`	`jobId`
none, or `completed` / `cancelled` / `failed`	a new job is inserted	`false`	the new job
`created` or `retry`	that job takes this payload and keeps its id, position and retry count	`true`	that job
`active`	left alone — the running attempt keeps the payload it fetched — and a new job is inserted behind it	`false`	the new job

A singletonKey on a plain send dedupes nothing on a standard queue: pg-boss rejects a second job through its per-policy unique indexes (short, exclusive, stately), and a queue’s policy is immutable. The upsert works on every policy, so the package never asks you to choose one. A host that created its queue as key_strict_fifo wants every job kept and ordered, not folded: enqueue there without key and set pg-boss’s own singletonKey through the boss. The payload is still validated and the trace still rides along; a deduplicated job carries the later request’s traceparent.

The same key is the handle for what a host does later, without a pg-boss id stored beside its record:

const live = await manager.findJobByKey('check', `check-${checkId}`);
// ok(JobDetails | null): the newest job that is still created, retry or active, with its state

const done = await manager.cancelByKey('check', `check-${checkId}`);
// ok({ cancelled }): every created / retry job with that key; an active one is left to finish

cancelByKey reads the waiting jobs and then cancels them, two statements: a job a worker claims in between is cancelled the way cancelJob would cancel it. Inside a caller’s transaction (db) a keyed enqueue rolls back with it like any other. dedupe-key.test.ts holds the table above on PGlite; the integration suite holds the lookup, the cancel and the rollback on a real Postgres.

Traces across the queue

Pass telemetry: { tracer, meter? } to createQueueDomain, or to defineQueue’s createWorker / createEnqueuer, and every job runs inside a consumer span named <queue> process, handed to the handler as job.span. With a meter, messaging.process.duration and messaging.client.consumed.messages count every job by queue and error.type. enqueue(payload, { parent }) writes the caller’s W3C traceparent into the payload under JOB_TRACE_KEY ($traceparent) — pg-boss has no headers — and the worker strips it before the payload meets the schema, so the job’s span is a child of the request that enqueued it. With telemetry on the enqueuer too, the enqueue itself is a producer span between the two.

Tracer, Span, Meter and Logger are structural and declared here. @octabits-io/telemetry satisfies them unchanged, and a test pins that. A host on the OpenTelemetry SDK adapts in a dozen lines.

The dead-letter audit sink

Every dead-lettered job is logged first, then run through the optional onDlq(scope, job, data) hook, then handed to onDlqAudit(scope, record). JobAuditRecord is discriminated on validPayload:

true: the typed payload, and attemptCount is the retry limit it exhausted — or, for a permanent failure (errorMessage says so), the attempts it made.
false: a schema-invalid payload, dead-lettered on its first attempt, with rawPayload: unknown and attemptCount: 1.

A sink that throws is caught and logged, and the scope is always disposed. DlqAuditRecord, DlqAuditRecordBase and DlqAuditSink are the same types under the names defineQueue has always used.

A store is the contract plus two primitives, write(row) and read(query):

import { createInMemoryJobAuditStore } from '@octabits-io/queue';
import { createPostgresJobAuditStore, jobAuditDdl } from '@octabits-io/queue/postgres';
import { createDrizzleJobAuditStore, jobAuditColumns } from '@octabits-io/queue/drizzle';

jobAuditDdl({ scope: { column: 'org_id' } });                         // the table, for your migrations
const sqlStore = createPostgresJobAuditStore({ db: pool, scope: { column: 'org_id' } });
const ormStore = createDrizzleJobAuditStore({ db, table: jobAuditLog, scope: { column: 'orgId' } });

store.record(record) returns Result<'recorded' | 'skipped_unscoped'>. store.onDlqAudit is the same write shaped for the sink seam, and it throws on a storage failure: the seam returns void, and a swallowed error would be a silently lost audit trail. The scoping rules live in the contract, once:

`scope`	Behaviour
omitted	no scope column stamped; every record written
`{ column }`	stamped from `record.scopeKey`; a record without one (a system queue that omits `resolveScopeKey`) is skipped → `skipped_unscoped`
`{ column, value }`	the fixed value on every row; `record.scopeKey` ignored

Reading it back

store.list(options?) returns Result<JobAuditPage>: the dead letters, newest first, one page at a time. Read through the store rather than querying the table: the scoping rules, the cursor and the read form of a row are the contract’s, the same on every adapter.

Option	Meaning
`queueName`	only rows for this queue
`scopeKey`	only rows stamped with this value — honoured by a `{ column }` store alone (below)
`limit`	page size; default 50, clamped to 1..200
`before`	keyset cursor: only rows with `id < before` — the previous page’s `nextBefore`

A page is { items: JobAuditEntry[], nextBefore: number | null }. nextBefore is null on the last page; the store fetches one row more than the page to know. A JobAuditEntry is the row as written plus what the table assigns: id (a number) and createdAt. Both timestamps read back in toISOString() form (UTC, milliseconds), whether the driver hands over a Date or Postgres’ text form in any session time zone. payload comes back as stored: an object, null, or a top-level JSON string that stays a string — the failed job’s data exactly as it was enqueued, names and addresses included where the job carried them, so a page that leaves the server goes through the same projection or redaction the job’s own records would. scopeValue is on an entry only when the store has a scope column. Only status = 'dead_letter' rows are read, so a table shared with in-flight job states lists its dead letters alone.

The scoping rules for reads mirror the write rules:

`scope`	`list` shows
omitted	every row; `options.scopeKey` ignored
`{ column }`	the rows stamped with `options.scopeKey` when given, every row when omitted
`{ column, value }`	the rows stamped with `value`, always; `options.scopeKey` ignored

A { column } store cannot tell an operator’s view from a scoped request. On a scoped request, pass the key, build a { column, value } store for that request, or query as a role under a row-level-security policy on the scope column. The policy filters whatever the caller passes.

A dead-letter page, with the worker’s store writing and a store per request reading one scope:

// The worker: one store for every scope, stamping each record's scopeKey.
const audit = createPostgresJobAuditStore({ db: pool, scope: { column: 'org_id' } });

// GET /jobs?queue=export&before=812 — a store fixed to the caller's scope, so no option widens it.
export async function listDeadLetters(request: Request, scopeKey: string): Promise<Response> {
  const store = createPostgresJobAuditStore({ db: pool, scope: { column: 'org_id', value: scopeKey } });
  const params = new URL(request.url).searchParams;
  const before = params.get('before');
  const page = await store.list({ queueName: params.get('queue') ?? undefined, before: before ? Number(before) : undefined });
  if (!page.ok) return Response.json({ error: page.error }, { status: 500 });
  // Project before it leaves the server: payload and errorMessage are the job's own data.
  return Response.json({ nextBefore: page.value.nextBefore, items: page.value.items.map(({ id, jobId, queueName, completedAt }) => ({ id, jobId, queueName, completedAt })) });
}

A hand-written store implements list too. createJobAuditStore({ scope, write, read }) gives it the rules for free: read(query) receives the resolved filter, the page size plus one and the cursor, and returns the rows as the driver hands them over.

One conformance suite runs the in-memory, plain-SQL and Drizzle stores against an embedded PGlite. A new adapter passes it or it is not an adapter. jobAuditDdl() and jobAuditColumns declare the same table column for column. The Drizzle payload column is a jsonb that does not re-parse a top-level JSON string, so a raw "73235" payload stays a string. In-memory stores share one table through createInMemoryJobAuditStore({ sameTableAs }), the way a writer and a reader share one in Postgres.

Two database roles

A deployment with one database role starts pg-boss as role: 'full' and is done: that process installs the pgboss schema, migrates it on an upgrade, consumes, supervises maintenance and runs the cron timekeeper. A deployment with two roles — an owner that runs the migrations once, and an application role that is NOSUPERUSER NOBYPASSRLS with no CREATE and runs the web processes and the worker (infra/single-host/init/01-roles.sh) — splits that across processes by role:

`role`	Installs / migrates	Consumes	Maintenance + timekeeper	Runs as
`'full'`	yes, on `start()`	yes	yes	the owner
`'migrator'`	yes, on `start()`, then waits for the index builds it queued	no	no	the owner, in the migration step
`'consumer'`	never — `start()` checks and refuses	yes	yes	the application role
`'producer'`	never — `start()` checks and refuses	no	no	the application role

The order of start-up is the owner’s migration step, then the application role’s processes:

// 1. The migration step, as the owner (DATABASE_MIGRATION_URL): the schema …
const { version } = await installPgBossSchema({ connectionString: ownerUrl });

// 2. … the queues, which are DDL too (a table per queue) …
const migrator = createBossManager({ connectionString: ownerUrl, role: 'migrator', logger });
await migrator.start();
for (const queue of [exportQueue, mailQueue]) await queue.createEnqueuer({ boss: migrator.getBoss() }).ensureQueue();
await migrator.stop();

// 3. … and the grants, after the queues, with the owner's SQL client.
for (const statement of pgBossGrantsSql('my_app')) await owner.query(statement);

// 4. The worker, as the application role (DATABASE_URL): consumes, never migrates.
const worker = createBossManager({ connectionString: appUrl, role: 'consumer', logger });
await worker.start();

// 5. The web process, as the application role: sends only.
const web = createBossManager({ connectionString: appUrl, role: 'producer', logger });
await web.start();

What each start() does and refuses:

'migrator' installs a missing schema or migrates an outdated one, with pg-boss’s own advisory lock against a second migrator. A migration hands its index builds (CREATE INDEX CONCURRENTLY on the job tables) to pg-boss’s background maintenance, which only a process with migration rights runs — a cluster of consumers would leave them pending forever — so the migrator’s start() waits until none is pending and fails the step, naming the statements, if one failed (indexBuildTimeoutMs, default ten minutes, bounds the wait). It starts no worker, no maintenance sweep and no timekeeper. installPgBossSchema() is that start, a read of the version reached and a stop, in one call; a schema already at the current version is left as it is, so the step is idempotent.
'consumer' consumes and runs maintenance and schedules exactly as 'full' does. Its start() is a check: a schema that is missing or behind fails with pg-boss cannot start with role 'consumer': pg-boss is not installed. A consumer never installs or migrates the schema — run installPgBossSchema() (or a process with role 'full') as the schema owner against this database first., never with permission denied halfway through DDL; pg-boss’s message stays as the cause.
'producer' sends; the same check, the same error.
'full' does everything; it is the one role that must run as the owner.

pgBossGrantsSql(appRole, { schema }) is the four statements the application role needs — USAGE on the schema, SELECT, INSERT, UPDATE, DELETE on all its tables, USAGE, SELECT on its sequences, EXECUTE on its functions — as an array with no trailing semicolons, so a client that runs one statement per call and one that takes a script are both served; the role and the schema are checked to be plain identifiers. Nothing in it grants CREATE. Run it after the queues are created: ON ALL TABLES covers the tables that exist when it runs, and a queue is a table of its own, so a release that adds a queue runs the same statements again in its migration step. apps/starter grants its app role with the same function (src/server/app.ts).

roles.integration.test.ts proves it against a Postgres with the two roles from the single-host init script: a consumer as the application role before the install fails with the error above; installPgBossSchema() as the owner returns the version and leaves no index build pending; a second run changes nothing; after the queues and the grants, a consumer and a producer as the application role — which still has no CREATE on pgboss — send, work and complete a job and keep a schedule.

Moved here

The queue lived in @octabits-io/framework/queue from July to 2026-09-13, and the audit contract in @octabits-io/postgres/job-audit-store. Each old path was a named re-export shim for one minor and has been removed:

Was	Is
`@octabits-io/framework/queue`	`@octabits-io/queue` (contract) + `@octabits-io/queue/pg-boss` (runtime)
`@octabits-io/framework/drizzle/job-audit-store`	`@octabits-io/queue/drizzle`
`@octabits-io/postgres/job-audit-store`	`@octabits-io/queue` + `@octabits-io/queue/postgres`

Stability

Pre-1.0: patches never break, and a minor says in the changelog what it breaks. The payload schemas, JOB_TRACE_KEY and the DDL helpers are part of the surface. A field a schema adds is a field every producer sends, and a column a DDL helper emits is a column a store writes.