@everystack/jobs

Background job queue with pluggable adapters — PostgreSQL (FOR UPDATE SKIP LOCKED) or AWS SQS. Publish, schedule, process, retry, and monitor jobs.

Install

pnpm add @everystack/jobs drizzle-orm

For SQS adapter:

pnpm add @aws-sdk/client-sqs

Entry Points

| Import | Description | |--------|-------------| | @everystack/jobs | Client, worker, handler, adapters | | @everystack/jobs/schema | Drizzle jobs table |

Quick Start

import {
  createJobClient,
  createJobWorker,
  PostgresJobAdapter,
} from '@everystack/jobs';
import { db } from './db';

// 1. Create adapter
const adapter = new PostgresJobAdapter(db);

// 2. Create client (for publishing jobs)
const client = createJobClient(adapter);

// 3. Create worker (for processing jobs)
const worker = createJobWorker(adapter, {
  handlers: {
    'send-email': async (payload, job) => {
      await sendEmail(payload.to, payload.subject, payload.body);
    },
    'process-image': async (payload, job) => {
      await processImage(payload.key);
    },
  },
  concurrency: 3,
  pollInterval: 1000,
});

// 4. Start the worker
worker.start();

Job Client

Publish and manage jobs:

const client = createJobClient(adapter);

// Publish immediately
const jobId = await client.publish('send-email', {
  to: '[email protected]',
  subject: 'Welcome',
  body: 'Hello!',
});

// Publish with options
const jobId = await client.publish('send-report', reportData, {
  maxAttempts: 5,
  priority: 10, // Higher = processed first
});

// Schedule for later
const jobId = await client.schedule(
  'send-digest',
  { userId: '123' },
  new Date('2025-01-01T09:00:00Z'),
);

// Query jobs
const job = await client.getJob(jobId);
const pending = await client.listJobs({ status: 'pending', limit: 50 });
const failed = await client.listJobs({ status: 'failed', type: 'send-email' });

// Retry a failed job
await client.retry(jobId);

// Get queue stats
const stats = await client.stats();
// → { pending: 12, active: 3, completed: 450, failed: 2, dead: 0 }

Job Worker

Process jobs with handlers:

const worker = createJobWorker(adapter, {
  // Map of job type → handler function
  handlers: {
    'send-email': async (payload, job) => {
      // payload is the data you published
      // job has id, type, attempts, maxAttempts, etc.
      await sendEmail(payload);
    },
  },

  concurrency: 5,        // Max concurrent jobs (default: 1)
  pollInterval: 1000,     // Poll interval in ms (default: 1000)
  workerId: 'worker-1',   // Unique worker ID (default: random UUID)
  shutdownTimeout: 30000, // Wait for active jobs on stop (default: 30s)
});

// Start processing
worker.start();

// Lifecycle events
worker.on('job:start', (job) => console.log(`Started: ${job.type}`));
worker.on('job:complete', (job) => console.log(`Done: ${job.id}`));
worker.on('job:fail', (job) => console.log(`Failed: ${job.id} — ${job.error}`));
worker.on('job:dead', (job) => console.log(`Dead: ${job.id} — max attempts reached`));

// Check status
worker.isRunning(); // → true

// Graceful shutdown (waits for active jobs)
await worker.stop();

Error Handling

Failed jobs are automatically retried with exponential backoff (2^attempts seconds). After maxAttempts (default: 3), the job moves to dead status.

| Status | Description | |--------|-------------| | pending | Waiting to be processed | | active | Currently being processed | | completed | Finished successfully | | failed | Failed, will retry | | dead | Exhausted all retry attempts |

Adapters

PostgreSQL (Recommended for most cases)

Uses FOR UPDATE SKIP LOCKED for reliable, concurrent job processing. No additional infrastructure — just your existing PostgreSQL database.

import { PostgresJobAdapter } from '@everystack/jobs';

const adapter = new PostgresJobAdapter(db);

Features:

• Atomic job acquisition with SKIP LOCKED
• Exponential backoff on failure
• Priority-based ordering
• Scheduled jobs (runAt)

AWS SQS

For high-throughput or distributed workers. Uses SQS for message delivery with optional PostgreSQL for metadata/visibility.

import { SQSJobAdapter } from '@everystack/jobs';

const adapter = new SQSJobAdapter(
  'https://sqs.us-east-1.amazonaws.com/123456789/my-queue',
  'https://sqs.us-east-1.amazonaws.com/123456789/my-dlq', // Optional DLQ
  'us-east-1',
  db, // Optional: Drizzle db for metadata tracking
);

Features:

• Long-polling message receive (20s)
• Automatic dead-letter queue support
• Up to 15-minute delay (SQS limit)
• Optional metadata DB for job listing/stats

HTTP Handler

Expose job management via HTTP:

import { createJobsHandler } from '@everystack/jobs';

const handler = createJobsHandler({
  adapter,
  basePath: '/api/jobs',
  auth: { verifyToken: async (token) => verifyJWT(token) },
});

Endpoints

| Method | Path | Description | |--------|------|-------------| | GET | /stats | Queue statistics | | GET | /jobs | List jobs (supports ?status=, ?type=, ?limit=, ?offset=) | | GET | /jobs/:id | Get job details | | POST | /jobs/:id/retry | Retry a failed/dead job (auth required) | | POST | /publish | Publish a new job (auth required) |

Schema

Add the jobs table to your Drizzle migrations:

import { jobs } from '@everystack/jobs/schema';
// Include in your Drizzle schema

The jobs table includes: id, type, payload, status, attempts, maxAttempts, priority, runAt, lockedAt, lockedBy, completedAt, failedAt, error, createdAt.

Peer Dependencies

| Package | Version | Required | |---------|---------|----------| | drizzle-orm | >=0.30.0 | For PostgreSQL adapter | | @aws-sdk/client-sqs | >=3.0.0 | For SQS adapter |

Part of everystack — a self-hosted application stack for Expo apps on AWS.

License

MIT