AbsoluteJS

@absolutejs/queue-postgres

@absolutejs/queue-postgresv0.1.4betaData & 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).

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';

#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

#Package commands

Scripts declared by this package manifest.

bun run buildrm -rf dist && bun build src/index.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

#API reference

Search the declarations exported by the current package type files. Expand a symbol to inspect its source-backed signature.

10 symbols
createNeonJobStoreexportPermalinkSource
TS
createNeonJobStore
Exported from @absolutejs/queue-postgres