A TypeScript job queue library backed by PostgreSQL. Jobs are stored as rows, locked via PostgreSQL functions, and processed by workers that support both polling and real-time NOTIFY/LISTEN.
- Durable job storage in PostgreSQL (13+)
- Pattern-based job routing to handlers
- Parallel workers with configurable concurrency
- Two processing modes: polling and PostgreSQL
NOTIFY/LISTEN - Automatic retries with configurable lock durations
- Batch job creation and acquisition
- Typed event system for observability
- SQL migrations managed automatically via Umzug
- Node.js 18+ or Bun
- PostgreSQL 13+
- TypeScript 5+
npm install pollocks pg
pg is a peer dependency. You provide your own Pool instance.
pollocks manages its own schema. Call migrate() once at startup:
importpgfrom"pg";import{Tools}from"pollocks";constpool=newpg.Pool({connectionString: "postgres://user:pass@localhost:5432/mydb",});consttools=newTools(pool);awaittools.migrate();const{ id }=awaittools.addJob({pattern: "send-email",payload: {to: "user@example.com",subject: "Welcome",body: "Your account is ready.",},});import{Worker}from"pollocks";constworker=newWorker(pool,{"send-email": async(job)=>{const{ to, subject, body }=job.payloadas{to: string;subject: string;body: string;};awaitsendEmail(to,subject,body);},});awaitworker.start();// Graceful shutdownprocess.on("SIGTERM",async()=>{awaitworker.stop();awaitpool.end();});The Tools class provides direct access to all job queue operations.
import{Tools}from"pollocks";consttools=newTools(pool);Runs all pending SQL migrations. Safe to call on every startup; already-applied migrations are skipped.
Enqueue a single job.
awaittools.addJob({pattern: "send-email",// required, routes to a handlerpayload: {to: "user@example.com"},// optional, defaults to {}runAfter: newDate("2025-01-01"),// optional, defaults to nowlockFor: 3600,// optional, lock duration in seconds, defaults to 3600});| Field | Type | Default | Description |
|---|---|---|---|
pattern | string | required | Routes the job to a matching handler |
payload | Record<string, unknown> | unknown[] | {} | Arbitrary JSON data attached to the job |
runAfter | Date | string | number | new Date() | Earliest time the job becomes eligible |
lockFor | number | 3600 | Seconds a job stays locked during processing |
Enqueue multiple jobs in a single database call.
constjobs=awaittools.addJobs([{pattern: "send-email",payload: {to: "a@example.com"}},{pattern: "send-email",payload: {to: "b@example.com"}},]);// jobs = [{ id: "01HX..." }, { id: "01HX..." }]Lock and return a single eligible job. Returns undefined if no job is available.
constjob=awaittools.acquireJob("worker-1",["send-email"]);Lock and return up to max eligible jobs.
constjobs=awaittools.acquireJobs(10,"worker-1",["send-email"]);Mark a job as completed.
Mark multiple jobs as completed.
Mark a job as failed. The job will be retried if it has remaining attempts.
awaittools.failJob(job.id,"Connection timeout");The Worker class handles job processing with automatic acquisition, execution, completion, and failure handling.
import{Worker}from"pollocks";constworker=newWorker(pool,handlers,config);Parameters:
| Parameter | Type | Description |
|---|---|---|
pool | Pool | A pg connection pool |
handlers | MessageHandlers | Map of pattern names to handler functions |
config | WorkerConfig | Optional configuration |
| Field | Type | Default | Description |
|---|---|---|---|
parallelism | number | 1 | Number of concurrent runner loops |
mode | "poll" | "listen" | "poll" | Processing mode |
pollIntervalMs | number | 2000 | Milliseconds between poll cycles |
lockedBy | string | auto-generated ULID | Identifier for this worker instance |
Start processing jobs. Spawns parallelism runner loops.
Graceful shutdown. Waits for all in-flight jobs to finish before returning.
Immediate shutdown. Marks all active jobs as failed and returns without waiting.
The worker.events emitter provides typed events for observability:
worker.events.on("start",({ patterns })=>{console.log(`Listening for: ${patterns.join(", ")}`);});worker.events.on("success",({ job, durationMs })=>{console.log(`Job ${job.id} completed in ${durationMs}ms`);});worker.events.on("failure",({ job, error, durationMs })=>{console.error(`Job ${job.id} failed after ${durationMs}ms:`,error);});| Event | Payload | Description |
|---|---|---|
start | { patterns: string[] } | Worker has started |
stop | {} | Worker has stopped |
shutdown | { forced: boolean } | Worker was killed |
poll | { runnerId: number } | Runner is polling for jobs |
listen | { runnerId: number, pattern: string } | Runner received a notification |
acquire | { runnerId: number, job: Job } | Runner acquired a job |
success | { runnerId: number, job: Job, durationMs: number } | Job completed successfully |
failure | { runnerId: number, job: Job, error: unknown, durationMs: number } | Job failed |
The Job type represents a row in the jobs table:
typeJob={id: string;created_at: Date;updated_at: Date|null;payload: Record<string,unknown>|unknown[];pattern: string;locked_by: string|null;locked_until: Date|null;locked_at: Date|null;last_error: string|null;run_after: Date;lock_for: number;attempts: number;max_attempts: number;};The default mode. Each runner loop calls acquireJob() on a fixed interval. Simple, reliable, and works with any PostgreSQL deployment including managed services that restrict LISTEN.
constworker=newWorker(pool,handlers,{mode: "poll",pollIntervalMs: 1000,});Uses PostgreSQL NOTIFY/LISTEN for near-instant job delivery. When a job is added, a notification triggers the worker to acquire it immediately. A periodic poll runs as a safety net to catch any missed notifications.
constworker=newWorker(pool,handlers,{mode: "listen",});Listen mode holds one additional database connection for the LISTEN subscription.
MIT