Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions js/CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0

### Added

- Add `@riverqueue/driver-knex` for inserting jobs on Postgres inside Knex transactions, including batches, unique jobs, and nested savepoints. Configure its `knexTransactionAdapter` on `PgDriver` to use native Knex transactions with `completeTx` in workers. [PR #1516](https://github.com/riverqueue/river/pull/1516).
- Support Node.js 24 across all packages. River automatically installs a `Temporal` polyfill when the runtime has no native implementation, preserving native `Temporal` on Node.js 26.

### Changed
Expand Down
1 change: 1 addition & 0 deletions js/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -105,6 +105,7 @@ packages; it uses Node's built-in `node:sqlite`. See the
| Package | Purpose |
| ---------------------------- | ------------------------------------------------------------------- |
| `riverqueue` | Job definitions, insertion, workers, runtime, queries, and events |
| `@riverqueue/driver-knex` | Insert jobs inside Knex transactions on Postgres |
| `@riverqueue/driver-pg` | Postgres through `node-postgres` |
| `@riverqueue/driver-prisma` | Insert jobs inside Prisma transactions |
| `@riverqueue/driver-sqlite` | SQLite through Node's built-in `node:sqlite` |
Expand Down
7 changes: 6 additions & 1 deletion js/docs/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -265,7 +265,7 @@ Without `{ tx }`, River's own transaction rolls the whole call back.

The job's insert notification is delivered when the transaction commits, so
workers never see a job whose transaction rolled back. Prisma and SQLite use
the same `{ tx }` option; see the [Prisma](../driver/prisma/README.md) and
the same `{ tx }` option; see the [Knex](../driver/knex/README.md), [Prisma](../driver/prisma/README.md), and
[SQLite](../driver/sqlite/README.md) drivers.

## Work jobs
Expand Down Expand Up @@ -361,6 +361,11 @@ transaction of any installed River driver, so with both the Postgres and
SQLite drivers installed, passing a SQLite transaction to a Postgres client's
handler would compile and then fail at runtime.

For native Knex transactions, configure `knexTransactionAdapter` on the
worker's `PgDriver` and use `new Workers<Knex.Transaction>()`. See the
[Knex example](../driver/knex/README.md#complete-jobs-inside-knex-transactions).
Installing the Knex driver alone does not enable its transactions on `PgDriver`.

## Run workers

Configure queues and start the client. `maxWorkers` bounds concurrent
Expand Down
7 changes: 4 additions & 3 deletions js/docs/databases.md
Original file line number Diff line number Diff line change
Expand Up @@ -62,9 +62,10 @@ On Postgres, `@riverqueue/migrate` serializes concurrent migrators with a
transaction-scoped advisory lock. YugabyteDB has advisory locks only behind a
preview flag, so there it takes no lock; run one migrator at a time.

`@riverqueue/driver-prisma` is deliberately producer-only. It lets application
writes and job insertion share a Prisma transaction, but it does not claim or
work jobs. Use the Postgres runtime in worker services.
[`@riverqueue/driver-knex`](../driver/knex/README.md) and
[`@riverqueue/driver-prisma`](../driver/prisma/README.md) provide producer
clients. They let application writes and job insertion share a Knex or Prisma
transaction. Use the Postgres runtime in worker services.

## SQLite

Expand Down
8 changes: 4 additions & 4 deletions js/docs/development.md
Original file line number Diff line number Diff line change
Expand Up @@ -208,7 +208,7 @@ order are Rust-only because JavaScript objects cannot preserve them.

## Preparing a release

Run this section's commands from the repository root. The eight publishable packages are `js/package.json` (`riverqueue`), the three drivers under `js/driver/*`, and `js/{migrate,worker-threads,test,cli}`. They share one version, independently of Go and Rust. Examples are private and stay at `0.0.0`. `VERSION` has no leading `v`; JavaScript Git tags use `js/vX.Y.Z`.
Run this section's commands from the repository root. The nine publishable packages are `js/package.json` (`riverqueue`), the four drivers under `js/driver/*`, and `js/{migrate,worker-threads,test,cli}`. They share one version, independently of Go and Rust. Examples are private and stay at `0.0.0`. `VERSION` has no leading `v`; JavaScript Git tags use `js/vX.Y.Z`.

1. Fetch changes and tags, choose the next JavaScript version (including a prerelease suffix when applicable), and create a release branch:

Expand Down Expand Up @@ -242,18 +242,18 @@ Run this section's commands from the repository root. The eight publishable pack
git push origin "js/v$VERSION"
```

6. Publish locally from the clean `master` checkout tagged above after its JavaScript checks pass. Use official Node.js 26 and the pinned pnpm version, and log in with an npm account that can publish all eight packages. A publication workflow is optional. Disable [provenance](https://docs.npmjs.com/generating-provenance-statements/) for local publication with `--provenance=false`, overriding the packages' `publishConfig.provenance: true`:
6. Publish locally from the clean `master` checkout tagged above after its JavaScript checks pass. Use official Node.js 26 and the pinned pnpm version, and log in with an npm account that can publish all nine packages. A publication workflow is optional. Disable [provenance](https://docs.npmjs.com/generating-provenance-statements/) for local publication with `--provenance=false`, overriding the packages' `publishConfig.provenance: true`:

```shell
test "$(git rev-parse HEAD)" = "$(git rev-parse "js/v$VERSION^{commit}")"
pnpm login
pnpm -C js install --frozen-lockfile
pnpm -C js run build:all
for package in js js/migrate js/driver/pg js/driver/prisma js/driver/sqlite js/worker-threads js/test js/cli; do
for package in js js/migrate js/driver/knex js/driver/pg js/driver/prisma js/driver/sqlite js/worker-threads js/test js/cli; do
(cd "$package" && pnpm publish --access public --no-git-checks --tag latest --provenance=false) || break
done
```

Complete npm's authentication prompts as needed. Publish packages individually because pnpm 10.22.0's recursive publication does not forward `--provenance=false` to npm. Run publication inside each package directory: `pnpm -C "$package" publish` in this version forwards extra arguments to npm and fails with `EUSAGE`. The loop publishes dependencies first and stops on failure; after a partial publication, remove the already-published packages from the loop before rerunning it. For a prerelease, use `--tag next` instead of `--tag latest`.

7. Once all eight packages are published, create a [GitHub release](https://github.com/riverqueue/river/releases/new) for `js/v$VERSION` and copy the version's `js/CHANGELOG.md` notes into its body. Mark alpha, beta, and RC versions as prereleases.
7. Once all nine packages are published, create a [GitHub release](https://github.com/riverqueue/river/releases/new) for `js/v$VERSION` and copy the version's `js/CHANGELOG.md` notes into its body. Mark alpha, beta, and RC versions as prereleases.
120 changes: 120 additions & 0 deletions js/driver/knex/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,120 @@
# `@riverqueue/driver-knex`

Insert River jobs through Knex on Postgres, including atomically with your
application's writes. This package provides an insertion client; use
`@riverqueue/driver-pg` to run workers against the same database and schema.

```sh
npm install riverqueue @riverqueue/driver-knex knex pg
```

```ts
import knex from "knex";
import { Client, defineJob } from "riverqueue";
import { KnexDriver } from "@riverqueue/driver-knex";

const db = knex({
client: "pg",
connection: process.env.DATABASE_URL ?? "postgres://localhost/river",
});
const client = new Client(new KnexDriver(db));
const syncAccount = defineJob<{ accountId: string }>()({
kind: "sync_account",
});

await db.transaction(async (tx) => {
const accountId = "acct_1";
await tx("account").insert({ id: accountId });
await client.insert(syncAccount, { accountId }, { tx });
});

await db.destroy(); // Close your application's pool at shutdown.
```

Pass a root Knex instance to `KnexDriver` and an active `Knex.Transaction` as
`{ tx }` to `insert` or `insertMany`. River uses that exact transaction,
including nested transactions backed by savepoints, and leaves commit and
rollback to the caller. An outer rollback removes both application writes and
jobs. Insert notifications reach workers only after commit.

Without `{ tx }`, River opens a Knex transaction around insertion, middleware,
hooks, and notifications. If any of them throws, the transaction rolls back.
The Knex instance remains caller-owned; River never destroys its pool.

Configure a River schema with `new KnexDriver(db, { schema: "river" })`.
Otherwise River uses the connection's `search_path`. Apply River migrations
separately with `@riverqueue/migrate` or `@riverqueue/cli`, using the same
database and schema. River's migrations are independent of Knex's application
migrations.

Job IDs remain `bigint`, timestamps retain microsecond precision as
`Temporal.Instant`, and JSON numbers beyond JavaScript's safe integer range
remain exact. Batch ordering and unique jobs follow River's normal semantics.

## Complete jobs inside Knex transactions

Workers run through `@riverqueue/driver-pg`. Add that package and configure
`knexTransactionAdapter` on its driver to pass native Knex transactions to
`completeTx`:

```ts
import knex, { type Knex } from "knex";
import { Pool } from "pg";
import { Client, defineJob, Workers } from "riverqueue";
import { knexTransactionAdapter } from "@riverqueue/driver-knex";
import { PgDriver } from "@riverqueue/driver-pg";

const connectionString =
process.env.DATABASE_URL ?? "postgres://localhost/river";
const db = knex({ client: "pg", connection: connectionString });
const pool = new Pool({ connectionString });
const syncAccount = defineJob<{ accountId: string }>()({
kind: "sync_account",
});
const workers = new Workers<Knex.Transaction>().add(
syncAccount,
async ({ completeTx, job }) => {
await db.transaction(async (tx) => {
await tx("account")
.where({ id: job.args.accountId })
.update({ synced: true });
await completeTx(tx);
});
}
);
const client = new Client(
new PgDriver(pool, { transactionAdapters: [knexTransactionAdapter] }),
{ queues: { default: { maxWorkers: 10 } }, workers }
);
const run = await client.start();
```

The application write and job completion use the same Knex transaction and
commit together. Knex owns commit, rollback, and nested savepoints. If the
transaction throws, both writes roll back, and River handles the worker's
error normally. Await the transaction before returning from the handler.

Configure the same database and River schema on the worker and insertion
clients. The adapter also enables Knex `{ tx }` options on the worker client's
other operations, including inserts from `ctx.client`. Importing the adapter
doesn't enable it automatically; configure each `PgDriver` that needs it.
Neither `riverqueue` nor `@riverqueue/driver-pg` depends on Knex.

At shutdown, stop `run`, then close the application's Knex and `pg` pools.

## Knex response processing

River uses parameterized `raw` queries with `queryContext({ river: true })`.
If your Knex configuration uses `postProcessResponse`, return the original
response when `queryContext?.river` is true. River needs the Postgres response's
`rows` array and its original column names and values for exact decoding.

## Requirements

Knex 3.1 or newer within the 3.x line, `pg` 8.x, and Node.js 24 or newer. This
adapter supports Knex's Postgres client (`client: "pg"` or `"postgresql"`).
For SQLite, use `@riverqueue/driver-sqlite`.

Install `riverqueue` at exactly this package's version. TypeScript users need
TypeScript 6.0 or newer and `@types/node`, with `"node"` in
`compilerOptions.types`.
50 changes: 50 additions & 0 deletions js/driver/knex/etc/driver-knex.api.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,50 @@
# API report: `@riverqueue/driver-knex`

<!-- Generated by scripts/api-reports.mjs. Do not edit directly. -->

This report contains declarations and TSDoc for names exported by the
package entry point. Private members, unexported implementation
declarations, and file layout are omitted.

## `KnexDriver`

```ts
/**
* River's insertion adapter for Knex on Postgres (`client: "pg"`).
*
* The Knex instance is caller-owned. Pass an active Knex transaction as
* `{ tx }` to insert jobs atomically with application writes. Otherwise River
* opens a Knex transaction for the insertion, its hooks, and notifications.
* Run workers with `@riverqueue/driver-pg`.
*/
export declare class KnexDriver {
/** Type-only marker for insertion with Knex transactions. */
readonly "~river"?: {
readonly capability: "insert";
readonly transaction: Knex.Transaction;
};
constructor(knex: Knex, options?: KnexDriverOptions);
}
```

## `KnexDriverOptions`

```ts
/** Postgres configuration owned by the Knex adapter. */
export interface KnexDriverOptions {
/** Postgres schema containing River's tables and functions. */
schema?: string;
}
```

## `knexTransactionAdapter`

```ts
/**
* Let a PgDriver use native Knex transactions, including `completeTx(tx)`
* in worker handlers. Configure `new PgDriver(pool, {
* transactionAdapters: [knexTransactionAdapter] })`. Knex owns transaction
* commit, rollback, and savepoints; River uses the supplied transaction only.
*/
export declare const knexTransactionAdapter: PostgresTransactionAdapter<Knex.Transaction>;
```
63 changes: 63 additions & 0 deletions js/driver/knex/package.json
Original file line number Diff line number Diff line change
@@ -0,0 +1,63 @@
{
"name": "@riverqueue/driver-knex",
"version": "0.2.0",
"description": "Knex driver for the riverqueue TypeScript client.",
"type": "module",
"sideEffects": false,
"main": "./dist/index.js",
"types": "./dist/index.d.ts",
"exports": {
".": {
"types": "./dist/index.d.ts",
"import": "./dist/index.js",
"default": "./dist/index.js"
}
},
"engines": {
"node": ">=24"
},
"files": [
"dist",
"src",
"!src/**/*.test.ts",
"README.md",
"LICENSE"
],
"scripts": {
"build": "node ../../node_modules/typescript/bin/tsc && node ../../scripts/copy-license.mjs",
"clean": "rm -rf dist",
"prepack": "pnpm run clean && pnpm run build"
},
"repository": {
"type": "git",
"url": "git+https://github.com/riverqueue/river.git",
"directory": "js/driver/knex"
},
"contributors": [
"Brandur Leach",
"Blake Gentry"
],
"license": "MPL-2.0",
"publishConfig": {
"access": "public",
"provenance": true
},
"peerDependencies": {
"knex": "^3.1.0",
"pg": "^8.0.0",
"riverqueue": "workspace:0.2.0"
},
"devDependencies": {
"knex": "^3.1.0",
"pg": "^8.22.0",
"riverqueue": "workspace:0.2.0",
"typescript": "^6.0.3"
},
"keywords": [
"river",
"job-queue",
"postgresql",
"knex",
"driver"
]
}
Loading
Loading