AbsoluteJS

@absolutejs/queue-postgres

@absolutejs/queue-postgresv0.1.8betaData & Sync

Postgres (Drizzle) storage adapter for @absolutejs/queue — postgres.js + Neon serverless

#Installation

BASH
bun add @absolutejs/queue-postgres

#Capabilities

Overview

Postgres storage adapter for @absolutejs/queue, built on Drizzle. The production JobStore: durable, with atomic multi-worker claiming via FOR UPDATE SKIP LOCKED. Ships convenience factories for both postgres.js and Neon's WebSocket driver (@neondatabase/serverless); the underlying buildPostgresJobStore accepts any Drizzle Postgres database, so other drivers (including Bun SQL and node-postgres) work too. Its portable JSONB encoding preserves payload objects across all supported drivers.

Usage with Neon (@neondatabase/serverless)

createNeonJobStore mirrors createPostgresJobStore but uses Neon's WebSocket Pool. Important: the queue's claimDue opens a transaction and selects with FOR UPDATE SKIP LOCKED. Neon's HTTP driver (drizzle-orm/neon-http) is single-statement and can't do row-level locks — use the WebSocket Pool driver here. Your app's other code can keep using the HTTP driver; they're independent.

Usage with any other Drizzle Postgres adapter

Both factories are thin wrappers around buildPostgresJobStore(db, jobs), which accepts any PgDatabase from Drizzle. If you use drizzle-orm/node-postgres, build the db yourself and pass it in:

Migrations

Add the table to your Drizzle schema so it's included in migrations:

Show 1 more

Then drizzle-kit generate / push as usual.

How claiming works

claimDue runs inside a transaction:

This guarantees a job is handed to exactly one worker even with many workers polling. Crashed workers are recovered by reapStuck (lease expiry → back to pending).

Atomic claim fencing

The adapter now assigns claim_token with each claim and implements completeClaim / failClaim as conditional updates on job ID, token and status = 'claimed'. An expired, canceled or superseded attempt returns false without changing the current job. Updated queue workers use this capability automatically; requireClaimFencing: true refuses an incompatible store.

Apply this additive migration before starting the updated adapter (or generate it from the exported queueJobsTable with your application's migration tool):

Drain older workers during rollout. Older binaries still perform unfenced updates and cannot provide this guarantee. Claim fencing protects the queue row; handlers must separately fence application writes and external side effects.

Show 4 more

Lease expiry consumes one attempt. Reaping moves an exhausted job to dead with a lease-expiry error, so repeated worker death cannot bypass maxAttempts.

The /drizzle entry point (0.1.8+) does not load the optional postgres or Neon connection factories. Use it with Bun SQL or an existing Drizzle connection, including schema generation. Existing root exports remain compatible.

Mixed worker versions

With queue 0.8.4, enable requireKindFiltering: true on workers. Claims are restricted to both the adapter's job definition and the worker's registered handlers before the database applies its claim limit. Empty handler sets claim nothing. Upgrade every worker attached to a shared table before producing new job kinds; an older, unfiltered worker can still claim and reject those jobs.

Outcomes

What you can build

Build on the supported package contract

Use @absolutejs/queue-postgres through its supported public entry points.

Hardening checklist

Production guidance

Make every external boundary explicitPin the deployed @absolutejs/queue-postgres version, replace example or memory-backed dependencies with durable implementations, bound external calls, protect credentials, and emit enough evidence to retry or recover safely.

Follow in order

Troubleshooting path

1
Trace from the first failed boundary
Reproduce the smallest canonical @absolutejs/queue-postgres example, confirm the supported entry point and version in the API explorer, then inspect the first boundary that did not produce its documented result.

#Usage with postgres.js

Partial snippet

Working example for Usage with postgres.js.

TS
import { createPostgresJobStore } from '@absolutejs/queue-postgres/postgres';
import { createJobRegistry, defineJobs, queue, t } from '@absolutejs/queue';
import postgres from 'postgres';

const jobs = defineJobs({
	'email.send': t.Object({ to: t.String(), subject: t.String() })
});
const registry = createJobRegistry(jobs).on(
	'email.send',
	async ({ to, subject }) => {}
);

// Share your app's existing postgres.js client (one pool)…
const client = postgres(process.env.DATABASE_URL, { prepare: false });
const store = createPostgresJobStore({ client, jobs });

// …or let the adapter open its own connection:
// const store = createPostgresJobStore({ connectionString: url, jobs });

app.use(queue({ registry, store }));

#Usage with Neon (@neondatabase/serverless)

Partial snippet

createNeonJobStore mirrors createPostgresJobStore but uses Neon's WebSocket Pool. Important: the queue's claimDue opens a transaction and selects with FOR UPDATE SKIP LOCKED. Neon's HTTP driver (drizzle-orm/neon-http) is single-statement and can't do row-level locks — use the WebSocket Pool driver here. Your app's other code can keep using the HTTP driver; they're independent.

TS
import {
	createNeonJobStore,
	neonConfig
} from '@absolutejs/queue-postgres/neon';
import { createJobRegistry, defineJobs, queue, t } from '@absolutejs/queue';
import { Pool } from '@neondatabase/serverless';

// Bun ships a global WebSocket; for node, polyfill once:
//   import ws from 'ws'; neonConfig.webSocketConstructor = ws;

const jobs = defineJobs({
	'email.send': t.Object({ to: t.String(), subject: t.String() })
});

// Share an existing Neon Pool…
const pool = new Pool({ connectionString: process.env.DATABASE_URL });
const store = createNeonJobStore({ jobs, pool });

// …or let the adapter open its own:
// const store = createNeonJobStore({ connectionString: url, jobs });

#Migrations

Partial snippet

Add the table to your Drizzle schema so it's included in migrations:

TS
export { queueJobsTable } from '@absolutejs/queue-postgres/drizzle';

#How claiming works

Partial snippet

claimDue runs inside a transaction:

SQL
SELECT … FROM queue_jobs
WHERE status = 'pending' AND run_at <= $now
ORDER BY run_at LIMIT $n
FOR UPDATE SKIP LOCKED;        -- concurrent workers skip locked rows
-- then UPDATE those ids → status='claimed'

#Public entry points

Supported entry points declared by this package manifest.

Package entry point declared in package.json.

@absolutejs/queue-postgres@absolutejs/queue-postgres/postgres@absolutejs/queue-postgres/neon@absolutejs/queue-postgres/manifest@absolutejs/queue-postgres/manifest.json@absolutejs/queue-postgres/drizzle

#Package commands

Scripts declared by this package manifest.

bun run buildrm -rf dist && bun build src/index.ts src/drizzle.ts src/postgresJobStore.ts src/neonJobStore.ts src/manifest.ts --outdir dist --sourcemap --target=bun --external drizzle-orm --external postgres --external @neondatabase/serverless --external @absolutejs/queue && tsc --project tsconfig.build.json && absolute-manifest emit
bun run formatprettier --write "./**/*.{ts,json,md}"
bun run testbun test
bun run test:pgQUEUE_TEST_DATABASE_URL=postgresql://postgres:postgres@127.0.0.1:55444/queue_test bun test
bun run typechecktsc --noEmit