Persistence

Redis holds live queue state, and BullMQ's retention is bounded — completed jobs age out, and a FLUSHALL takes your history with it. Durable Postgres storage gives OpenQueue a paper trail: run history with full payloads and event timelines, dynamic schedules that survive restarts, and alert configuration. Workbench reads all of it when present.

Two ways to reach Postgres

Postgres is a store — the durable-state axis. You reach it two ways, and the choice is about who owns the schema:

• postgresAdapter (this page). You bring your own Drizzle db and run the migrations yourself with drizzle-kit. It composes alongside a transport — typically the redis sugar (BullMQ delivers, Postgres persists). This is the power-user path: OpenQueue's tables live in your database, on your migration history, sharing your connection pool. The demo spells the composition out — worldBullmq({ url, storage: postgresAdapter({ db, schema }) }).

• worldPostgres (Worlds). A self-migrating, zero-Redis world: one Postgres connection runs the queue and stores history, and the world owns its own schema — it applies its versioned migrations on world.start() under an advisory lock, so you never run drizzle-kit. Use it when you want a single-dependency deploy.

  import { defineConfig } from '@openqueue/sdk';
  import { worldPostgres } from '@openqueue/world-postgres';
  
  export default defineConfig({
    namespace: 'my-app',
    dirs: ['./worker'],
    world: worldPostgres({ url: process.env.DATABASE_URL!, migrations: 'auto' }),
  });

The rest of this page covers the postgresAdapter path in detail.

Quick setup

bunx openqueue add persistence

This scaffolds src/queue-schema.ts and drizzle.config.ts, adds drizzle-orm and drizzle-kit to your package.json, and prints the storage block to paste into your worker config. It never overwrites: existing files are left untouched (with merge hints instead), and re-running it just reports the current state. The sections below explain everything it sets up.

The tables

defineQueueSchema() returns plain Drizzle pg-core tables, so they ride your existing drizzle-kit setup — no separate migration tool.

Table names are deliberately short (runs, catalog, schedules), so put them in a dedicated Postgres schema instead of public:

import { defineQueueSchema } from '@openqueue/sdk';

export const queueSchema = defineQueueSchema({ schema: 'jobs' });

export const {
  queueCatalog,
  queueSchedules,
  queueScheduleInstances,
  queueRuns,
  queueRunEvents,
  alertChannels,
  alertRules,
} = queueSchema;

Re-exporting each table individually is what lets drizzle-kit discover them.

Generating migrations

Point a drizzle-kit config at the file and scope it to the queue schema:

import { defineConfig } from 'drizzle-kit';

export default defineConfig({
  dialect: 'postgresql',
  schema: './src/queue-schema.ts',
  out: './drizzle',
  dbCredentials: {
    url: process.env.DATABASE_URL!,
  },
  schemaFilter: ['jobs'],
  migrations: {
    table: '__drizzle_migrations',
    schema: 'jobs',
  },
});

bunx drizzle-kit generate   # emits CREATE SCHEMA "jobs" + all tables
bunx drizzle-kit migrate

schemaFilter keeps drizzle-kit from touching your application tables; the CREATE SCHEMA IF NOT EXISTS "jobs" statement is generated automatically. If your app already has a drizzle-kit config, you can instead add ./src/queue-schema.ts to its schema array and 'jobs' to its schemaFilter — one migration history for everything.

Wiring the adapter

Pass the same schema object (or just the schema name) to postgresAdapter in your worker config:

import { defineConfig, postgresAdapter } from '@openqueue/sdk';
import { db } from './src/db';
import { queueSchema } from './src/queue-schema';

export default defineConfig({
  namespace: 'my-app',
  dirs: ['./worker'],
  redis: { url: process.env.REDIS_URL! },
  storage: postgresAdapter({ db, schema: queueSchema }),
});

db is your existing Drizzle database instance — the adapter issues plain Drizzle queries through it, so it shares your pool, your transactions config, and your connection lifecycle.

What gets written

Every run lifecycle event flows through the adapter: the runs row is upserted with the latest status, and a run_events row is appended with the event payload — so you get both "where is this run now" and "what happened, in order". Failures store the serialized error; ctx.progress() patches land in metadata.

Runs also carry meta from trigger time. OpenQueue treats it as opaque application data, and indexes the JSON document for containment filters:

await exportCsv.trigger(payload, {
  meta: { tenantId: tenant.id, customerId: customer.id, tags: ['billing'] },
});

Tags are stored separately for tag filters. Workbench's run filters (task, status, schedule, meta, tags, time range) map directly onto generic queue state rather than application-specific columns.

Retention

Durable run history is pruned automatically. Each window counts from the run's finish time — a run that never finished (still queued or executing) is never pruned — and false keeps that category forever:

export default defineConfig({
  // ...
  retention: {
    completed: 30, // days to keep completed/canceled runs (default)
    failed: 90, // days to keep failed runs (default)
    logs: 30, // days to keep run events + spans (default)
  },
});

The defaults follow trigger.dev: 30 days of logs is their Pro-plan default, with the retention timer starting when the run completes. Run records are additionally capped (30/90 days) because self-hosted storage is yours to pay for — opt out per field with false if you'd rather keep them.

The worker sweeps hourly (plus once shortly after boot): aged runs go by their bucket's window, and events/spans go when older than logs or orphaned by a pruned run. One [openqueue] retention: pruned … line is logged when anything was deleted. The sweep runs through the store's optional prune capability, so custom stores without it are left untouched.

The sweep is built for production backlogs: deletes run in bounded batches (up to ~200k rows per category per sweep), so a first sweep over millions of rows stays cheap and the hourly cadence drains the rest incrementally. Replicas sharing a database coordinate through a Postgres advisory lock — exactly one prunes per sweep — and the timers carry ±10% jitter so a fleet doesn't fire in lockstep.

The worker app wires the sweep up for you. Embedded runtimes don't get it automatically: if you boot createQueueWorker(...) yourself, pass retention in its options to opt in — without it, no retention runs. createRetentionSweeper(runtime.runs, resolveRetentionPolicy({ ... })) is the manual escape hatch when you need to own the cadence.

This is separate from BullMQ's delivery-layer retention (removeOnComplete/removeOnFail, 7/30 days), which already trimmed finished jobs out of Redis — that bounds live queue state, while retention bounds the durable paper trail.

Suggested patterns

• One schema object, two consumers. Define queueSchema once and hand it to both drizzle-kit (migrations) and postgresAdapter (runtime). A schema-name string in one place and an object in the other drifts eventually.

• Storage is required for dynamic schedules and alerts. Without an adapter, task.schedules.create() throws and alert rules have nowhere to live. Declarative cron tasks work either way.

• Age-based pruning is built in. Retention caps run history at 30/90/30 days out of the box. Reach for your own cron task only for criteria the age-based sweep can't express — per-tenant windows, or archiving to cold storage before deletion (run_events cascades on run delete either way).

• Worker pools share one set of tables. Point every pool's adapter at the same schema — run history is keyed by run id, and schedules deduplicate. The catalog table mirrors the Redis catalog's last-writer-wins behavior across pools (see Scaling).