Using River from TypeScript

River’s TypeScript client inserts and works jobs on Node.js. Postgres and SQLite drivers provide the full runtime; the Prisma driver is insertion-only. Jobs share the River schema with Go, Ruby, and Rust.

Use Node.js 26 or newer, TypeScript 6, and an ESM project. Source is in river/js; the package is riverqueue. The shared getting-started guide and feature pages include TypeScript examples.

Basic usage

Install the core package along with the driver for your database library:

# Using node-postgres (pg)
pnpm add riverqueue @riverqueue/driver-pg pg zod

# Using Prisma
pnpm add riverqueue @riverqueue/driver-prisma

Initialize a client with:

import { Pool } from "pg";
import { Client } from "riverqueue";
import { PgDriver } from "@riverqueue/driver-pg";

const pool = new Pool({ connectionString: "postgres://localhost/mydb" });
const client = new Client(new PgDriver(pool));

Define a job with a stable kind and a schema that validates its JSON arguments:

import { Workers, defineJob } from "riverqueue";
import { z } from "zod";

const sort = defineJob({
  kind: "sort",
  schema: z.object({ strings: z.array(z.string()) }),
});

const result = await client.insert(sort, { strings: ["whale", "tiger", "bear"] });
result.job; // The inserted job row; its ID is a bigint.

For producer-only definitions without runtime validation, use defineJob({ kind: "sort" }). Persisted field names must match the receiving worker, regardless of its language.

Working jobs

Register a handler for the job definition, then start a client with that worker registry and a queue. Keep the run handle to stop it during application shutdown:

const workers = new Workers().add(sort, async ({ job }) => {
  console.log([...job.args.strings].sort());
});

const workerClient = new Client(new PgDriver(pool), {
  workers,
  queues: { default: { maxWorkers: 10 } },
});

const run = await workerClient.start();

// Later, during application shutdown:
await run.stop({ timeout: { seconds: 30 } });

Run migrations first and handle process signals as described in graceful shutdown. The Prisma driver cannot run workers.

Insertion options

Inserts take an options parameter to customize features of the inserted job:

const result = await client.insert(
  sort, { strings: ["whale", "tiger", "bear"] },
  {
    queue: "high_priority",
    priority: 2,
    maxAttempts: 5,
  }
);

Inserting unique jobs

Unique jobs are supported through unique, and can be made unique by args, period, and queue. If a job matching unique properties is found on insert, the insert is skipped and the existing job returned.

const result = await client.insert(
  sort, { strings: ["whale", "tiger", "bear"] },
  {
    unique: {
      byArgs: true,
      byQueue: true,
      byPeriod: { minutes: 15 }, // unique within 15-minute windows
    },
  }
);

// contains either a newly inserted job, or an existing one
result.job;

Inserting jobs in bulk

insertMany atomically inserts jobs, with per-entry options and potentially different definitions:

const results = await client.insertMany([
  { job: sort, args: { strings: ["whale", "tiger"] }, options: { maxAttempts: 5 } },
  { job: sort, args: { strings: ["bear", "fox"] }, options: { queue: "high_priority" } },
]);

Inserting in a transaction

Pass a transaction handle as the tx option to insert or insertMany. There’s no need to create a separate client for the transaction.

With node-postgres

node-postgres explicitly doesn’t define any high-level transaction constructs, so usage looks like manually issuing BEGIN, COMMIT, and ROLLBACK queries on a client checked out from the pool:

Using a client checked out from the pool:

const pool = new Pool({ connectionString: "postgres://localhost/mydb" });
const poolClient = await pool.connect();

try {
  await poolClient.query("BEGIN");

  await client.insert(sort, { strings: ["whale"] }, { tx: poolClient });
  await client.insert(sort, { strings: ["tiger"] }, { tx: poolClient });

  await poolClient.query("COMMIT");
} catch (e) {
  await poolClient.query("ROLLBACK");
  throw e;
} finally {
  poolClient.release();
}

With Prisma

Pass Prisma’s transaction client as { tx } to each insertion so the jobs commit or roll back with the enclosing transaction:

await prisma.$transaction(async (tx) => {
  await client.insert(sort, { strings: ["whale"] }, { tx });
  await client.insert(sort, { strings: ["tiger"] }, { tx });
});

Drivers

node-postgres

Use River with node-postgres (pg):

pnpm add riverqueue @riverqueue/driver-pg pg zod

Then initialize driver and client:

import { Pool } from "pg";
import { Client } from "riverqueue";
import { PgDriver } from "@riverqueue/driver-pg";

const pool = new Pool({ connectionString: "postgres://localhost/mydb" });
const client = new Client(new PgDriver(pool));

Prisma

Use River with Prisma:

pnpm add riverqueue @riverqueue/driver-prisma

Then initialize driver and client:

import { PrismaClient } from "@prisma/client";
import { Client } from "riverqueue";
import { PrismaDriver } from "@riverqueue/driver-prisma";

const prisma = new PrismaClient();
const client = new Client(new PrismaDriver(prisma));