Queue
Durable typed job queue for Bun + Elysia. Define jobs once via TypeBox schemas, validate payloads at the boundary, dispatch by kind, and run on the store that fits your deployment — in-memory, Postgres, or Redis. At-least-once delivery with stuck-lease reap, exponential-backoff retries, dead-lettering on exhausted attempts, and an OpenTelemetry span per run.
This page covers the quick start and the storage layer. For defining work — schemas, handlers, retries, recurring and one-shot triggers — see Queue Jobs. For running it — the worker, admin routes, metrics, and tracing — see Queue Operations.
#Quick Start
Four moves: define jobs, register handlers, pick a store, wire the Elysia plugin. queue.enqueue(kind, payload) is decorated onto the Elysia context so any route can dispatch.
import { Elysia } from 'elysia';
import {
createInMemoryJobStore, createJobRegistry,
defineJobs, queue, t
} from '@absolutejs/queue';
// 1. Define jobs once — the single source of truth for payload types
// AND runtime validation.
const jobs = defineJobs({
'email.send': t.Object({
to: t.String({ format: 'email' }),
subject: t.String(),
body: t.String(),
}),
'image.resize': t.Object({
src: t.String({ format: 'uri' }),
width: t.Number({ minimum: 1 }),
}),
});
// 2. Register handlers by kind — payloads are inferred from the schema.
const registry = createJobRegistry(jobs)
.on('email.send', async (payload) => {
await dispatchEmail(payload);
})
.on('image.resize', async ({ src, width }) => {
await resize(src, width);
});
// 3. Pick a store + wire the Elysia plugin (in-process worker
// auto-starts).
const store = createInMemoryJobStore(jobs);
const app = new Elysia().use(queue({ registry, store }));
// 4. Enqueue from any route.
app.post('/send', ({ queue }) =>
queue.enqueue('email.send', {
to: 'user@example.com',
subject: 'hi',
body: 'hi',
}),
);#Stores & Adapters
The worker contract is small — claim due jobs, complete, fail (with retry or dead-letter), reap stuck leases. Optional methods power the admin tooling.
type JobStore<Jobs> = {
enqueue(input): Promise<JobId>; // validates; idempotencyKey dedup
claimDue(options): Promise<Job[]>; // atomic multi-worker claim
complete(id): Promise<void>;
fail(id, options): Promise<void>; // retryAt OR dead
reapStuck(options): Promise<number>; // expired leases → pending
// Optional — powers the admin routes (501 when missing):
cancel?(id); retry?(id); get?(id);
list?(options); listByKind?(kind, options);
countByStatus?();
};One store ships with the core package; two adapter packages cover durable deployments. Pick by deployment model:
| Store | Package | Version | Best for |
|---|---|---|---|
createInMemoryJobStore | @absolutejs/queue | 0.7.1 | Tests and single-process apps. snapshot()/restore() round-trips state so the host can persist on SIGTERM and rehydrate on restart. |
createPostgresJobStore | @absolutejs/queue-postgres | 0.1.4 | The production default — Drizzle + postgres.js against the Postgres you already run, with atomic multi-worker claims via FOR UPDATE SKIP LOCKED. |
createNeonJobStore | @absolutejs/queue-postgres | 0.1.4 | Neon serverless Postgres — same store over Neon's WebSocket Pool driver, since claimDue needs real transactions and row-level locks. |
createRedisJobStore | @absolutejs/queue-redis | 0.0.3 | High-throughput, low-latency claims — atomic Lua scripts keep the critical section to one round-trip. |
#Postgres Adapter
@absolutejs/queue-postgres is a Drizzle-based implementation of JobStore with convenience factories for postgres.js and Neon's WebSocket driver — and buildPostgresJobStore underneath, which accepts any Drizzle Postgres database if you're on another driver.
import { createPostgresJobStore } from '@absolutejs/queue-postgres/postgres';
import postgres from 'postgres';
// 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 });
// Neon — WebSocket Pool driver. claimDue opens a transaction and
// selects FOR UPDATE SKIP LOCKED; neon-http is single-statement and
// can't do row-level locks, so the queue uses the Pool. Your app's
// other code can keep using the HTTP driver — they're independent.
import { createNeonJobStore } from '@absolutejs/queue-postgres/neon';
const serverless = createNeonJobStore({
connectionString: process.env.NEON_DB_URL,
jobs,
});The adapter implements the full JobStore plus every optional method (cancel / retry / list / get / listByKind / countByStatus), so the admin routes light up automatically. The schema is exported as queueJobsTable if you want to manage migrations alongside your own.
#Redis Adapter
@absolutejs/queue-redis claims and reaps via atomic Lua scripts, so two workers can't race-claim the same job. The narrow RedisCommandClient contract means ioredis and node-redis v4+ both work — no client peer dep.
import type { JobMapFromDefinition } from '@absolutejs/queue';
import { createRedisJobStore } from '@absolutejs/queue-redis';
import { Redis } from 'ioredis';
// No client peer dep — ioredis and node-redis v4+ both structurally
// satisfy the narrow RedisCommandClient contract.
const store = createRedisJobStore<JobMapFromDefinition<typeof jobs>>({
client: new Redis(process.env.REDIS_URL),
keyPrefix: 'myapp:queue:', // default 'absolutejs:queue:'
});| Key | Type | Purpose |
|---|---|---|
<prefix>job:<id> | HASH | One record per job. |
<prefix>due | ZSET | Scheduling — jobs sorted by runAt. |
<prefix>claimed | ZSET | Active leases keyed by lockedAt + leaseMs. |
<prefix>idempotency:<key> | STRING | idempotencyKey → job id, set with NX for dedup. |
<prefix>kind:<kind> | SET | Job ids by kind — lazy index. |
Choose Redis for high-throughput claims and low-latency dispatch — it also pairs naturally with @absolutejs/sync-bus-redis if you're already running Redis for the sync cluster bus.