node-river

Node.js client for River, a Postgres-backed job queue. Supports job insertion and worker-based processing with per-queue concurrency control.

Installation

npm install @odunlamizo/node-river

Setup

import { RiverClient } from '@odunlamizo/node-river';
import { PgDriver } from '@odunlamizo/node-river/drivers/pg';

const driver = new PgDriver({ connectionString: process.env.DATABASE_URL! });

const client = new RiverClient(driver, {
  queues: {
    default: { concurrency: 10 },
    emails: { concurrency: 50 },
  },
  maxAttempts: 3,
});

await client.verifyConnection();

Inserting Jobs

queue is required on every insert — there is no default fallback.

// Single job
const result = await client.insert(
  { kind: 'send_email', to: '[email protected]' },
  { queue: 'emails' },
);
console.log(result.job); // inserted Job record
console.log(result.skipped); // true if deduplicated

// Multiple jobs in one transaction (all succeed or all roll back)
const results = await client.insertMany([
  { args: { kind: 'send_email', to: '[email protected]' }, opts: { queue: 'emails' } },
  { args: { kind: 'send_email', to: '[email protected]' }, opts: { queue: 'emails' } },
]);

// Inside an existing transaction
const tx = await driver.getTx();
try {
  await tx.query('BEGIN');
  await client.insertTx(tx, { kind: 'send_email', to: '[email protected]' }, { queue: 'emails' });
  await tx.query('COMMIT');
} catch (e) {
  await tx.query('ROLLBACK');
  throw e;
} finally {
  tx.release();
}

Unique Jobs

Use uniqueOpts to prevent duplicate jobs from being enqueued.

// Deduplicate by specific args
await client.insert(
  { kind: 'send_email', to: '[email protected]' },
  { queue: 'emails', uniqueOpts: { byArgs: ['to'] } },
);

// Deduplicate by queue + args
await client.insert(
  { kind: 'send_email', to: '[email protected]' },
  { queue: 'emails', uniqueOpts: { byQueue: true, byArgs: ['to'] } },
);

// Deduplicate within a time window (e.g. one per 60-second period)
await client.insert(
  { kind: 'send_email', to: '[email protected]' },
  { queue: 'emails', uniqueOpts: { byPeriod: 60 }, scheduledAt: new Date() },
);

Processing Jobs (Workers)

Implement the Worker<T> interface for each job kind, register it with addWorker, then call work(). Each queue is polled independently at the configured concurrency limit.

import { Job, JobArgs, Worker } from '@odunlamizo/node-river/types';

interface SendEmailArgs extends JobArgs<{ to: string }> {
  kind: 'send_email';
}

class SendEmailWorker implements Worker<SendEmailArgs> {
  async work(job: Job<SendEmailArgs>): Promise<void> {
    await sendEmail(job.args.to);
  }
}

client.addWorker('send_email', new SendEmailWorker());
client.work(); // starts polling all configured queues

Throwing inside work() marks the job as failed. If attempt < maxAttempts it is retried with exponential backoff; otherwise it is discarded.

Shutdown

Always call close() on shutdown to drain the poll timer and release database connections.

process.on('SIGTERM', async () => {
  await client.close();
});

Configuration

| Option | Type | Description | | -------------- | ----------------------------------------- | ------------------------------------------------------------------- | | queues | Record<string, { concurrency: number }> | Queues to poll and their concurrency limits. Required for work(). | | maxAttempts | number | Default max attempts for inserted jobs. | | pollInterval | number | Milliseconds between polls. Defaults to 1000. | | clientId | string | Unique ID for this client instance. Defaults to hostname-pid. |

License

MIT