diff --git a/js/CHANGELOG.md b/js/CHANGELOG.md index 9d8e2d797..d3f7f1f91 100644 --- a/js/CHANGELOG.md +++ b/js/CHANGELOG.md @@ -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 diff --git a/js/README.md b/js/README.md index 3e0f35037..14907287b 100644 --- a/js/README.md +++ b/js/README.md @@ -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` | diff --git a/js/docs/README.md b/js/docs/README.md index 91c650d97..6e786fb69 100644 --- a/js/docs/README.md +++ b/js/docs/README.md @@ -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 @@ -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()`. 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 diff --git a/js/docs/databases.md b/js/docs/databases.md index c368d516d..7f2b49a19 100644 --- a/js/docs/databases.md +++ b/js/docs/databases.md @@ -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 diff --git a/js/docs/development.md b/js/docs/development.md index eb9a28c44..9dd9d845f 100644 --- a/js/docs/development.md +++ b/js/docs/development.md @@ -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: @@ -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. diff --git a/js/driver/knex/README.md b/js/driver/knex/README.md new file mode 100644 index 000000000..b6121b369 --- /dev/null +++ b/js/driver/knex/README.md @@ -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().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`. diff --git a/js/driver/knex/etc/driver-knex.api.md b/js/driver/knex/etc/driver-knex.api.md new file mode 100644 index 000000000..95b0343ee --- /dev/null +++ b/js/driver/knex/etc/driver-knex.api.md @@ -0,0 +1,50 @@ +# API report: `@riverqueue/driver-knex` + + + +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; +``` diff --git a/js/driver/knex/package.json b/js/driver/knex/package.json new file mode 100644 index 000000000..721c96d8f --- /dev/null +++ b/js/driver/knex/package.json @@ -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" + ] +} diff --git a/js/driver/knex/src/driver.integration.test.ts b/js/driver/knex/src/driver.integration.test.ts new file mode 100644 index 000000000..ba291d274 --- /dev/null +++ b/js/driver/knex/src/driver.integration.test.ts @@ -0,0 +1,148 @@ +import { randomUUID } from "node:crypto"; +import knex, { type Knex } from "knex"; +import { Pool } from "pg"; +import { Client, defineJob, type InsertClient } from "riverqueue"; +import { createMigrator } from "../../../migrate/src/index.js"; +import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; +import { KnexDriver } from "./index.js"; + +const TEST_DATABASE_URL = + process.env.TEST_DATABASE_URL ?? + "postgres://localhost:5432/river_test?sslmode=disable"; + +describe("KnexDriver transactions", () => { + const schema = `knex_${randomUUID().replaceAll("-", "")}`; + const job = defineJob<{ value?: string }>()({ kind: "knex_test" }); + let client: InsertClient; + let db: Knex; + let pool: Pool; + + beforeAll(async () => { + pool = new Pool({ connectionString: TEST_DATABASE_URL }); + await pool.query(`CREATE SCHEMA "${schema}"`); + await createMigrator({ pool, schema }).migrateUp(); + db = knex({ + client: "pg", + connection: TEST_DATABASE_URL, + pool: { min: 0, max: 2 }, + }); + await db.schema + .withSchema(schema) + .createTable("application_row", (table) => { + table.text("value").notNullable(); + }); + client = new Client(new KnexDriver(db, { schema })); + }); + + afterAll(async () => { + await db.destroy(); + await pool.query(`DROP SCHEMA "${schema}" CASCADE`); + await pool.end(); + }); + + afterEach(async () => { + await db.withSchema(schema).table("application_row").delete(); + await db.withSchema(schema).table("river_job").delete(); + }); + + it("commits application writes and jobs together", async () => { + const value = "quotes ' ? $1 \\ and Unicode ☃"; + await db.transaction(async (tx) => { + await tx.withSchema(schema).table("application_row").insert({ value }); + const inserted = await client.insert(job, { value }, { tx }); + expect(inserted.job.args).toEqual({ value }); + expect(await db.withSchema(schema).table("river_job")).toHaveLength(0); + expect(await tx.withSchema(schema).table("river_job")).toHaveLength(1); + }); + expect(await db.withSchema(schema).table("application_row")).toEqual([ + { value }, + ]); + expect(await db.withSchema(schema).table("river_job")).toHaveLength(1); + }); + + it("honors caller-owned savepoints", async () => { + const failure = new Error("rollback savepoint"); + await db.transaction(async (tx) => { + await client.insert(job, { value: "outer" }, { tx }); + await expect( + tx.transaction(async (nested) => { + await client.insert(job, { value: "inner" }, { tx: nested }); + throw failure; + }) + ).rejects.toBe(failure); + expect( + await tx.withSchema(schema).table("river_job").select("args") + ).toEqual([{ args: { value: "outer" } }]); + }); + expect(await db.withSchema(schema).table("river_job")).toHaveLength(1); + }); + + it("preserves raw responses when application postprocessing opts out", async () => { + const processed = db.withUserParams({}); + processed.client.config.postProcessResponse = ( + result: unknown, + context: unknown + ) => { + if ( + typeof context === "object" && + context !== null && + "river" in context && + context.river === true + ) + return result; + throw new Error("application postprocessing must not change River rows"); + }; + const result = await new Client( + new KnexDriver(processed, { schema }) + ).insert(job, {}); + expect(result.status).toBe("inserted"); + }); + + it("rejects completed transactions and transaction constructors", async () => { + const tx = await db.transaction(); + expect(() => new KnexDriver(tx)).toThrow("root Knex instance"); + await tx.rollback(); + await expect(client.insert(job, {}, { tx })).rejects.toThrow( + "active Knex transaction" + ); + }); + + it("rolls back application writes and jobs together", async () => { + const failure = new Error("rollback transaction"); + await expect( + db.transaction(async (tx) => { + await tx + .withSchema(schema) + .table("application_row") + .insert({ value: "rolled back" }); + await client.insertMany( + [ + { args: {}, job }, + { args: { value: "second" }, job }, + ], + { tx } + ); + throw failure; + }) + ).rejects.toBe(failure); + expect(await db.withSchema(schema).table("application_row")).toHaveLength( + 0 + ); + expect(await db.withSchema(schema).table("river_job")).toHaveLength(0); + }); + + it("rolls back its own transaction when insert middleware fails", async () => { + const failure = new Error("after insert"); + const failingClient = new Client(new KnexDriver(db, { schema }), { + insertMiddleware: [ + async (_context, next) => { + await next(); + throw failure; + }, + ], + }); + await expect(failingClient.insert(job, {})).rejects.toBe(failure); + expect(await db.withSchema(schema).table("river_job")).toHaveLength(0); + expect((await client.insert(job, {})).status).toBe("inserted"); + }); +}); diff --git a/js/driver/knex/src/driver.test.ts b/js/driver/knex/src/driver.test.ts new file mode 100644 index 000000000..a204f7838 --- /dev/null +++ b/js/driver/knex/src/driver.test.ts @@ -0,0 +1,82 @@ +import knex, { type Knex } from "knex"; +import { Pool, type ClientBase } from "pg"; +import { Client, defineJob, type InsertClient } from "riverqueue"; +import { describe, expect, expectTypeOf, it } from "vitest"; +import { KnexDriver, knexTransactionAdapter } from "./index.js"; +import { PgDriver } from "../../pg/src/index.js"; + +describe("KnexDriver", () => { + it("adds Knex transactions to the configured Postgres client's types", async () => { + const pool = new Pool(); + try { + const client = new Client( + new PgDriver(pool, { transactionAdapters: [knexTransactionAdapter] }) + ); + expectTypeOf(client).toEqualTypeOf< + import("riverqueue").Client + >(); + const typed = (tx: Knex.Transaction, root: Knex) => { + void client.jobs.get(1n, { tx }); + // @ts-expect-error A root Knex instance is not a transaction. + void client.jobs.get(1n, { tx: root }); + }; + expect(typed).toBeTypeOf("function"); + } finally { + await pool.end(); + } + }); + + it("exposes an insertion-only client with Knex transactions", async () => { + const db = knex({ client: "pg" }); + try { + const client = new Client(new KnexDriver(db)); + expectTypeOf(client).toEqualTypeOf>(); + // @ts-expect-error Knex does not support the worker runtime API. + await expect(client.start()).rejects.toThrow(); + } finally { + await db.destroy(); + } + }); + + it("rejects a root client passed as a transaction", async () => { + const db = knex({ client: "pg" }); + try { + const client = new Client(new KnexDriver(db)); + await expect( + // @ts-expect-error A root Knex instance cannot be used as a transaction. + client.insert(defineJob({ kind: "test" }), {}, { tx: db }) + ).rejects.toThrow("active Knex transaction"); + } finally { + await db.destroy(); + } + }); + + it("rejects invalid schemas before issuing queries", async () => { + const db = knex({ client: "pg" }); + try { + for (const schema of ["", "a.b", "bad$1", "bad?", "a".repeat(47)]) { + expect(() => new KnexDriver(db, { schema })).toThrow(); + } + expect( + () => new KnexDriver(db, { schema: "a".repeat(46) }) + ).not.toThrow(); + } finally { + await db.destroy(); + } + }); + + it("rejects unsupported clients", () => { + for (const value of [null, {}, () => {}, { raw: () => {} }]) { + expect(() => new KnexDriver(value as unknown as Knex)).toThrow( + "Knex requires a Postgres client" + ); + } + const wrongDialect = Object.assign(() => {}, { + client: { dialect: "sqlite3" }, + raw: () => {}, + }); + expect(() => new KnexDriver(wrongDialect as unknown as Knex)).toThrow( + "Knex requires a Postgres client" + ); + }); +}); diff --git a/js/driver/knex/src/driver.ts b/js/driver/knex/src/driver.ts new file mode 100644 index 000000000..9b5ecf1fa --- /dev/null +++ b/js/driver/knex/src/driver.ts @@ -0,0 +1,65 @@ +import type { Knex } from "knex"; +import { ConfigurationError } from "riverqueue"; +import { PostgresInserter, registerDriver } from "riverqueue/unstable-driver"; +import { + queryKnex, + validateKnexClient, + validateKnexTransaction, +} from "./transaction.js"; + +/** + * 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 class KnexDriver { + /** Type-only marker for insertion with Knex transactions. */ + declare readonly "~river"?: { + readonly capability: "insert"; + readonly transaction: Knex.Transaction; + }; + + constructor(knex: Knex, options: KnexDriverOptions = {}) { + validateKnexClient(knex); + if (knex.isTransaction) { + throw new ConfigurationError( + "KnexDriver requires a root Knex instance; pass transactions as { tx }" + ); + } + const inserter = new PostgresInserter({ + backend: "postgres-knex-insert", + name: "Knex", + operationScope: (tx, callback) => { + if (tx === undefined) return knex.transaction(callback); + validateKnexTransaction(tx); + return callback(tx); + }, + query: async ( + tx: Knex.Transaction | undefined, + sql: string, + ...values: unknown[] + ): Promise => { + const result = await queryKnex(tx ?? knex, { + text: sql, + values, + }); + return result.rows as T; + }, + ...(options.schema === undefined ? {} : { schema: options.schema }), + }); + registerDriver(this, { + backend: inserter.backend, + capability: "insert", + operations: inserter, + }); + } +} + +/** Postgres configuration owned by the Knex adapter. */ +export interface KnexDriverOptions { + /** Postgres schema containing River's tables and functions. */ + schema?: string; +} diff --git a/js/driver/knex/src/index.ts b/js/driver/knex/src/index.ts new file mode 100644 index 000000000..b97d2e300 --- /dev/null +++ b/js/driver/knex/src/index.ts @@ -0,0 +1,12 @@ +import type { Knex } from "knex"; + +export { KnexDriver } from "./driver.js"; +export type { KnexDriverOptions } from "./driver.js"; +export { knexTransactionAdapter } from "./transaction.js"; + +declare module "riverqueue" { + interface RiverTransactionRegistry { + /** Knex transactions on Postgres. */ + "@riverqueue/driver-knex": Knex.Transaction; + } +} diff --git a/js/driver/knex/src/runtime.integration.test.ts b/js/driver/knex/src/runtime.integration.test.ts new file mode 100644 index 000000000..139774473 --- /dev/null +++ b/js/driver/knex/src/runtime.integration.test.ts @@ -0,0 +1,323 @@ +import { randomUUID } from "node:crypto"; +import knex, { type Knex } from "knex"; +import { Pool, type ClientBase } from "pg"; +import { + Client, + defineJob, + exactJsonNumber, + type JobRow, + type WorkHandler, + Workers, +} from "riverqueue"; +import { PilotClient, type PilotFactory } from "riverqueue/unstable-driver"; +import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; +import { createMigrator } from "../../../migrate/src/index.js"; +import { PgDriver } from "../../pg/src/index.js"; +import { knexTransactionAdapter } from "./index.js"; + +const TEST_DATABASE_URL = + process.env.TEST_DATABASE_URL ?? + "postgres://localhost:5432/river_test?sslmode=disable"; + +class CompanionClient extends PilotClient {} + +describe("Knex transactional completion", () => { + const schema = `knex_work_${randomUUID().replaceAll("-", "")}`; + const definition = defineJob({ kind: "knex_work" }); + let db: Knex; + let pool: Pool; + + beforeAll(async () => { + pool = new Pool({ connectionString: TEST_DATABASE_URL }); + await pool.query(`CREATE SCHEMA "${schema}"`); + await createMigrator({ pool, schema }).migrateUp(); + db = knex({ + client: "pg", + connection: TEST_DATABASE_URL, + pool: { min: 0, max: 2 }, + }); + await db.schema + .withSchema(schema) + .createTable("application_row", (table) => { + table.text("value").notNullable(); + }); + }); + + afterAll(async () => { + await db.destroy(); + await pool.query(`DROP SCHEMA "${schema}" CASCADE`); + await pool.end(); + }); + + afterEach(async () => { + await db.withSchema(schema).table("application_row").delete(); + await db.withSchema(schema).table("river_job").delete(); + }); + + const runWorker = async ( + handler: WorkHandler, + createPilot?: PilotFactory + ): Promise => { + const driver = new PgDriver(pool, { + schema, + transactionAdapters: [knexTransactionAdapter], + }); + const options = { + completionBatchSize: 1, + leaderElectionDisabled: true, + queues: { + default: { + fetchCooldown: { milliseconds: 1 }, + maxWorkers: 1, + pollInterval: { milliseconds: 5 }, + }, + }, + workers: new Workers().add(definition, handler), + }; + const client = + createPilot === undefined + ? new Client(driver, options) + : new CompanionClient(driver, options, createPilot); + const inserted = await client.insert( + definition, + { id: exactJsonNumber("9223372036854775807") }, + { maxAttempts: 1 } + ); + const events = client.subscribe({ + kinds: ["job_completed", "job_failed"], + signal: AbortSignal.timeout(10_000), + }); + const run = await client.start(); + try { + let completed = false; + for await (const event of events) { + if (event.kind === "subscription_lag") continue; + if (event.job.id !== inserted.job.id) continue; + completed = true; + break; + } + expect(completed).toBe(true); + const row = await client.jobs.get(inserted.job.id); + expect(row).not.toBeNull(); + return row!; + } finally { + events.close(); + await run.stop({ timeout: { seconds: 5 } }); + } + }; + + it("commits application writes and exact completion values together", async () => { + let completed: JobRow | undefined; + const row = await runWorker(async ({ completeTx, job, setMetadata }) => { + setMetadata("custom", "value"); + await db.transaction(async (tx) => { + await tx + .withSchema(schema) + .table("application_row") + .insert({ value: "committed ? $1 ' ☃" }); + completed = await completeTx(tx, { + output: { id: exactJsonNumber("9223372036854775807") }, + }); + expect(completed.id).toBe(job.id); + expect(completed.createdAt.toString()).toBe(job.createdAt.toString()); + expect(completed.finalizedAt).toBeInstanceOf(Temporal.Instant); + expect(completed.state).toBe("completed"); + expect(JSON.stringify(completed.args)).toBe( + '{"id":9223372036854775807}' + ); + expect( + ( + await pool.query( + `SELECT state FROM "${schema}".river_job WHERE id = $1`, + [job.id.toString()] + ) + ).rows[0].state + ).toBe("running"); + expect( + await db.withSchema(schema).table("application_row") + ).toHaveLength(0); + }); + // A later handler failure cannot undo a completion already committed. + throw new Error("after commit"); + }); + expect(row.state).toBe("completed"); + expect(row.metadata.custom).toBe("value"); + expect(JSON.stringify(row.metadata.output)).toBe( + '{"id":9223372036854775807}' + ); + expect(await db.withSchema(schema).table("application_row")).toEqual([ + { value: "committed ? $1 ' ☃" }, + ]); + expect(completed?.finalizedAt?.toString()).toBe( + row.finalizedAt?.toString() + ); + }); + + it("inserts and reads through the configured runtime transaction adapter", async () => { + const client = new Client( + new PgDriver(pool, { + schema, + transactionAdapters: [knexTransactionAdapter], + }) + ); + const tx = await db.transaction(); + let id: bigint; + try { + const { job } = await client.insert( + definition, + { + id: exactJsonNumber("9223372036854775807"), + }, + { tx } + ); + id = job.id; + expect((await client.jobs.get(id, { tx }))?.args).toEqual(job.args); + expect(await client.jobs.get(id)).toBeNull(); + } finally { + await tx.rollback(); + } + expect(await client.jobs.get(id)).toBeNull(); + }); + + it("keeps completion interceptors inside the Knex transaction", async () => { + let expectedTx: Knex.Transaction | undefined; + let intercepted = false; + const row = await runWorker( + async ({ completeTx }) => { + await db.transaction(async (tx) => { + expectedTx = tx; + await completeTx(tx); + expect(intercepted).toBe(true); + expect(await tx.withSchema(schema).table("application_row")).toEqual([ + { value: "intercepted" }, + ]); + throw new Error("rollback interceptor and completion"); + }); + }, + () => ({ + intercept: { + complete: async (context, next) => { + if (context.tx !== expectedTx) return next(); + intercepted = true; + await expectedTx + .withSchema(schema) + .table("application_row") + .insert({ value: "intercepted" }); + const result = await next(); + expect( + await db.withSchema(schema).table("application_row") + ).toHaveLength(0); + return result; + }, + }, + }) + ); + expect(intercepted).toBe(true); + expect(row.state).toBe("discarded"); + expect(await db.withSchema(schema).table("application_row")).toHaveLength( + 0 + ); + }); + + it("rejects completed and unconfigured transactions before writing", async () => { + const tx = await db.transaction(); + const unconfigured = new Client(new PgDriver(pool, { schema })); + try { + // @ts-expect-error Knex transactions require a configured adapter. + await expect(unconfigured.jobs.get(1n, { tx })).rejects.toMatchObject({ + code: "backend_mismatch", + }); + await tx.rollback(); + const row = await runWorker(async ({ completeTx }) => { + await expect(completeTx(tx)).rejects.toThrow("active Knex transaction"); + throw new Error("transaction closed"); + }); + expect(row.state).toBe("discarded"); + } finally { + if (!tx.isCompleted()) await tx.rollback(); + } + }); + + it("rejects completion from an attempt whose identity changed", async () => { + let rejected = false; + const row = await runWorker(async ({ completeTx, job }) => { + await db.transaction(async (tx) => { + await tx + .withSchema(schema) + .table("river_job") + .where({ id: job.id.toString() }) + .update({ attempt: job.attempt + 1 }); + await expect(completeTx(tx)).rejects.toThrow("no longer running"); + rejected = true; + // Roll the simulated rescue back, allowing this attempt to finish. + throw new Error("rollback attempt change"); + }); + }); + expect(rejected).toBe(true); + expect(row.state).toBe("discarded"); + expect(row.attempt).toBe(1); + }); + + it("rolls back application writes and completion together", async () => { + const row = await runWorker(async ({ completeTx }) => { + await db.transaction(async (tx) => { + await tx + .withSchema(schema) + .table("application_row") + .insert({ value: "rolled back" }); + await completeTx(tx, { output: { rolledBack: true } }); + throw new Error("rollback both writes"); + }); + }); + expect(row.state).toBe("discarded"); + expect(row.metadata).not.toHaveProperty("output"); + expect(await db.withSchema(schema).table("application_row")).toHaveLength( + 0 + ); + }); + + it("uses native savepoints without ending the outer transaction", async () => { + let nestedTx: Knex.Transaction | undefined; + const row = await runWorker(async ({ completeTx, job }) => { + await db.transaction(async (tx) => { + await tx + .withSchema(schema) + .table("application_row") + .insert({ value: "outer" }); + await expect( + tx.transaction(async (nested) => { + nestedTx = nested; + await nested + .withSchema(schema) + .table("application_row") + .insert({ value: "inner" }); + await completeTx(nested, { output: { inner: true } }); + throw new Error("rollback savepoint"); + }) + ).rejects.toThrow("rollback savepoint"); + expect( + ( + await tx + .withSchema(schema) + .table("river_job") + .where({ id: job.id.toString() }) + .first() + ).state + ).toBe("running"); + expect(tx.isCompleted()).toBe(false); + await completeTx(tx, { output: { outer: true } }); + // The original transaction remains usable after completeTx. + await tx + .withSchema(schema) + .table("application_row") + .insert({ value: "after completion" }); + }); + }); + expect(nestedTx).toBeDefined(); + expect(row.state).toBe("completed"); + expect(row.metadata.output).toEqual({ outer: true }); + expect( + await db.withSchema(schema).table("application_row").orderBy("value") + ).toEqual([{ value: "after completion" }, { value: "outer" }]); + }); +}); diff --git a/js/driver/knex/src/transaction.ts b/js/driver/knex/src/transaction.ts new file mode 100644 index 000000000..59f281d89 --- /dev/null +++ b/js/driver/knex/src/transaction.ts @@ -0,0 +1,86 @@ +import type { Knex } from "knex"; +import { ConfigurationError } from "riverqueue"; +import type { + PostgresTransactionAdapter, + PostgresTransactionQuery, + PostgresTransactionResult, +} from "riverqueue/unstable-driver"; + +/** + * 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 const knexTransactionAdapter: PostgresTransactionAdapter = + Object.freeze({ + matches: (value: unknown): value is Knex.Transaction => + typeof value === "function" && + "isTransaction" in value && + value.isTransaction === true && + "raw" in value && + typeof value.raw === "function", + query: (transaction: unknown, query: PostgresTransactionQuery) => { + validateKnexTransaction(transaction); + return queryKnex(transaction, query); + }, + validate: validateKnexTransaction, + }); + +/** Run River's fixed SQL through Knex without changing its bound values. */ +export async function queryKnex( + knex: Knex, + query: PostgresTransactionQuery +): Promise> { + // River's schema names cannot contain '$' or '?'. Its fixed SQL uses $n + // only for parameters. Escape literal '?' (including Postgres JSON + // operators) before introducing Knex's bindings, preserving repeated and + // out-of-order numbered parameters. + const bindings: Knex.RawBinding[] = []; + const statement = query.text + .replaceAll("?", "\\?") + .replace(/\$(\d+)/g, (_match, index: string) => { + bindings.push(query.values[Number(index) - 1] as Knex.RawBinding); + return "?"; + }); + const raw = knex + .raw>(statement, bindings) + .queryContext({ river: true }); + if (query.types !== undefined) raw.options({ types: query.types }); + return raw; +} + +/** Validate a Knex client configured for Postgres. */ +export function validateKnexClient(value: unknown): asserts value is Knex { + if ( + typeof value !== "function" || + !("raw" in value) || + typeof value.raw !== "function" || + !("client" in value) || + typeof value.client !== "object" || + value.client === null || + !("dialect" in value.client) || + value.client.dialect !== "postgresql" + ) { + throw new ConfigurationError( + 'Knex requires a Postgres client (client: "pg") for River' + ); + } +} + +/** Validate a live Knex transaction, including a nested savepoint. */ +export function validateKnexTransaction( + value: unknown +): asserts value is Knex.Transaction { + validateKnexClient(value); + if ( + !value.isTransaction || + !("isCompleted" in value) || + typeof value.isCompleted !== "function" || + (value as Knex.Transaction).isCompleted() + ) { + throw new ConfigurationError( + "River requires an active Knex transaction for { tx }" + ); + } +} diff --git a/js/driver/knex/tsconfig.json b/js/driver/knex/tsconfig.json new file mode 100644 index 000000000..88c94deed --- /dev/null +++ b/js/driver/knex/tsconfig.json @@ -0,0 +1,9 @@ +{ + "extends": "../../tsconfig.base.json", + "compilerOptions": { + "rootDir": "src", + "outDir": "dist" + }, + "exclude": ["src/**/*.integration.test.ts", "src/**/*.test.ts"], + "include": ["src"] +} diff --git a/js/driver/pg/README.md b/js/driver/pg/README.md index 82a97ea5c..a3940e835 100644 --- a/js/driver/pg/README.md +++ b/js/driver/pg/README.md @@ -57,6 +57,13 @@ The transaction costs two more round trips per call, for `BEGIN` and `COMMIT`. To insert many jobs, pass them to one `insertMany` call rather than calling `insert` for each. +To use native Knex transactions, configure +`new PgDriver(pool, { transactionAdapters: [knexTransactionAdapter] })`, +importing the adapter from `@riverqueue/driver-knex`. This enables both +`completeTx(tx)` in workers and `{ tx }` on client operations. See the +[Knex transaction example](../knex/README.md#complete-jobs-inside-knex-transactions). +The adapter is optional; this package has no Knex dependency. + Apply migrations explicitly with `@riverqueue/migrate` or `@riverqueue/cli` before starting workers. diff --git a/js/driver/pg/etc/driver-pg.api.md b/js/driver/pg/etc/driver-pg.api.md index a99311601..c1e58d8e9 100644 --- a/js/driver/pg/etc/driver-pg.api.md +++ b/js/driver/pg/etc/driver-pg.api.md @@ -20,26 +20,72 @@ declarations, and file layout are omitted. * leases from the pool, so a driver constructed from a single client * requires `{ tx }` for insertions. */ -export declare class PgDriver { +export declare class PgDriver { /** * Type-only marker: a full runtime driver whose transactions are any * node-postgres client (`pg.Client` or a `PoolClient` checked out from a - * pool) after `BEGIN`. + * pool) after `BEGIN`, plus the transaction types supported by configured + * `transactionAdapters`. */ readonly "~river"?: { readonly capability: "runtime"; - readonly transaction: ClientBase; + readonly transaction: ClientBase | Transaction; }; - constructor(client: Pool | PoolClient | PgClient, options?: PgDriverOptions); + constructor( + client: Pool | PoolClient | PgClient, + options?: PgDriverOptions + ); } ``` ## `PgDriverOptions` -```ts +````ts /** Construction options owned by the Postgres backend. */ -export interface PgDriverOptions { +export interface PgDriverOptions { /** Postgres schema containing River's tables and functions. */ schema?: string; + /** + * Enable native ORM transactions for `completeTx(tx)` in workers and + * `{ tx }` on client operations. Import the adapter from its driver package + * and configure it on each `PgDriver` that needs it. The ORM owns commit, + * rollback, and savepoints; River uses the supplied transaction only. + * + * Knex and the pg pool must connect to the same database. Configure the + * same River schema on the worker and insertion clients. Knex remains an + * optional dependency, installed with `@riverqueue/driver-knex`. + * + * @example + * ```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 = "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().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 } + * ); + * ``` + */ + transactionAdapters?: readonly PostgresTransactionAdapter[]; } -``` +```` diff --git a/js/driver/pg/src/database.ts b/js/driver/pg/src/database.ts index 8d272772e..06e3e91e6 100644 --- a/js/driver/pg/src/database.ts +++ b/js/driver/pg/src/database.ts @@ -8,14 +8,14 @@ import { postgresCapabilitiesFromRow, quoteIdentifier, type PostgresCapabilities, + type PostgresTransactionAdapter, + type PostgresTransactionResult, } from "riverqueue/unstable-driver"; import type { Client as PgClient, - ClientBase, Pool, PoolClient, QueryConfig, - QueryResult, QueryResultRow, } from "pg"; import { RiverError } from "riverqueue"; @@ -55,9 +55,15 @@ export class PgDatabase { */ #capabilities: PostgresCapabilities | undefined; readonly #client: PgQueryable; + readonly #transactionAdapters: readonly PostgresTransactionAdapter[]; /** Wrap a validated pool or client; `schema` is already validated. */ - constructor(client: Pool | PoolClient | PgClient, schema: string | null) { + constructor( + client: Pool | PoolClient | PgClient, + schema: string | null, + transactionAdapters: readonly PostgresTransactionAdapter[] = [] + ) { + this.#transactionAdapters = [...transactionAdapters]; this.#client = client; this.pool = isPool(client) ? client : null; this.schemaName = schema; @@ -72,7 +78,7 @@ export class PgDatabase { * the first result stored wins, and nothing is cached after a failure. */ async capabilities( - options?: PgOperationOptions | InsertDriverOptions + options?: PgOperationOptions | InsertDriverOptions ): Promise { if (this.#capabilities !== undefined) return this.#capabilities; const result = await this.query<{ @@ -167,8 +173,8 @@ export class PgDatabase { operation: string, text: string, values: unknown[], - options?: PgOperationOptions | InsertDriverOptions - ): Promise> { + options?: PgOperationOptions | InsertDriverOptions + ): Promise> { const queryable = this.resolveQueryable(operation, options); const config: QueryConfig = { text, @@ -198,7 +204,7 @@ export class PgDatabase { operation: string, text: string, values: unknown[] - ): Promise> { + ): Promise> { return this.withConnection(signal, (options) => this.query(operation, text, values, options) ); @@ -212,7 +218,7 @@ export class PgDatabase { */ async withConnection( signal: AbortSignal | undefined, - run: (options: { readonly tx?: ClientBase }) => Promise + run: (options: { readonly tx?: unknown }) => Promise ): Promise { if (signal === undefined || this.pool === null) return run({}); const acquiring = this.pool.connect(); @@ -240,9 +246,9 @@ export class PgDatabase { operation: string, text: string, values: unknown[], - options: { readonly signal?: AbortSignal; readonly tx?: ClientBase } = {}, + options: { readonly signal?: AbortSignal; readonly tx?: unknown } = {}, cancelPrefix?: string - ): Promise> { + ): Promise> { const signal = options.signal; if (signal === undefined) { return this.query( @@ -359,13 +365,30 @@ export class PgDatabase { /** The caller's transaction client when given, else the configured pool or client. */ resolveQueryable( operation: string, - options?: PgOperationOptions | InsertDriverOptions - ): PgQueryable { + options?: PgOperationOptions | InsertDriverOptions + ): { + query( + config: QueryConfig + ): Promise>; + } { if (options?.tx === undefined) return this.#client; + const tx = options.tx; + for (const adapter of this.#transactionAdapters) { + if (!adapter.matches(tx)) continue; + adapter.validate(tx); + return { + query: (config) => + adapter.query(tx, { + text: config.text, + types: config.types, + values: config.values ?? [], + }), + }; + } if (!isQueryable(options.tx)) { throw backendMismatchError( operation, - "the transaction is not a node-postgres client" + "the transaction is not a node-postgres client or a configured transaction adapter" ); } return options.tx; @@ -390,7 +413,7 @@ function isPool(value: Pool | PoolClient | PgClient): value is Pool { /** Whether a value can run node-postgres queries. */ export function isQueryable(value: unknown): value is PgQueryable { return ( - (typeof value === "object" || typeof value === "function") && + typeof value === "object" && value !== null && "query" in value && typeof value.query === "function" diff --git a/js/driver/pg/src/driver.ts b/js/driver/pg/src/driver.ts index 2eb2f8cae..3287ad65c 100644 --- a/js/driver/pg/src/driver.ts +++ b/js/driver/pg/src/driver.ts @@ -67,23 +67,24 @@ const RIVER_SCHEMA_RE = /^[A-Za-z_][A-Za-z0-9_]*$/; * leases from the pool, so a driver constructed from a single client * requires `{ tx }` for insertions. */ -export class PgDriver { +export class PgDriver { /** * Type-only marker: a full runtime driver whose transactions are any * node-postgres client (`pg.Client` or a `PoolClient` checked out from a - * pool) after `BEGIN`. + * pool) after `BEGIN`, plus the transaction types supported by configured + * `transactionAdapters`. */ declare readonly "~river"?: { readonly capability: "runtime"; - readonly transaction: ClientBase; + readonly transaction: ClientBase | Transaction; }; constructor( client: Pool | PoolClient | PgClient, - options: PgDriverOptions = {} + options: PgDriverOptions = {} ) { - const runtime = new PgRuntime(client, options); - registerDriver(this, runtime.driverRecord()); + const runtime = new PgRuntime(client, options); + registerDriver(this, runtime.driverRecord()); } } @@ -91,12 +92,12 @@ export class PgDriver { * @internal A registered driver whose operations are callable, for this * package's tests. */ -export function testPgDriver( +export function testPgDriver( client: Pool | PoolClient | PgClient, - options: PgDriverOptions = {} -): PgRuntime { - const runtime = new PgRuntime(client, options); - registerDriver(runtime, runtime.driverRecord()); + options: PgDriverOptions = {} +): PgRuntime { + const runtime = new PgRuntime(client, options); + registerDriver(runtime, runtime.driverRecord()); return runtime; } @@ -104,10 +105,12 @@ export function testPgDriver( * @internal The operations behind a {@link PgDriver}, which River reaches * through its private driver registry. */ -export class PgRuntime implements RuntimeDriver { +export class PgRuntime implements RuntimeDriver< + ClientBase | Transaction +> { declare readonly "~river"?: { readonly capability: "runtime"; - readonly transaction: ClientBase; + readonly transaction: ClientBase | Transaction; }; readonly backend = "postgres" as const; @@ -123,7 +126,7 @@ export class PgRuntime implements RuntimeDriver { constructor( client: Pool | PoolClient | PgClient, - options: PgDriverOptions = {} + options: PgDriverOptions = {} ) { if (!isQueryable(client)) { throw configurationError( @@ -136,12 +139,16 @@ export class PgRuntime implements RuntimeDriver { if (schema !== undefined) validateSchema(schema); this.#connection = client; - this.#db = new PgDatabase(client, schema ?? null); + this.#db = new PgDatabase( + client, + schema ?? null, + options.transactionAdapters + ); this.pool = this.#db.pool; } /** What River's private registry records about this driver. */ - driverRecord(): DriverRecord { + driverRecord(): DriverRecord { return { backend: this.backend, capability: "runtime", @@ -152,7 +159,7 @@ export class PgRuntime implements RuntimeDriver { operations: this, ...(this.pool === null ? {} - : { database: new PgPilotDatabase(this.#db, this.pool) }), + : { database: new PgPilotDatabase(this.#db, this.pool) }), }; } @@ -252,7 +259,7 @@ export class PgRuntime implements RuntimeDriver { */ jobClaim( params: JobClaimParams, - options?: JobClaimOptions + options?: JobClaimOptions ): Promise { return jobSql.jobClaim(this.#db, params, options); } @@ -264,7 +271,10 @@ export class PgRuntime implements RuntimeDriver { */ jobCompleteMany( commands: readonly JobCompletionCommand[], - options?: { readonly signal?: AbortSignal; readonly tx?: ClientBase } + options?: { + readonly signal?: AbortSignal; + readonly tx?: ClientBase | Transaction; + } ): Promise { return jobSql.jobCompleteMany(this.#db, commands, options); } @@ -304,7 +314,7 @@ export class PgRuntime implements RuntimeDriver { */ jobInsert( params: JobInsertParams, - options?: InsertDriverOptions + options?: InsertDriverOptions ): Promise { return jobSql.jobInsert(this.#db, params, options); } @@ -319,7 +329,7 @@ export class PgRuntime implements RuntimeDriver { */ jobInsertMany( params: readonly JobInsertParams[], - options?: InsertDriverOptions + options?: InsertDriverOptions ): Promise { return jobSql.jobInsertMany(this.#db, params, options); } @@ -585,7 +595,7 @@ export class PgRuntime implements RuntimeDriver { */ notifyInsert( queues: readonly string[], - options?: InsertDriverOptions + options?: InsertDriverOptions ): Promise { return notifySql.notifyInsert(this.#db, queues, options); } @@ -676,8 +686,8 @@ export class PgRuntime implements RuntimeDriver { * @internal */ async operationScope( - tx: ClientBase | undefined, - callback: (tx: ClientBase) => Promise + tx: ClientBase | Transaction | undefined, + callback: (tx: ClientBase | Transaction) => Promise ): Promise { if (tx !== undefined) return callback(tx); if (this.pool === null) { @@ -785,7 +795,7 @@ export class PgRuntime implements RuntimeDriver { leader: RuntimeLeader, attemptedBefore: Temporal.Instant, jobs: readonly RuntimeJobRescue[], - options?: InsertDriverOptions + options?: InsertDriverOptions ): Promise { return maintenanceSql.maintenanceRescue( this.#db, @@ -816,7 +826,7 @@ export class PgRuntime implements RuntimeDriver { maintenanceTransaction( leader: RuntimeLeader, batch: RuntimeMaintenanceBatch, - callback: (tx: ClientBase) => Promise + callback: (tx: ClientBase | Transaction) => Promise ): Promise { return maintenanceSql.maintenanceTransaction( this.#db, diff --git a/js/driver/pg/src/pilot.ts b/js/driver/pg/src/pilot.ts index 51b5c01ee..aa8b02d51 100644 --- a/js/driver/pg/src/pilot.ts +++ b/js/driver/pg/src/pilot.ts @@ -11,7 +11,7 @@ import type { PilotDatabase, } from "riverqueue/unstable-driver"; -import { isQueryable, type PgDatabase } from "./database.js"; +import type { PgDatabase } from "./database.js"; import { backendMismatchError } from "./errors.js"; import { abortablePromise, PgClientLease } from "./lease.js"; import { jobLoadClaimed } from "./sql/jobs.js"; @@ -19,7 +19,9 @@ import { jobDeleteFinalized } from "./sql/maintenance.js"; import { notifyMany } from "./sql/notify.js"; /** A pilot's view of a pool-backed `PgDriver`. */ -export class PgPilotDatabase implements PilotDatabase { +export class PgPilotDatabase implements PilotDatabase< + ClientBase | Transaction +> { readonly backend = "postgres"; readonly schema: string | null; @@ -33,7 +35,9 @@ export class PgPilotDatabase implements PilotDatabase { } async connection( - callback: (handle: ClientBase) => PromiseLike | Result, + callback: ( + handle: ClientBase | Transaction + ) => PromiseLike | Result, options: { readonly signal?: AbortSignal } = {} ): Promise { const acquiring = this.#pool.connect(); @@ -101,48 +105,48 @@ export class PgPilotDatabase implements PilotDatabase { async deleteFinalizedJobs( params: FinalizedJobDeleteParams, - options: { readonly tx?: ClientBase } = {} + options: { readonly tx?: ClientBase | Transaction } = {} ): Promise { return jobDeleteFinalized( this.#db, params, options.tx === undefined ? {} - : { tx: requireTransaction(options, "deleteFinalizedJobs") } + : { tx: this.#requireTransaction(options, "deleteFinalizedJobs") } ); } async loadClaimed( ids: readonly bigint[], - options: { readonly tx: ClientBase } + options: { readonly tx: ClientBase | Transaction } ): Promise { return jobLoadClaimed( this.#db, ids, - requireTransaction(options, "loadClaimed") + this.#requireTransaction(options, "loadClaimed") ); } async notify( topic: "control" | "insert", payloads: readonly string[], - options: { readonly tx: ClientBase } + options: { readonly tx: ClientBase | Transaction } ): Promise { - const tx = requireTransaction(options, "notify"); + const tx = this.#requireTransaction(options, "notify"); await notifyMany(this.#db, notificationTopic(topic), payloads, { tx }); } async transaction( - callback: (tx: ClientBase) => PromiseLike | Result, + callback: (tx: ClientBase | Transaction) => PromiseLike | Result, options: { readonly signal?: AbortSignal; - readonly tx?: ClientBase; + readonly tx?: ClientBase | Transaction; } = {} ): Promise { if (options.tx !== undefined) { // Like River for Go, run directly in the caller's transaction without // a savepoint; the caller rolls it back when this fails. - const tx = requireTransaction(options, "pilotTransaction"); + const tx = this.#requireTransaction(options, "pilotTransaction"); options.signal?.throwIfAborted(); const result = await callback(tx); options.signal?.throwIfAborted(); @@ -155,6 +159,20 @@ export class PgPilotDatabase implements PilotDatabase { callback ); } + + #requireTransaction( + options: { readonly tx?: unknown } | undefined, + operation: string + ): ClientBase | Transaction { + const tx = options?.tx; + if (tx === undefined) + throw backendMismatchError( + operation, + "a caller-owned transaction is required" + ); + this.#db.resolveQueryable(operation, { tx }); + return tx as ClientBase | Transaction; + } } function notificationTopic(topic: string): string { @@ -169,17 +187,3 @@ function notificationTopic(topic: string): string { ); } } - -function requireTransaction( - options: { readonly tx?: unknown } | undefined, - operation: string -): ClientBase { - const tx = options?.tx; - if (!isQueryable(tx)) { - throw backendMismatchError( - operation, - "the transaction is not a node-postgres client" - ); - } - return tx as ClientBase; -} diff --git a/js/driver/pg/src/sql/jobs.ts b/js/driver/pg/src/sql/jobs.ts index 5727ebe6a..12fac477d 100644 --- a/js/driver/pg/src/sql/jobs.ts +++ b/js/driver/pg/src/sql/jobs.ts @@ -1,7 +1,7 @@ /** Job insertion, claiming, completion, and administration queries. */ import { Buffer } from "node:buffer"; import { randomBytes } from "node:crypto"; -import type { ClientBase, QueryResult } from "pg"; +import type { QueryResult } from "pg"; import type { JobRow, JobState, JsonObject } from "riverqueue"; import { ValidationError } from "riverqueue"; import type { @@ -296,7 +296,7 @@ export async function jobGetCancelRequested( export async function jobClaim( db: PgDatabase, params: JobClaimParams, - options: { readonly signal?: AbortSignal; readonly tx?: ClientBase } = {} + options: { readonly signal?: AbortSignal; readonly tx?: unknown } = {} ): Promise { if (params.queues.length === 0) return { jobs: [] }; const names = new Set(); @@ -311,7 +311,7 @@ export async function jobClaim( const query = ( text: string, values: unknown[] - ): Promise> => + ): Promise, "rowCount" | "rows">> => options.tx === undefined ? db.queryAfterAcquire(options.signal, "jobClaim", text, values) : db.query("jobClaim", text, values, { tx: options.tx }); @@ -373,7 +373,7 @@ export async function jobClaim( export async function jobCompleteMany( db: PgDatabase, commands: readonly JobCompletionCommand[], - options?: { readonly signal?: AbortSignal; readonly tx?: ClientBase } + options?: { readonly signal?: AbortSignal; readonly tx?: unknown } ): Promise { if (commands.length === 0) return []; validateCompletionCommands(commands); @@ -629,7 +629,7 @@ export async function jobUpdate( export async function jobInsert( db: PgDatabase, params: JobInsertParams, - options?: InsertDriverOptions + options?: InsertDriverOptions ): Promise { const results = await jobInsertMany(db, [params], options); const result = results[0]; @@ -646,7 +646,7 @@ export async function jobInsert( export async function jobInsertMany( db: PgDatabase, params: readonly JobInsertParams[], - options?: InsertDriverOptions + options?: InsertDriverOptions ): Promise { if (params.length === 0) return []; @@ -861,7 +861,7 @@ function decodeClaimedJobs(rows: readonly PgJobRow[]): JobClaimResult { export async function jobLoadClaimed( db: PgDatabase, ids: readonly bigint[], - tx: ClientBase + tx: unknown ): Promise { const unique = new Set(); for (const id of ids) { diff --git a/js/driver/pg/src/sql/leader.ts b/js/driver/pg/src/sql/leader.ts index 55d428e6c..e13c0dc6b 100644 --- a/js/driver/pg/src/sql/leader.ts +++ b/js/driver/pg/src/sql/leader.ts @@ -133,7 +133,7 @@ export async function leaderResign( } function mapOneLeader( - result: QueryResult + result: Pick, "rows"> ): PgLeader | null { const row = result.rows[0]; return row === undefined diff --git a/js/driver/pg/src/sql/maintenance.ts b/js/driver/pg/src/sql/maintenance.ts index aaa609d9e..f01e101de 100644 --- a/js/driver/pg/src/sql/maintenance.ts +++ b/js/driver/pg/src/sql/maintenance.ts @@ -1,7 +1,7 @@ /** Leader-fenced maintenance: scheduling, rescue, cleanup, and reindexing. */ import { quoteIdentifier } from "riverqueue/unstable-driver"; import { Buffer } from "node:buffer"; -import type { ClientBase, PoolClient } from "pg"; +import type { PoolClient } from "pg"; import type { AttemptError, JobRow, JsonObject } from "riverqueue"; import type { FinalizedJobDeleteParams, @@ -391,9 +391,9 @@ export function maintenanceRescue( leader: RuntimeLeader, attemptedBefore: Temporal.Instant, jobs: readonly RuntimeJobRescue[], - tx?: ClientBase + tx?: unknown ): Promise { - const rescue = (transaction: ClientBase): Promise => + const rescue = (transaction: unknown): Promise => jobRescueMany( db, { @@ -449,7 +449,7 @@ export function maintenanceTransaction( db: PgDatabase, leader: RuntimeLeader, batch: RuntimeMaintenanceBatch, - callback: (tx: ClientBase) => Promise + callback: (tx: PoolClient) => Promise ): Promise { return withMaintenanceLeader( db, @@ -684,7 +684,7 @@ async function holdLeaderFence( db: PgDatabase, operation: string, leader: RuntimeLeader, - tx: ClientBase + tx: unknown ): Promise { const held = await db.query( `${operation}LeaderFence`, @@ -711,8 +711,8 @@ async function withLeaderFenceIn( db: PgDatabase, leader: RuntimeLeader, operation: string, - tx: ClientBase, - run: (transaction: ClientBase) => Promise + tx: unknown, + run: (transaction: unknown) => Promise ): Promise { if (!(await holdLeaderFence(db, operation, leader, tx))) return null; return run(tx); diff --git a/js/driver/pg/src/sql/rows.ts b/js/driver/pg/src/sql/rows.ts index 484628a61..79ac309a1 100644 --- a/js/driver/pg/src/sql/rows.ts +++ b/js/driver/pg/src/sql/rows.ts @@ -66,7 +66,9 @@ export interface PgScheduleRow extends PgJobRow { } /** Decode the first row of a result, or null when there is none. */ -export function mapOneJob(result: QueryResult): JobRow | null { +export function mapOneJob( + result: Pick, "rows"> +): JobRow | null { const row = result.rows[0]; return row === undefined ? null : toJobRow(row); } @@ -77,7 +79,9 @@ export function mapOneJob(result: QueryResult): JobRow | null { * that act on a job by ID use it, so an operator can cancel, retry, or delete * a row another engine wrote that River can't fully read. */ -export function mapOneJobPartial(result: QueryResult): JobRow | null { +export function mapOneJobPartial( + result: Pick, "rows"> +): JobRow | null { const row = result.rows[0]; return row === undefined ? null : toJobRowPartial(row).job; } diff --git a/js/driver/pg/src/types.ts b/js/driver/pg/src/types.ts index e3673d45b..bccf5b8f7 100644 --- a/js/driver/pg/src/types.ts +++ b/js/driver/pg/src/types.ts @@ -1,16 +1,58 @@ -import type { ClientBase } from "pg"; +import type { PostgresTransactionAdapter } from "riverqueue/unstable-driver"; import type { AttemptError, JobRow, JsonObject } from "riverqueue"; /** Construction options owned by the Postgres backend. */ -export interface PgDriverOptions { +export interface PgDriverOptions { /** Postgres schema containing River's tables and functions. */ schema?: string; + /** + * Enable native ORM transactions for `completeTx(tx)` in workers and + * `{ tx }` on client operations. Import the adapter from its driver package + * and configure it on each `PgDriver` that needs it. The ORM owns commit, + * rollback, and savepoints; River uses the supplied transaction only. + * + * Knex and the pg pool must connect to the same database. Configure the + * same River schema on the worker and insertion clients. Knex remains an + * optional dependency, installed with `@riverqueue/driver-knex`. + * + * @example + * ```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 = "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().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 } + * ); + * ``` + */ + transactionAdapters?: readonly PostgresTransactionAdapter[]; } /** Options common to Postgres semantic operations. */ export interface PgOperationOptions { /** Caller-owned transaction connection used for the entire operation. */ - tx?: ClientBase; + tx?: unknown; } /** Exact semantic inputs for cancelling a job. */ diff --git a/js/driver/prisma/src/driver.integration.test.ts b/js/driver/prisma/src/driver.integration.test.ts deleted file mode 100644 index 52d1eeba5..000000000 --- a/js/driver/prisma/src/driver.integration.test.ts +++ /dev/null @@ -1,268 +0,0 @@ -import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; -import { Pool } from "pg"; -import type { PoolClient } from "pg"; -import { - Client, - defineJob, - exactJsonNumber, - type InsertClient, - isExactJsonNumber, - type JsonValue, -} from "riverqueue"; -import { PrismaDriver } from "./driver.js"; -import type { PrismaClientLike } from "./driver.js"; - -const TEST_DATABASE_URL = - process.env.TEST_DATABASE_URL ?? - "postgres://localhost:5432/river_test?sslmode=disable"; -const filePrefix = `prisma_${Math.random().toString(36).slice(2, 8)}`; - -class PgPrismaAdapter implements PrismaClientLike { - constructor(private readonly pool: Pool | PoolClient) {} - - async $queryRawUnsafe( - sql: string, - ...values: unknown[] - ): Promise { - const result = await this.pool.query(sql, values); - return result.rows as T; - } -} - -/** A root client whose `$transaction` stands in for Prisma's. */ -class PgPrismaRootAdapter extends PgPrismaAdapter { - constructor(private readonly rootPool: Pool) { - super(rootPool); - } - - async $transaction( - callback: (tx: PrismaClientLike) => Promise - ): Promise { - const client = await this.rootPool.connect(); - try { - await client.query("BEGIN"); - try { - const result = await callback(new PgPrismaAdapter(client)); - await client.query("COMMIT"); - return result; - } catch (error: unknown) { - await client.query("ROLLBACK"); - throw error; - } - } finally { - client.release(); - } - } -} - -function job(suffix: string) { - return defineJob<{ key?: string; n?: number }>()({ - kind: `${filePrefix}_${suffix}`, - }); -} - -describe("PrismaDriver integration", () => { - let pool: Pool; - let client: InsertClient; - - beforeAll(() => { - pool = new Pool({ connectionString: TEST_DATABASE_URL }); - client = new Client(new PrismaDriver(new PgPrismaRootAdapter(pool))); - }); - - afterAll(async () => { - await pool.end(); - }); - - afterEach(async () => { - await pool.query("DELETE FROM river_job WHERE kind LIKE $1", [ - `${filePrefix}%`, - ]); - }); - - it("inserts and decodes an exact job row", async () => { - const definition = job("basic"); - - const result = await client.insert(definition, { key: "value" }); - - expect(result.job.id).toBeGreaterThan(0n); - expect(result.job.kind).toBe(definition.kind); - expect(result.job.args).toEqual({ key: "value" }); - expect(result.job.createdAt).toBeInstanceOf(Temporal.Instant); - expect(result.job.scheduledAt).toBeInstanceOf(Temporal.Instant); - expect(result.status).toBe("inserted"); - }); - - it("inserts with all options", async () => { - const definition = job("opts"); - const future = Temporal.Now.instant() - .add({ hours: 1 }) - .round({ roundingMode: "trunc", smallestUnit: "microsecond" }); - - const result = await client.insert( - definition, - { n: 42 }, - { - maxAttempts: 5, - metadata: { source: "integration" }, - priority: 3, - queue: "high_priority", - scheduledAt: future, - tags: ["tag_one", "tag_two"], - } - ); - - expect(result.job.maxAttempts).toBe(5); - expect(result.job.metadata).toEqual({ source: "integration" }); - expect(result.job.priority).toBe(3); - expect(result.job.queue).toBe("high_priority"); - expect(result.job.scheduledAt.toString()).toBe(future.toString()); - expect(result.job.state).toBe("scheduled"); - }); - - it("clamps a wide maxAttempts to Postgres's smallint like Go", async () => { - const result = await client.insert( - job("clamped"), - {}, - { maxAttempts: 40_000 } - ); - - expect(result.job.maxAttempts).toBe(32_767); - }); - - it("preserves order for heterogeneous batches", async () => { - const a = job("batch_a"); - const b = job("batch_b"); - - const results = await client.insertMany([ - { args: { n: 1 }, job: a }, - { args: { n: 2 }, job: b }, - ]); - - expect(results.map((result) => result.job.kind)).toEqual([a.kind, b.kind]); - expect(new Set(results.map((result) => result.job.id)).size).toBe(2); - }); - - it("returns the existing row for a unique conflict", async () => { - const definition = job("unique"); - const unique = { byArgs: true as const, byQueue: true as const }; - - const first = await client.insert(definition, { key: "same" }, { unique }); - const second = await client.insert(definition, { key: "same" }, { unique }); - - expect(first.status).toBe("inserted"); - expect(second.status).toBe("duplicate"); - expect(second.job.id).toBe(first.job.id); - }); - - it("keeps the existing job's kind on a unique skip of another kind", async () => { - const a = job("unique_kind_a"); - const b = job("unique_kind_b"); - const options = { unique: { byArgs: true, excludeKind: true } } as const; - - const first = await client.insert(a, { key: "same" }, options); - const single = await client.insert(b, { key: "same" }, options); - const [batched] = await client.insertMany([ - { args: { key: "same" }, job: b, options }, - ]); - - expect(first.status).toBe("inserted"); - for (const result of [single, batched]) { - expect(result.status).toBe("duplicate"); - expect(result.job.id).toBe(first.job.id); - expect(result.job.kind).toBe(a.kind); - } - const stored = await pool.query<{ kind: string }>( - "SELECT kind FROM river_job WHERE kind = ANY($1)", - [[a.kind, b.kind]] - ); - expect(stored.rows).toEqual([{ kind: a.kind }]); - }); - - it("uses the exact caller-owned transaction", async () => { - const definition = job("tx"); - const poolClient = await pool.connect(); - try { - await poolClient.query("BEGIN"); - const tx = new PgPrismaAdapter(poolClient); - await client.insert(definition, {}, { tx }); - await poolClient.query("ROLLBACK"); - } finally { - poolClient.release(); - } - - const result = await pool.query( - "SELECT count(*)::int AS count FROM river_job WHERE kind = $1", - [definition.kind] - ); - expect(result.rows[0]?.count).toBe(0); - }); - - it("keeps integers beyond JavaScript's safe range exact", async () => { - const definition = defineJob<{ id: JsonValue }>()({ - kind: `${filePrefix}_exact`, - }); - - const result = await client.insert( - definition, - { id: exactJsonNumber("9007199254740993") }, - { metadata: { tenant: exactJsonNumber("9007199254740995") } } - ); - - expect(isExactJsonNumber(result.job.args.id)).toBe(true); - expect(JSON.stringify(result.job.args)).toBe('{"id":9007199254740993}'); - expect(JSON.stringify(result.job.metadata)).toBe( - '{"tenant":9007199254740995}' - ); - }); - - it("notifies producers of available jobs when the transaction commits", async () => { - const available = job("notify_available"); - const scheduled = job("notify_scheduled"); - const queue = `${filePrefix}_notify`; - const listener = await pool.connect(); - const payloads: string[] = []; - listener.on("notification", (message) => { - if (message.payload !== undefined) payloads.push(message.payload); - }); - const poolClient = await pool.connect(); - try { - const schema = await listener.query<{ schema: string }>( - "SELECT current_schema()::text AS schema" - ); - await listener.query(`LISTEN "${schema.rows[0]!.schema}.river_insert"`); - await poolClient.query("BEGIN"); - const tx = new PgPrismaAdapter(poolClient); - await client.insertMany( - [ - { - args: {}, - job: scheduled, - options: { - queue: `${queue}_later`, - scheduledAt: Temporal.Now.instant().add({ hours: 1 }), - }, - }, - { args: {}, job: available, options: { queue } }, - ], - { tx } - ); - // NOTIFY is transactional: nothing is delivered before commit. - await listener.query("SELECT 1"); - expect(payloads).toEqual([]); - await poolClient.query("COMMIT"); - - const deadline = Date.now() + 2_000; - while (payloads.length === 0 && Date.now() < deadline) { - await listener.query("SELECT 1"); - } - expect(payloads.map((payload) => JSON.parse(payload))).toEqual([ - { queue }, - ]); - } finally { - poolClient.release(); - await listener.query("UNLISTEN *"); - listener.release(); - } - }); -}); diff --git a/js/driver/prisma/src/driver.ts b/js/driver/prisma/src/driver.ts index c7785f599..0c037b498 100644 --- a/js/driver/prisma/src/driver.ts +++ b/js/driver/prisma/src/driver.ts @@ -1,32 +1,11 @@ -import { Buffer } from "node:buffer"; -import { randomBytes } from "node:crypto"; -import type { DurationInput, JobRow } from "riverqueue"; -import { ConfigurationError, parseJsonObject } from "riverqueue"; -import type { - DriverInsertResult, - InsertDriver, - InsertDriverOptions, - JobInsertParams, - PostgresCapabilities, -} from "riverqueue/unstable-driver"; +import type { DurationInput } from "riverqueue"; +import { ConfigurationError } from "riverqueue"; import { - decodeAttemptError, - decodeJobState, durationToMilliseconds, - POSTGRES_CAPABILITIES_SQL, - postgresCapabilitiesFromRow, - postgresTimestamp, - quoteIdentifier, + PostgresInserter, registerDriver, - UNIQUE_INSERT_NONCE_KEY, - uniqueBitmaskFromStates, - uniqueBitmaskToStates, - uniqueInsertConflictSql, } from "riverqueue/unstable-driver"; -const RIVER_SCHEMA_MAX_BYTES = 46; -const RIVER_SCHEMA_RE = /^[A-Za-z_][A-Za-z0-9_]*$/; - /** The raw-query surface shared by Prisma clients and transaction clients. */ export interface PrismaClientLike { $queryRawUnsafe(query: string, ...values: unknown[]): Promise; @@ -66,33 +45,6 @@ export interface PrismaDriverOptions { }; } -/** - * Row shape returned by the insert query. JSON, timestamps, IDs, and bytes - * are selected as text so values are decoded exactly by River rather than - * by Prisma, which parses JSON numbers as doubles. - */ -interface PrismaJobRow extends Record { - args: string; - attempt: number; - attempted_at: string | null; - attempted_by: string[] | null; - created_at: string; - errors: string[] | null; - finalized_at: string | null; - id: string; - kind: string; - max_attempts: number; - metadata: string; - priority: number; - queue: string; - scheduled_at: string; - state: string; - tags: string[] | null; - unique_key: string | null; - unique_skipped_as_duplicate: boolean; - unique_states: string | null; -} - /** * River's insertion adapter for Prisma on Postgres. * @@ -139,311 +91,36 @@ export function testPrismaDriver( return inserter; } -/** - * @internal The operations behind a {@link PrismaDriver}, which River - * reaches through its private driver registry. - */ -export class PrismaInserter implements InsertDriver { - declare readonly "~river"?: { - readonly capability: "insert"; - readonly transaction: PrismaClientLike; - }; - - /** Identifies this driver's backend in River's errors and diagnostics. */ - readonly backend = "postgres-prisma-insert" as const; - - /** - * The server's capabilities once detected, cached for this driver's - * lifetime like River for Go's drivers. - */ - #capabilities: PostgresCapabilities | undefined; - readonly #prisma: PrismaClientLike; - readonly #qualifiedJobSequence: string; - readonly #schemaName: string | null; - readonly #schemaPrefix: string; - readonly #transactionOptions: PrismaTransactionOptions | undefined; - +/** @internal Prisma transport for the shared Postgres insertion engine. */ +export class PrismaInserter extends PostgresInserter { constructor(prisma: PrismaClientLike, options: PrismaDriverOptions = {}) { if (!isPrismaClientLike(prisma)) { throw new ConfigurationError( "PrismaDriver requires a Prisma client with $queryRawUnsafe" ); } - if (options.schema !== undefined) validateSchema(options.schema); - - this.#prisma = prisma; - this.#transactionOptions = transactionOptions(options.transactionOptions); - this.#schemaName = options.schema ?? null; - this.#schemaPrefix = - options.schema === undefined ? "" : `${quoteIdentifier(options.schema)}.`; - this.#qualifiedJobSequence = - options.schema === undefined - ? quoteIdentifier("river_job_id_seq") - : `${quoteIdentifier(options.schema)}.${quoteIdentifier("river_job_id_seq")}`; - } - - /** - * Insert one job, or return the existing job that holds its unique key. - * - * Part of River's unstable driver interface, which `Client` calls; - * applications insert through the client instead. Its parameter and result - * types come from `riverqueue/unstable-driver` and may change in any - * release. - */ - async jobInsert( - params: JobInsertParams, - options?: InsertDriverOptions - ): Promise { - const results = await this.jobInsertMany([params], options); - const result = results[0]; - if (result === undefined) { - throw new Error("Prisma returned no row for an inserted River job"); - } - return result; - } - - /** - * Insert an ordered batch of jobs atomically. - * - * Part of River's unstable driver interface, which `Client` calls; - * applications insert through the client instead. Its parameter and result - * types come from `riverqueue/unstable-driver` and may change in any - * release. - */ - async jobInsertMany( - params: readonly JobInsertParams[], - options?: InsertDriverOptions - ): Promise { - if (params.length === 0) return []; - - const queryable = options?.tx ?? this.#prisma; - // Without `xmax`, as on YugabyteDB, each row carries a nonce like - // SQLite's, and a returned row without its own nonce already existed. - const { uniqueInsertMode } = await this.#detect(queryable); - const nonces = - uniqueInsertMode === "metadata_nonce" - ? params.map(() => randomBytes(8).toString("hex")) - : null; - const jobTable = this.#name("river_job"); - const stateInBitmask = this.#name("river_job_state_in_bitmask"); - const stateType = this.#name("river_job_state"); - const sql = ` - WITH raw_job_data AS ( - SELECT - input_order, args, coalesce(created_at, now()) AS created_at, - kind, max_attempts, metadata, priority, queue, - coalesce(scheduled_at, now()) AS scheduled_at, - state_text AS state, - ARRAY(SELECT jsonb_array_elements_text(tags_json)) AS tags, - CASE WHEN unique_key_hex IS NULL THEN NULL - ELSE decode(unique_key_hex, 'hex') END AS unique_key, - unique_states_text::bit(8) AS unique_states - FROM unnest( - $1::jsonb[], $2::text[], $3::smallint[], $4::jsonb[], - $5::smallint[], $6::text[], $7::timestamptz[], $8::text[], - $9::jsonb[], $10::text[], $11::text[], $13::timestamptz[] - ) WITH ORDINALITY AS input( - args, kind, max_attempts, metadata, priority, queue, - scheduled_at, state_text, tags_json, unique_key_hex, - unique_states_text, created_at, input_order - ) - ), - normalized_job_data AS ( - SELECT - *, - unique_key IS NOT NULL - AND unique_states IS NOT NULL - AND ${stateInBitmask}(unique_states, state::${stateType}) - AS is_unique - FROM raw_job_data - ), - prepared_job_data AS ( - SELECT - *, - nextval($12::regclass) AS proposed_id - FROM normalized_job_data - ), - inserted_jobs AS ( - INSERT INTO ${jobTable} ( - id, args, created_at, kind, max_attempts, metadata, priority, - queue, scheduled_at, state, tags, unique_key, unique_states - ) - SELECT - proposed_id, args, created_at, kind, max_attempts, metadata, - priority, queue, scheduled_at, state::${stateType}, tags, - unique_key, unique_states - FROM prepared_job_data - ORDER BY input_order - ON CONFLICT (unique_key) - WHERE unique_key IS NOT NULL - AND unique_states IS NOT NULL - AND ${stateInBitmask}(unique_states, state) - DO UPDATE SET kind = river_job.kind - RETURNING *, ${uniqueInsertConflictSql(uniqueInsertMode)} AS conflicted - ) - SELECT - inserted_jobs.id::text AS id, - inserted_jobs.args::text AS args, - inserted_jobs.attempt, - ${utcText("inserted_jobs.attempted_at")} AS attempted_at, - inserted_jobs.attempted_by, - ${utcText("inserted_jobs.created_at")} AS created_at, - inserted_jobs.errors::text[] AS errors, - ${utcText("inserted_jobs.finalized_at")} AS finalized_at, - inserted_jobs.kind, - inserted_jobs.max_attempts, - inserted_jobs.metadata::text AS metadata, - inserted_jobs.priority, - inserted_jobs.queue, - ${utcText("inserted_jobs.scheduled_at")} AS scheduled_at, - inserted_jobs.state::text AS state, - inserted_jobs.tags, - encode(inserted_jobs.unique_key, 'hex') AS unique_key, - inserted_jobs.unique_states::text AS unique_states, - inserted_jobs.conflicted AS unique_skipped_as_duplicate - FROM prepared_job_data - JOIN inserted_jobs ON CASE - WHEN prepared_job_data.is_unique THEN - inserted_jobs.unique_key = prepared_job_data.unique_key - AND inserted_jobs.unique_states IS NOT NULL - AND ${stateInBitmask}(inserted_jobs.unique_states, inserted_jobs.state) - ELSE inserted_jobs.id = prepared_job_data.proposed_id - END - ORDER BY prepared_job_data.input_order - `; - - const rows = await queryable.$queryRawUnsafe( - sql, - params.map(({ encodedArgs }) => encodedArgs), - params.map(({ kind }) => kind), - params.map(({ maxAttempts }) => clampSmallint(maxAttempts)), - params.map(({ metadata }, index) => - JSON.stringify( - nonces === null - ? metadata - : { ...metadata, [UNIQUE_INSERT_NONCE_KEY]: nonces[index] } - ) - ), - params.map(({ priority }) => clampSmallint(priority)), - params.map(({ queue }) => queue), - params.map(({ scheduledAt }) => - scheduledAt === undefined ? null : postgresTimestamp(scheduledAt) - ), - params.map(({ state }) => state), - params.map(({ tags }) => JSON.stringify(tags)), - params.map(({ uniqueKey }) => - uniqueKey === null ? null : Buffer.from(uniqueKey).toString("hex") - ), - params.map(({ uniqueStates }) => - uniqueStates === null ? null : uniqueBitmaskFromStates(uniqueStates) - ), - this.#qualifiedJobSequence, - params.map(({ createdAt }) => - createdAt === undefined ? null : postgresTimestamp(createdAt) - ) - ); - if (rows.length !== params.length) { - throw new Error( - `Prisma returned ${rows.length} rows for ${params.length} River inserts` - ); - } - return rows.map((row, index) => { - const job = toJobRow(row); - const duplicate = - nonces === null - ? row.unique_skipped_as_duplicate - : job.metadata[UNIQUE_INSERT_NONCE_KEY] !== nonces[index]; - return { job, status: duplicate ? "duplicate" : "inserted" }; + const txOptions = transactionOptions(options.transactionOptions); + super({ + backend: "postgres-prisma-insert", + name: "Prisma", + operationScope: (tx, callback) => { + if (tx !== undefined) return callback(tx); + if (typeof prisma.$transaction !== "function") { + throw new ConfigurationError( + "the Prisma client given to PrismaDriver has no $transaction, so " + + "River can't open a transaction of its own for this operation; " + + "pass { tx } or construct PrismaDriver with a root PrismaClient" + ); + } + return txOptions === undefined + ? prisma.$transaction(callback) + : prisma.$transaction(callback, txOptions); + }, + query: (tx, sql, ...values) => + (tx ?? prisma).$queryRawUnsafe(sql, ...values), + ...(options.schema === undefined ? {} : { schema: options.schema }), }); } - - /** - * Notify producers of new jobs in each of `queues`. NOTIFY is - * transactional, so in a caller-owned transaction it's delivered only on - * commit. - * - * Part of River's unstable driver interface, which `Client` calls after - * inserting jobs. Its parameter types come from - * `riverqueue/unstable-driver` and may change in any release. - */ - async notifyInsert( - queues: readonly string[], - options?: InsertDriverOptions - ): Promise { - if (queues.length === 0) return; - const queryable = options?.tx ?? this.#prisma; - // A server without LISTEN/NOTIFY, like YugabyteDB by default, gets none. - if (!(await this.#detect(queryable)).supportsListenNotify) return; - // Counting the notifications' rows makes Postgres send each one while - // returning no `void` column for Prisma to decode. - await queryable.$queryRawUnsafe( - ` - WITH notifications AS ( - SELECT pg_notify( - concat(coalesce($1::text, current_schema()), '.', 'river_insert'), - concat('{"queue": ', to_json(queue)::text, '}') - ) - FROM unnest($2::text[]) AS queue - ) - SELECT count(*)::int AS notified FROM notifications - `, - this.#schemaName, - queues - ); - } - - /** - * Run one River operation in a transaction, like River for Go's - * `dbutil.WithTxV`. With `tx`, the operation joins that caller-owned - * transaction client. Otherwise it runs in an interactive transaction on - * the root Prisma client, which commits when `callback` resolves and rolls - * back when it rejects. - * - * @internal - */ - async operationScope( - tx: PrismaClientLike | undefined, - callback: (tx: PrismaClientLike) => Promise - ): Promise { - if (tx !== undefined) return callback(tx); - const prisma = this.#prisma; - if (typeof prisma.$transaction !== "function") { - throw new ConfigurationError( - "the Prisma client given to PrismaDriver has no $transaction, so " + - "River can't open a transaction of its own for this operation; " + - "pass { tx } or construct PrismaDriver with a root PrismaClient" - ); - } - const run = (transaction: PrismaClientLike): Promise => - callback(transaction); - return this.#transactionOptions === undefined - ? prisma.$transaction(run) - : prisma.$transaction(run, this.#transactionOptions); - } - - /** - * The server's capabilities, detected with `queryable` the first time. - * Concurrent first callers may each detect; the first result stored wins. - */ - async #detect(queryable: PrismaClientLike): Promise { - if (this.#capabilities !== undefined) return this.#capabilities; - const [row] = await queryable.$queryRawUnsafe< - { - product: unknown; - version_num: unknown; - yb_listen_notify_enabled: unknown; - }[] - >(POSTGRES_CAPABILITIES_SQL); - if (row === undefined) { - throw new Error("Postgres returned no server capabilities"); - } - this.#capabilities ??= postgresCapabilitiesFromRow(row); - return this.#capabilities; - } - - #name(value: string): string { - return `${this.#schemaPrefix}${quoteIdentifier(value)}`; - } } function transactionOptions( @@ -474,80 +151,3 @@ function isPrismaClientLike(value: unknown): value is PrismaClientLike { typeof value.$queryRawUnsafe === "function" ); } - -/** Parse a timestamp selected through {@link utcText}. */ -function parseUtcInstant(value: string): Temporal.Instant { - return Temporal.Instant.from(value); -} - -/** - * Decode an inserted row exactly. Every value arrives as text chosen by the - * query, so decoding cannot lose precision: an integer beyond 2^53 in args, - * metadata, or errors stays exact instead of making a committed insert - * throw. - */ -function toJobRow(row: PrismaJobRow): JobRow { - const state = decodeJobState(row.state); - if (!/^-?(0|[1-9]\d*)$/.test(row.id)) { - throw new TypeError(`invalid River job ID: ${JSON.stringify(row.id)}`); - } - - return { - args: parseJsonObject(row.args), - attempt: row.attempt, - attemptedAt: - row.attempted_at === null ? null : parseUtcInstant(row.attempted_at), - attemptedBy: row.attempted_by ?? [], - createdAt: parseUtcInstant(row.created_at), - errors: (row.errors ?? []).map(decodeAttemptError), - finalizedAt: - row.finalized_at === null ? null : parseUtcInstant(row.finalized_at), - id: BigInt(row.id), - kind: row.kind, - maxAttempts: row.max_attempts, - metadata: parseJsonObject(row.metadata), - priority: row.priority, - queue: row.queue, - scheduledAt: parseUtcInstant(row.scheduled_at), - state, - tags: row.tags ?? [], - uniqueKey: - row.unique_key === null - ? null - : new Uint8Array(Buffer.from(row.unique_key, "hex")), - uniqueStates: - row.unique_states === null - ? null - : uniqueBitmaskToStates(Number.parseInt(row.unique_states, 2)), - }; -} - -/** - * Select a timestamptz as UTC ISO 8601 text with microseconds, independent of - * the session's `DateStyle` and `TimeZone`. - */ -function utcText(column: string): string { - return `to_char(${column} AT TIME ZONE 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.US"Z"')`; -} - -function validateSchema(value: string): void { - if (!RIVER_SCHEMA_RE.test(value)) { - throw new TypeError( - "Postgres schema must start with a letter or underscore and contain only letters, numbers, and underscores" - ); - } - if (Buffer.byteLength(value, "utf8") > RIVER_SCHEMA_MAX_BYTES) { - throw new TypeError( - `Postgres schema must not exceed ${RIVER_SCHEMA_MAX_BYTES} bytes so River notification topics remain valid` - ); - } -} - -/** - * Clamps an integer to Postgres's 16-bit `smallint` columns. Like River - * for Go's Postgres drivers, an insert stores a wider `max_attempts` or - * `priority` as 32767 instead of failing; SQLite stores the full value. - */ -function clampSmallint(value: number): number { - return Math.min(value, 32_767); -} diff --git a/js/etc/riverqueue.api.md b/js/etc/riverqueue.api.md index 8418e3a83..17ceceea5 100644 --- a/js/etc/riverqueue.api.md +++ b/js/etc/riverqueue.api.md @@ -2638,6 +2638,11 @@ export interface RiverPlugin { */ export interface RiverTransactionRegistry {} +interface RiverTransactionRegistry { + /** Knex transactions on Postgres. */ + "@riverqueue/driver-knex": Knex.Transaction; +} + interface RiverTransactionRegistry { /** node-postgres clients (`pg.Client` or a pool's `PoolClient`). */ "@riverqueue/driver-pg": ClientBase; diff --git a/js/etc/riverqueue.unstable-driver.api.md b/js/etc/riverqueue.unstable-driver.api.md index 272ff2394..0f7678bcb 100644 --- a/js/etc/riverqueue.unstable-driver.api.md +++ b/js/etc/riverqueue.unstable-driver.api.md @@ -1940,6 +1940,84 @@ export declare function postgresCapabilitiesFromRow(row: { }): PostgresCapabilities; ``` +## `PostgresInserter` + +```ts +/** Shared insertion SQL and exact row decoding for Postgres ORM adapters. */ +export declare class PostgresInserter< + Transaction, +> implements InsertDriver { + readonly "~river"?: { + readonly capability: "insert"; + readonly transaction: Transaction; + }; + readonly backend: string; + constructor(options: PostgresInserterOptions); + /** + * Insert one job, or return the existing job that holds its unique key. + * + * Part of River's unstable driver interface, which `Client` calls; + * applications insert through the client instead. Its parameter and result + * types come from `riverqueue/unstable-driver` and may change in any + * release. + */ + jobInsert( + params: JobInsertParams, + options?: InsertDriverOptions + ): Promise; + /** + * Insert an ordered batch of jobs atomically. + * + * Part of River's unstable driver interface, which `Client` calls; + * applications insert through the client instead. Its parameter and result + * types come from `riverqueue/unstable-driver` and may change in any + * release. + */ + jobInsertMany( + params: readonly JobInsertParams[], + options?: InsertDriverOptions + ): Promise; + /** + * Notify producers of new jobs in each of `queues`. NOTIFY is + * transactional, so in a caller-owned transaction it's delivered only on + * commit. + * + * Part of River's unstable driver interface, which `Client` calls after + * inserting jobs. Its parameter types come from + * `riverqueue/unstable-driver` and may change in any release. + */ + notifyInsert( + queues: readonly string[], + options?: InsertDriverOptions + ): Promise; + /** Run the operation in the adapter's transaction scope. */ + operationScope( + tx: Transaction | undefined, + callback: (tx: Transaction) => Promise + ): Promise; +} +``` + +## `PostgresInserterOptions` + +```ts +/** Transport and transaction callbacks for Postgres insertion adapters. */ +export interface PostgresInserterOptions { + backend: string; + name: string; + operationScope( + tx: Transaction | undefined, + callback: (tx: Transaction) => Promise + ): Promise; + query( + tx: Transaction | undefined, + sql: string, + ...values: unknown[] + ): Promise; + schema?: string; +} +``` + ## `postgresTimestamp` ```ts @@ -1953,6 +2031,49 @@ export declare function postgresCapabilitiesFromRow(row: { export declare function postgresTimestamp(value: Temporal.Instant): string; ``` +## `PostgresTransactionAdapter` + +```ts +/** + * Query transport for a caller-owned Postgres transaction. First-party ORM + * packages provide these adapters; applications configure them on PgDriver. + * An adapter never begins, commits, rolls back, or releases the transaction. + */ +export interface PostgresTransactionAdapter { + /** Recognize the ORM's transaction handle, including a completed handle. */ + matches(transaction: unknown): transaction is Transaction; + /** Run River's SQL on that exact transaction with the supplied parsers. */ + query( + transaction: unknown, + query: PostgresTransactionQuery + ): Promise>; + /** Reject completed transactions and unsupported database dialects. */ + validate(transaction: unknown): void; +} +``` + +## `PostgresTransactionQuery` + +```ts +/** River's numbered-parameter SQL and query-scoped Postgres type parsers. */ +export interface PostgresTransactionQuery { + readonly text: string; + /** Driver-owned parsers; the adapter must pass these through unchanged. */ + readonly types?: unknown; + readonly values: readonly unknown[]; +} +``` + +## `PostgresTransactionResult` + +```ts +/** Query results used by River, independent of an ORM or pg package. */ +export interface PostgresTransactionResult { + readonly rowCount: number | null; + readonly rows: Row[]; +} +``` + ## `PreparedInsertParams` ```ts diff --git a/js/package.json b/js/package.json index a617dfbff..505c55202 100644 --- a/js/package.json +++ b/js/package.json @@ -99,6 +99,7 @@ "eslint": "^10.8.0", "eslint-config-prettier": "^10.1.8", "fast-check": "^4.10.2", + "knex": "^3.1.0", "knip": "^6.38.0", "license-checker-rseidelsohn": "^5.0.1", "pg": "^8.22.0", diff --git a/js/pnpm-lock.yaml b/js/pnpm-lock.yaml index 742e4fed1..08deb606b 100644 --- a/js/pnpm-lock.yaml +++ b/js/pnpm-lock.yaml @@ -43,6 +43,9 @@ importers: fast-check: specifier: ^4.10.2 version: 4.10.2 + knex: + specifier: ^3.1.0 + version: 3.3.0(pg@8.22.0) knip: specifier: ^6.38.0 version: 6.38.0 @@ -111,6 +114,21 @@ importers: specifier: ^6.0.3 version: 6.0.3 + driver/knex: + devDependencies: + knex: + specifier: ^3.1.0 + version: 3.3.0(pg@8.22.0) + pg: + specifier: ^8.22.0 + version: 8.22.0 + riverqueue: + specifier: workspace:0.2.0 + version: link:../.. + typescript: + specifier: ^6.0.3 + version: 6.0.3 + driver/pg: dependencies: postgres-array: @@ -1522,6 +1540,9 @@ packages: color-name@1.1.4: resolution: {integrity: sha512-dOy+3AuW3a2wNbZHIuMZpTcgjGuLU/uBL/ubcZF9OXbDo8ff4O8yVp5Bf0efS8uEoYo5q4Fx7dY9OgQGXgAsQA==} + colorette@2.0.19: + resolution: {integrity: sha512-3tlv/dIP7FWvj3BsbHrGLJ6l/oKh1O3TcgBqMn+yyCagOxc23fyzDS6HypQbgxWbkpDnf52p1LuR4eWDQ/K9WQ==} + commander@10.0.1: resolution: {integrity: sha512-y4Mg2tXshplEbSGzx7amzPwKKOCGuoSRP/CjEdwwk0FOGlUbq6lKuoyDZTNZkmxHdJtp54hdfY/JUrdL7Xfdug==} engines: {node: '>=14'} @@ -1600,6 +1621,15 @@ packages: resolution: {integrity: sha512-VqKjzBLejbSMT4IgbmVgDjpkYrNWUYJnbCGo874u7MMKIWsILRX+OpX/gTk8MqjpT1A/c6HY2dCA77ZN0lkQ2Q==} engines: {node: '>=12'} + debug@4.3.4: + resolution: {integrity: sha512-PRWFHuSU3eDtQJPvnNY7Jcket1j0t5OuOsFzPPzsekD52Zl8qUfFIPEiswXqIvHWGVHOgX+7G/vCNNhehwxfkQ==} + engines: {node: '>=6.0'} + peerDependencies: + supports-color: '*' + peerDependenciesMeta: + supports-color: + optional: true + debug@4.4.3: resolution: {integrity: sha512-RGwwWnwQvkVfavKVt22FGLw+xYSdzARwm0ru6DhTVA3umU5hZc28V3kO4stgYryrTlLpuvgI9GiijltAjNbcqA==} engines: {node: '>=6.0'} @@ -1665,6 +1695,10 @@ packages: resolution: {integrity: sha512-xUtoPkMggbz0MPyPiIWr1Kp4aeWJjDZ6SMvURhimjdZgsRuDplF5/s9hcgGhyXMhs+6vpnuoiZ2kFiu3FMnS8Q==} engines: {node: '>=18'} + es-errors@1.3.0: + resolution: {integrity: sha512-Zf5H2Kxt2xjTvbJvP2ZWLEICxA6j+hAmMzIlypy4xcBg1vKVnx89Wy0GbS+kf5cwCVFFzdCFh2XSCFNULS6csw==} + engines: {node: '>= 0.4'} + es-module-lexer@2.2.0: resolution: {integrity: sha512-3lGxdTXCLfe1MYfTz1y2ksAAUM4NAOP6rPEjxGJVKO7TZ5+tvHCaQWGpC4Y3IXvW3ece0Cz1cIP4FWBxOnGCTQ==} @@ -1704,6 +1738,10 @@ packages: jiti: optional: true + esm@3.2.25: + resolution: {integrity: sha512-U1suiZ2oDVWv4zPO56S0NcR5QriEahGtdN2OR6FiOG4WJvcjBVFB0qI4+eKoWFH483PKGuLuu6V8Z4T5g63UVA==} + engines: {node: '>=6'} + espree@11.2.0: resolution: {integrity: sha512-7p3DrVEIopW1B1avAGLuCSh1jubc01H2JHc8B4qqGblmg5gI9yumBgACjWo4JlIc04ufug4xJ3SQI8HkS/Rgzw==} engines: {node: ^20.19.0 || ^22.13.0 || >=24} @@ -1815,6 +1853,9 @@ packages: engines: {node: ^8.16.0 || ^10.6.0 || >=11.0.0} os: [darwin] + function-bind@1.1.2: + resolution: {integrity: sha512-7XHNxH7qX9xG5mIwxkhumTox/MIRNcOgDrxWsMt2pAr23WHp6MrRlN7FBSFpCpr+oVO0F744iUgR82nJMfG2SA==} + generate-function@2.3.1: resolution: {integrity: sha512-eeB5GfMNeevm/GRYq20ShmsaGcmI81kIX2K9XQx5miC8KdHaC6Jm0qQ8ZNeGOi7wYB8OsdxKs+Y2oVuTFuVwKQ==} @@ -1822,12 +1863,19 @@ packages: resolution: {integrity: sha512-DyFP3BM/3YHTQOCUL/w0OZHR0lpKeGrxotcHWcqNEdnltqFwXVfhEBQ94eIo34AfQpo0rGki4cyIiftY06h2Fg==} engines: {node: 6.* || 8.* || >= 10.*} + get-package-type@0.1.0: + resolution: {integrity: sha512-pjzuKtY64GYfWizNAJ0fr9VqttZkNiK2iS430LtIHzjBEr6bX8Am2zm4sW4Ro5wjWW5cAlRL1qAMTcXbjNAO2Q==} + engines: {node: '>=8.0.0'} + get-port-please@3.2.0: resolution: {integrity: sha512-I9QVvBw5U/hw3RmWpYKRumUeaDgxTPd401x364rLmWBJcOQ753eov1eTgzDqRG9bqFIfDc7gfzcQEWrUri3o1A==} get-tsconfig@4.14.3: resolution: {integrity: sha512-++QEw4DIY7WGoukz+/+A/8dGYPT9l9yIadnmSgZ8Rjr3YVSVDipQSO9CdnJo9ePqFqUUqh+wk9uIaoiAwsiPkA==} + getopts@2.3.0: + resolution: {integrity: sha512-5eDf9fuSXwxBL6q5HX+dhDj+dslFGWzU5thZ9kNKUkcPtaPdatmUFKwHFrLb/uf/WpA4BHET+AX3Scl56cAjpA==} + giget@3.3.1: resolution: {integrity: sha512-r+mvuDjrjMpsdw46Kmeydb8bdHm7wOKw8wNBtTndkjbPjgAp5oUJUxRE76wZFknxIPokfWvep2qSXK37aXE6zg==} hasBin: true @@ -1853,6 +1901,10 @@ packages: resolution: {integrity: sha512-EykJT/Q1KjTWctppgIAgfSO0tKVuZUjhgMr17kqTumMl6Afv3EISleU7qZUzoXDFTAHTDC4NOoG/ZxU3EvlMPQ==} engines: {node: '>=8'} + hasown@2.0.4: + resolution: {integrity: sha512-T2UbfbBEF32wiepXIsMlTW9+dDYC6wMh/t/vYA4tuOMKqWz/n3vr1NFSxQiyP+zk2mXsoMA/i/7qV6LKut1t1A==} + engines: {node: '>= 0.4'} + highlight.js@10.7.3: resolution: {integrity: sha512-tzcUFauisWKNHaRkN4Wjl/ZA07gENAjFl3J/c480dprkGTg5EQstgaNFqBfUqCq54kZRIEcreTsAgF/m2quD7A==} @@ -1902,10 +1954,18 @@ packages: resolution: {integrity: sha512-5Hh7Y1wQbvY5ooGgPbDaL5iYLAPzMTUrjMulskHLH6wnv/A+1q5rgEaiuqEjB+oxGXIVZs1FF+R/KPN3ZSQYYg==} engines: {node: '>=12'} + interpret@2.2.0: + resolution: {integrity: sha512-Ju0Bz/cEia55xDwUWEa8+olFpCiQoypjnQySseKtmjNrnps3P+xfpUmGr90T7yjlVJmOtybRvPXhKMbHr+fWnw==} + engines: {node: '>= 0.10'} + ip-address@10.7.3: resolution: {integrity: sha512-A1kdq/tSb5QjvKvAMgIoEvDBIgL7qaqVP/jkvSwYYRZ9iEzvPpopxp2wQfu3SuZRHtpHNxMn8Fs0bS+gf5Xmwg==} engines: {node: '>= 12'} + is-core-module@2.17.0: + resolution: {integrity: sha512-J/vG0zBCbIKOQFfufSwyXdMrsohyJIUNkrnmo6WZGzoM7tr/lsbfW5b2BvisL6zsyMzK9UxV9L6c7AoFbyXHOA==} + engines: {node: '>= 0.4'} + is-extglob@2.1.1: resolution: {integrity: sha512-SbKbANkN603Vi4jEZv49LeVJMn4yGwsbzZworEoyEiutsN3nJYdbO36zfhGJ6QEDpOZIFkDtnq5JRxmvl3jsoQ==} engines: {node: '>=0.10.0'} @@ -1982,6 +2042,40 @@ packages: keyv@4.5.4: resolution: {integrity: sha512-oxVHkHR/EJf2CNXnWxRLW6mg7JyCCUcG0DtEGmL2ctUo1PNTin1PUil+r/+4r5MpVgC/fn1kjsx7mjSujKqIpw==} + knex@3.3.0: + resolution: {integrity: sha512-LgWl031hNuLv9Lhxdd9093zULa2aNcoP04Bk29e5NidgRcEr/T0rAr2WYi4BhjTiCDoQiRU56meUE5rppLKZzQ==} + engines: {node: '>=16'} + hasBin: true + peerDependencies: + better-sqlite3: '*' + mariadb: '*' + mysql: '*' + mysql2: '*' + pg: '*' + pg-native: '*' + pg-query-stream: ^4.14.0 + sqlite3: '*' + tedious: '*' + peerDependenciesMeta: + better-sqlite3: + optional: true + mariadb: + optional: true + mysql: + optional: true + mysql2: + optional: true + pg: + optional: true + pg-native: + optional: true + pg-query-stream: + optional: true + sqlite3: + optional: true + tedious: + optional: true + knip@6.38.0: resolution: {integrity: sha512-umytiMCaZf02MV9SOO1FEDgbDf6h20VZk3ruJE0kh52kkGjYZFRUAXHJWYoLhvhoHGINO/7GXVJmQ2vxTBvl6A==} engines: {node: ^20.19.0 || >=22.12.0} @@ -2170,6 +2264,9 @@ packages: resolution: {integrity: sha512-tzzskb3bG8LvYGFF/mDTpq3jpI6Q9wc3LEmBaghu+DdCssd1FakN7Bc0hVNmEyGq1bq3RgfkCb3cmQLpNPOroA==} engines: {node: '>=4'} + ms@2.1.2: + resolution: {integrity: sha512-sGkPx+VjMtmA6MX27oA4FBFELFCZZ4S4XqeGOXCv68tT+jb3vk/RyaKWP0PTKyWtmLSM0b+adUTEvbs1PEaH2w==} + ms@2.1.3: resolution: {integrity: sha512-6FlzubTLZG3J2a/NVCAleEhjzq5oxgHyaCU9yYXvcLsvoVaHJq/s5xXI6/XXP6tz7R9xAOtHnSO/tXtF3WRTlA==} @@ -2312,6 +2409,9 @@ packages: resolution: {integrity: sha512-ojmeN0qd+y0jszEtoY48r0Peq5dwMEkIlCOu6Q5f41lfkswXuKtYrhgoTpLnyIcHm24Uhqx+5Tqm2InSwLhE6Q==} engines: {node: '>=8'} + path-parse@1.0.7: + resolution: {integrity: sha512-LDJzPVEEEPR+y48z93A0Ed0yXb8pAByGWo/k5YYdYgpY2/2EsOsksJrq7lOHxryrVOn1ejG6oAp8ahvOIQD8sw==} + path-scurry@2.0.2: resolution: {integrity: sha512-3O/iVVsJAPsOnpwWIeD+d6z/7PmqApyQePUtCndjatj/9I5LylHvt5qluFaBT3I5h3r1ejfR056c+FCv+NnNXg==} engines: {node: 18 || 20 || >=22} @@ -2328,6 +2428,9 @@ packages: pg-connection-string@2.14.0: resolution: {integrity: sha512-XwWDGcLRGCXAR8F/AM5bG7Q+A3Wm2s6QeEjlOKZLlH3UYcguiqCWKyWXVag5TLTIjR7oOJUY8kcADaZgWPyLeg==} + pg-connection-string@2.6.2: + resolution: {integrity: sha512-ch6OwaeaPYcova4kKZ15sbJ2hKb/VP48ZD2gE7i1J+L4MspCtBMAx8nMgz7bksc7IojCIIWuEhHibSMFH8m8oA==} + pg-int8@1.0.1: resolution: {integrity: sha512-WCtabS6t3c8SkpDBUlb1kjOs7l66xsGdKpIPZsg4wR+B3+u9UAum2odSsF9tnvxg80h4ZxLWMy4pRjOsFIqQpw==} engines: {node: '>=4.0.0'} @@ -2506,6 +2609,10 @@ packages: real-require@1.0.0: resolution: {integrity: sha512-P4nbQYQfePJxRSmY+v/KINxVucm4NF3p3s7pJveMTtom52FR4YGltUQLB8idDXwDDWW+eYrWDFbuzUnjoWHF7g==} + rechoir@0.8.0: + resolution: {integrity: sha512-/vxpCXddiX8NGfGO/mTafwjq4aFa/71pvamip0++IQk3zG8cbCj0fifNPrjjF1XMXUne91jL9OoxmdykoEtifQ==} + engines: {node: '>= 10.13.0'} + remeda@2.33.4: resolution: {integrity: sha512-ygHswjlc/opg2VrtiYvUOPLjxjtdKvjGz1/plDhkG66hjNjFr1xmfrs2ClNFo/E6TyUFiwYNh53bKV26oBoMGQ==} @@ -2517,9 +2624,18 @@ packages: resolution: {integrity: sha512-Xf0nWe6RseziFMu+Ap9biiUbmplq6S9/p+7w7YXP/JBHhrUDDUhwa+vANyubuqfZWTveU//DYVGsDG7RKL/vEw==} engines: {node: '>=0.10.0'} + resolve-from@5.0.0: + resolution: {integrity: sha512-qYg9KP24dD5qka9J47d0aVky0N+b4fTU89LN9iDnjB5waksiC49rvMB0PrUJQGoTmH50XPiqOvAjDfaijGxYZw==} + engines: {node: '>=8'} + resolve-pkg-maps@1.0.0: resolution: {integrity: sha512-seS2Tj26TBVOC2NIc2rOe2y2ZO7efxITtLZcGSOnHHNOQ7CkiUBfw0Iw2ck6xkIhPwLhKNLS8BO+hEpngQlqzw==} + resolve@1.22.13: + resolution: {integrity: sha512-Dj/cW9zBV2aRj6XzCxuGIs3ESZKv02jixne3q6jFZYZ3/Tg09gHHviSMDL4kObh9pqSahdRHEnDoaP2gn8YLhA==} + engines: {node: '>= 0.4'} + hasBin: true + ret@0.5.0: resolution: {integrity: sha512-I1XxrZSQ+oErkRR4jYbAyEEu2I0avBvvMM5JN+6EBprOGRCs63ENqZ3vjavq8fBw2+62G5LF5XelKwuJpcvcxw==} engines: {node: '>=10'} @@ -2673,10 +2789,18 @@ packages: resolution: {integrity: sha512-zFObLMyZeEwzAoKCyu1B91U79K2t7ApXuQfo8OuxwXLDgcKxuwM+YvcbIhm6QWqz7mHUH1TVytR1PwVVjEuMig==} engines: {node: '>=14.18'} + supports-preserve-symlinks-flag@1.0.0: + resolution: {integrity: sha512-ot0WnXS9fgdkgIcePe6RHNk1WA8+muPa6cSjeR3V8K27q9BB1rTE3R1p7Hv0z1ZyAc8s6Vvv8DIyWf681MAt0w==} + engines: {node: '>= 0.4'} + tar@7.5.22: resolution: {integrity: sha512-MFO/QzvtAOmJbkhOaCTvbGcFN9L9b+JunIsDwaKljSOdcLMea3NJ1k9Usz/rjdfSXTq4dfzfeS7W4p4YOAAHeA==} engines: {node: '>=18'} + tarn@3.1.2: + resolution: {integrity: sha512-3RTvqKZcK/17jnJ8rMKFXbyNogywTs1z0gVPPwFsJGX46rkmUHOdIaSQ/aVO1rS7nH+soiXiWk7rvUXxndm8Dg==} + engines: {node: '>=8.0.0'} + thenify-all@1.6.0: resolution: {integrity: sha512-RNxQH/qI8/t3thXJDwcstUO4zeqo64+Uy/+sNVRBx4Xn2OX+OZ9oP+iJnNFqplFra2ZUVeKCSa2oVWi3T4uVmA==} engines: {node: '>=0.8'} @@ -2688,6 +2812,10 @@ packages: resolution: {integrity: sha512-e2zZ96wSChazBsbENf/Pcm/4swHt2cEKQ92rhUjkL9GCKiTDJIaTBenjE/m9DXi0QBmTMDkFDdOomUy20A1tDQ==} engines: {node: '>=20'} + tildify@2.0.0: + resolution: {integrity: sha512-Cc+OraorugtXNfs50hU9KS369rFXCfgGLpfCfvlc+Ud5u6VWmUQsOAa9HbTvheQdYnrdJqqv1e5oIqXppMYnSw==} + engines: {node: '>=8'} + tinybench@2.9.0: resolution: {integrity: sha512-0+DUvqWMValLmha6lr4kD8iAMK1HzV0/aKnCtWb9v9641TnP/MFb7Pc2bxoxQjTXAErryXVgUOfv2YqNllqGeg==} @@ -4135,6 +4263,8 @@ snapshots: color-name@1.1.4: {} + colorette@2.0.19: {} + commander@10.0.1: {} common-ancestor-path@2.0.0: {} @@ -4201,6 +4331,10 @@ snapshots: dependencies: d3-array: 3.2.4 + debug@4.3.4: + dependencies: + ms: 2.1.2 + debug@4.4.3: dependencies: ms: 2.1.3 @@ -4242,6 +4376,8 @@ snapshots: environment@1.1.0: {} + es-errors@1.3.0: {} + es-module-lexer@2.2.0: {} escalade@3.2.0: {} @@ -4300,6 +4436,8 @@ snapshots: transitivePeerDependencies: - supports-color + esm@3.2.25: {} + espree@11.2.0: dependencies: acorn: 8.17.0 @@ -4399,18 +4537,24 @@ snapshots: fsevents@2.3.3: optional: true + function-bind@1.1.2: {} + generate-function@2.3.1: dependencies: is-property: 1.0.2 get-caller-file@2.0.5: {} + get-package-type@0.1.0: {} + get-port-please@3.2.0: {} get-tsconfig@4.14.3: dependencies: resolve-pkg-maps: 1.0.0 + getopts@2.3.0: {} + giget@3.3.1: {} glob-parent@6.0.2: @@ -4431,6 +4575,10 @@ snapshots: has-flag@4.0.0: {} + hasown@2.0.4: + dependencies: + function-bind: 1.1.2 + highlight.js@10.7.3: {} hosted-git-info@9.0.3: @@ -4473,8 +4621,14 @@ snapshots: internmap@2.0.3: {} + interpret@2.2.0: {} + ip-address@10.7.3: {} + is-core-module@2.17.0: + dependencies: + hasown: 2.0.4 + is-extglob@2.1.1: {} is-fullwidth-code-point@3.0.0: {} @@ -4530,6 +4684,27 @@ snapshots: dependencies: json-buffer: 3.0.1 + knex@3.3.0(pg@8.22.0): + dependencies: + colorette: 2.0.19 + commander: 10.0.1 + debug: 4.3.4 + escalade: 3.2.0 + esm: 3.2.25 + get-package-type: 0.1.0 + getopts: 2.3.0 + interpret: 2.2.0 + lodash: 4.18.1 + pg-connection-string: 2.6.2 + rechoir: 0.8.0 + resolve-from: 5.0.0 + tarn: 3.1.2 + tildify: 2.0.0 + optionalDependencies: + pg: 8.22.0 + transitivePeerDependencies: + - supports-color + knip@6.38.0: dependencies: fdir: 6.5.0(picomatch@4.0.7) @@ -4734,6 +4909,8 @@ snapshots: mri@1.2.0: {} + ms@2.1.2: {} + ms@2.1.3: {} mysql2@3.24.4(@types/node@26.1.1): @@ -4950,6 +5127,8 @@ snapshots: path-key@3.1.1: {} + path-parse@1.0.7: {} + path-scurry@2.0.2: dependencies: lru-cache: 11.5.2 @@ -4964,6 +5143,8 @@ snapshots: pg-connection-string@2.14.0: {} + pg-connection-string@2.6.2: {} + pg-int8@1.0.1: {} pg-pool@3.14.0(pg@8.22.0): @@ -5128,14 +5309,27 @@ snapshots: real-require@1.0.0: {} + rechoir@0.8.0: + dependencies: + resolve: 1.22.13 + remeda@2.33.4: {} require-directory@2.1.1: {} require-from-string@2.0.2: {} + resolve-from@5.0.0: {} + resolve-pkg-maps@1.0.0: {} + resolve@1.22.13: + dependencies: + es-errors: 1.3.0 + is-core-module: 2.17.0 + path-parse: 1.0.7 + supports-preserve-symlinks-flag: 1.0.0 + ret@0.5.0: {} retry@0.12.0: {} @@ -5297,6 +5491,8 @@ snapshots: has-flag: 4.0.0 supports-color: 7.2.0 + supports-preserve-symlinks-flag@1.0.0: {} + tar@7.5.22: dependencies: '@isaacs/fs-minipass': 4.0.1 @@ -5305,6 +5501,8 @@ snapshots: minizlib: 3.1.0 yallist: 5.0.0 + tarn@3.1.2: {} + thenify-all@1.6.0: dependencies: thenify: 3.3.1 @@ -5317,6 +5515,8 @@ snapshots: dependencies: real-require: 1.0.0 + tildify@2.0.0: {} + tinybench@2.9.0: {} tinyexec@1.3.0: {} diff --git a/js/scripts/api-reports.mjs b/js/scripts/api-reports.mjs index 843abbd8c..4ec7cbc8f 100644 --- a/js/scripts/api-reports.mjs +++ b/js/scripts/api-reports.mjs @@ -76,6 +76,11 @@ const reports = [ name: "@riverqueue/cli", output: "cli/etc/cli.api.md", }, + { + entry: "driver/knex/dist/index.d.ts", + name: "@riverqueue/driver-knex", + output: "driver/knex/etc/driver-knex.api.md", + }, { entry: "driver/pg/dist/index.d.ts", name: "@riverqueue/driver-pg", diff --git a/js/scripts/check-licenses.mjs b/js/scripts/check-licenses.mjs index 62e7645cc..0fb188dec 100644 --- a/js/scripts/check-licenses.mjs +++ b/js/scripts/check-licenses.mjs @@ -9,6 +9,7 @@ const repositoryRoot = resolve(dirname(fileURLToPath(import.meta.url)), ".."); const packageDirectories = [ ".", "migrate", + "driver/knex", "driver/pg", "driver/prisma", "driver/sqlite", diff --git a/js/scripts/check-packages.mjs b/js/scripts/check-packages.mjs index d97cbf098..0867fe1a3 100644 --- a/js/scripts/check-packages.mjs +++ b/js/scripts/check-packages.mjs @@ -37,6 +37,7 @@ const rootPackageJson = JSON.parse( const packageSpecs = [ { directory: ".", name: "riverqueue" }, { directory: "migrate", name: "@riverqueue/migrate" }, + { directory: "driver/knex", name: "@riverqueue/driver-knex" }, { directory: "driver/pg", name: "@riverqueue/driver-pg" }, { directory: "driver/prisma", name: "@riverqueue/driver-prisma" }, { directory: "driver/sqlite", name: "@riverqueue/driver-sqlite" }, @@ -393,6 +394,7 @@ async function verifyConsumers(archiveByPackageName) { ` import * as river from "riverqueue"; import * as cli from "@riverqueue/cli"; +import * as knexDriver from "@riverqueue/driver-knex"; import * as pg from "@riverqueue/driver-pg"; import * as prisma from "@riverqueue/driver-prisma"; import * as sqlite from "@riverqueue/driver-sqlite"; @@ -400,7 +402,7 @@ import * as migrate from "@riverqueue/migrate"; import * as testing from "@riverqueue/test"; import * as workerThreads from "@riverqueue/worker-threads"; -for (const packageNamespace of [river, cli, pg, prisma, sqlite, migrate, testing, workerThreads]) { +for (const packageNamespace of [river, cli, knexDriver, pg, prisma, sqlite, migrate, testing, workerThreads]) { if (Object.keys(packageNamespace).length === 0) throw new Error("empty package namespace"); } @@ -449,6 +451,7 @@ if (cron("@every 1s").next(start).toString() !== "2026-10-11T12:34:57Z") { const packageNames = [ "riverqueue", "@riverqueue/cli", + "@riverqueue/driver-knex", "@riverqueue/driver-pg", "@riverqueue/driver-prisma", "@riverqueue/driver-sqlite", @@ -565,6 +568,8 @@ import type { RunHandle, } from "riverqueue"; import { run as runCli } from "@riverqueue/cli"; +import { KnexDriver, knexTransactionAdapter } from "@riverqueue/driver-knex"; +import type { Knex } from "knex"; import { PgDriver } from "@riverqueue/driver-pg"; import { PrismaDriver, type PrismaClientLike } from "@riverqueue/driver-prisma"; import { @@ -586,12 +591,36 @@ const emailJob = defineJob({ }); declare const driver: RuntimeDriver; declare const prisma: PrismaClientLike; +declare const knex: Knex; +declare const knexTx: Knex.Transaction; +const knexClient = new Client(new KnexDriver(knex)); +void knexClient.insert(emailJob, { address: "a@example.com" }, { tx: knexTx }); +// @ts-expect-error Knex is an insertion-only driver. +void knexClient.start(); +// @ts-expect-error A root Knex instance is not a transaction. +void knexClient.insert(emailJob, { address: "a@example.com" }, { tx: knex }); declare const pgPool: Pool; declare const pgTx: PoolClient; declare const sqliteDatabase: DatabaseSync; const client = new Client(driver); const pgDriver = new PgDriver(pgPool); const pgClient = new Client(pgDriver); +const knexWorkers = new Workers().add(emailJob, async ({ completeTx, client }) => { + await completeTx(knexTx); + await client.insert(emailJob, { address: "followup@example.com" }, { tx: knexTx }); + // @ts-expect-error A root Knex instance is not a transaction. + await completeTx(knex); +}); +const knexRuntimeClient = new Client( + new PgDriver(pgPool, { transactionAdapters: [knexTransactionAdapter] }), + { workers: knexWorkers }, +); +void knexRuntimeClient.jobs.get(1n, { tx: knexTx }); +void knexRuntimeClient.jobs.get(1n, { tx: pgTx }); +// @ts-expect-error A root Knex instance is not a transaction. +void knexRuntimeClient.jobs.get(1n, { tx: knex }); +// @ts-expect-error Knex transactions require a configured adapter. +void pgClient.jobs.get(1n, { tx: knexTx }); const sqliteDriver = new SqliteDriver(sqliteDatabase); const sqliteClient = new Client(sqliteDriver); const migrators: Migrator[] = [ diff --git a/js/scripts/check-readme-snippets.mjs b/js/scripts/check-readme-snippets.mjs index a7b94a40a..390f2b6c0 100644 --- a/js/scripts/check-readme-snippets.mjs +++ b/js/scripts/check-readme-snippets.mjs @@ -13,6 +13,7 @@ const readmes = [ .sort() .map((name) => `docs/${name}`), "cli/README.md", + "driver/knex/README.md", "driver/pg/README.md", "driver/prisma/README.md", "driver/sqlite/README.md", @@ -93,6 +94,10 @@ function checkProgram(directory, virtualFiles, readme) { outDir: undefined, paths: { "@riverqueue/cli": [resolve(repositoryRoot, "cli/dist/index.d.ts")], + knex: [resolve(repositoryRoot, "node_modules/knex/types/index.d.ts")], + "@riverqueue/driver-knex": [ + resolve(repositoryRoot, "driver/knex/dist/index.d.ts"), + ], "@riverqueue/driver-pg": [ resolve(repositoryRoot, "driver/pg/dist/index.d.ts"), ], diff --git a/js/scripts/packed-tests/packages.test.mjs b/js/scripts/packed-tests/packages.test.mjs index c6489d870..b6b734022 100644 --- a/js/scripts/packed-tests/packages.test.mjs +++ b/js/scripts/packed-tests/packages.test.mjs @@ -20,6 +20,7 @@ const consumerDirectory = fileURLToPath(new URL("..", import.meta.url)); const SATELLITES = [ "@riverqueue/cli", + "@riverqueue/driver-knex", "@riverqueue/driver-pg", "@riverqueue/driver-prisma", "@riverqueue/driver-sqlite", diff --git a/js/scripts/packed-tests/postgres.test.mjs b/js/scripts/packed-tests/postgres.test.mjs index 2a6f14db5..14d00d1c1 100644 --- a/js/scripts/packed-tests/postgres.test.mjs +++ b/js/scripts/packed-tests/postgres.test.mjs @@ -3,8 +3,10 @@ import { randomBytes } from "node:crypto"; import process from "node:process"; import { after, before, describe, it } from "node:test"; +import { knexTransactionAdapter } from "@riverqueue/driver-knex"; import { PgDriver } from "@riverqueue/driver-pg"; import { createMigrator } from "@riverqueue/migrate"; +import knex from "knex"; import pg from "pg"; import { Client, @@ -47,6 +49,81 @@ describe( await pool.end(); }); + it("completes jobs atomically with native Knex transactions", async () => { + const db = knex({ client: "pg", connection: DATABASE_URL }); + try { + await db.schema.withSchema(schema).createTable("account", (table) => { + table.integer("id").primary(); + }); + const workers = new Workers().add( + accountJob, + async ({ completeTx, job }) => { + await db.transaction(async (tx) => { + await tx + .withSchema(schema) + .table("account") + .insert({ id: job.args.accountId }); + const completed = await completeTx(tx, { + output: { exact: exactJsonNumber(INT8_MAX.toString()) }, + }); + assert.equal(completed.state, "completed"); + assert.equal( + jsonNumberToBigInt(completed.metadata.output.exact), + INT8_MAX + ); + const { rows } = await pool.query( + `SELECT state FROM ${schema}.river_job WHERE id = $1`, + [job.id.toString()] + ); + assert.equal( + rows[0].state, + "running", + "completion is invisible before commit" + ); + if (job.args.accountId === 2) + throw new Error("roll back Knex transaction"); + }); + } + ); + const client = new Client( + new PgDriver(pool, { + schema, + transactionAdapters: [knexTransactionAdapter], + }), + { queues: { default: { maxWorkers: 1 } }, workers } + ); + const committed = await client.insert( + accountJob, + { accountId: 1 }, + { maxAttempts: 1 } + ); + const rolledBack = await client.insert( + accountJob, + { accountId: 2 }, + { maxAttempts: 1 } + ); + const events = await workUntilFinalized(client, [ + committed.job.id, + rolledBack.job.id, + ]); + assert.equal(events.get(committed.job.id)?.kind, "job_completed"); + assert.equal( + (await client.jobs.get(committed.job.id)).state, + "completed" + ); + assert.equal(events.get(rolledBack.job.id)?.kind, "job_failed"); + assert.equal( + (await client.jobs.get(rolledBack.job.id)).state, + "discarded" + ); + assert.deepEqual(await db.withSchema(schema).table("account"), [ + { id: 1 }, + ]); + } finally { + await db.destroy(); + } + }); + it("works exact int64 args end to end", async () => { const workers = new Workers().add(accountJob, ({ job, recordOutput }) => { recordOutput({ accountId: job.args.accountId }); diff --git a/js/src/internal/postgres-inserter.ts b/js/src/internal/postgres-inserter.ts new file mode 100644 index 000000000..676a2b2e8 --- /dev/null +++ b/js/src/internal/postgres-inserter.ts @@ -0,0 +1,422 @@ +import { Buffer } from "node:buffer"; +import { randomBytes } from "node:crypto"; +import type { JobRow } from "../job.js"; +import { parseJsonObject } from "../json.js"; +import type { + DriverInsertResult, + InsertDriver, + InsertDriverOptions, + JobInsertParams, +} from "../driver.js"; +import { decodeAttemptError, decodeJobState } from "../driver-codecs.js"; +import { + POSTGRES_CAPABILITIES_SQL, + postgresCapabilitiesFromRow, + UNIQUE_INSERT_NONCE_KEY, + uniqueInsertConflictSql, + type PostgresCapabilities, +} from "./postgres-capabilities.js"; +import { postgresTimestamp } from "./timestamp.js"; +import { quoteIdentifier } from "./sql.js"; +import { + uniqueBitmaskFromStates, + uniqueBitmaskToStates, +} from "../unique-bitmask.js"; + +const RIVER_SCHEMA_MAX_BYTES = 46; +const RIVER_SCHEMA_RE = /^[A-Za-z_][A-Za-z0-9_]*$/; + +/** Shared insertion SQL and exact row decoding for Postgres ORM adapters. */ +export class PostgresInserter< + Transaction, +> implements InsertDriver { + declare readonly "~river"?: { + readonly capability: "insert"; + readonly transaction: Transaction; + }; + + readonly backend: string; + #capabilities: PostgresCapabilities | undefined; + readonly #options: PostgresInserterOptions; + readonly #qualifiedJobSequence: string; + readonly #schemaName: string | null; + readonly #schemaPrefix: string; + + constructor(options: PostgresInserterOptions) { + if (options.schema !== undefined) validateSchema(options.schema); + this.backend = options.backend; + this.#options = options; + this.#schemaName = options.schema ?? null; + this.#schemaPrefix = + options.schema === undefined ? "" : `${quoteIdentifier(options.schema)}.`; + this.#qualifiedJobSequence = `${this.#schemaPrefix}${quoteIdentifier("river_job_id_seq")}`; + } + + /** + * Insert one job, or return the existing job that holds its unique key. + * + * Part of River's unstable driver interface, which `Client` calls; + * applications insert through the client instead. Its parameter and result + * types come from `riverqueue/unstable-driver` and may change in any + * release. + */ + async jobInsert( + params: JobInsertParams, + options?: InsertDriverOptions + ): Promise { + const results = await this.jobInsertMany([params], options); + const result = results[0]; + if (result === undefined) { + throw new Error( + `${this.#options.name} returned no row for an inserted River job` + ); + } + return result; + } + + /** + * Insert an ordered batch of jobs atomically. + * + * Part of River's unstable driver interface, which `Client` calls; + * applications insert through the client instead. Its parameter and result + * types come from `riverqueue/unstable-driver` and may change in any + * release. + */ + async jobInsertMany( + params: readonly JobInsertParams[], + options?: InsertDriverOptions + ): Promise { + if (params.length === 0) return []; + + const queryable = options?.tx; + // Without `xmax`, as on YugabyteDB, each row carries a nonce like + // SQLite's, and a returned row without its own nonce already existed. + const { uniqueInsertMode } = await this.#detect(queryable); + const nonces = + uniqueInsertMode === "metadata_nonce" + ? params.map(() => randomBytes(8).toString("hex")) + : null; + const jobTable = this.#name("river_job"); + const stateInBitmask = this.#name("river_job_state_in_bitmask"); + const stateType = this.#name("river_job_state"); + const sql = ` + WITH raw_job_data AS ( + SELECT + input_order, args, coalesce(created_at, now()) AS created_at, + kind, max_attempts, metadata, priority, queue, + coalesce(scheduled_at, now()) AS scheduled_at, + state_text AS state, + ARRAY(SELECT jsonb_array_elements_text(tags_json)) AS tags, + CASE WHEN unique_key_hex IS NULL THEN NULL + ELSE decode(unique_key_hex, 'hex') END AS unique_key, + unique_states_text::bit(8) AS unique_states + FROM unnest( + $1::jsonb[], $2::text[], $3::smallint[], $4::jsonb[], + $5::smallint[], $6::text[], $7::timestamptz[], $8::text[], + $9::jsonb[], $10::text[], $11::text[], $13::timestamptz[] + ) WITH ORDINALITY AS input( + args, kind, max_attempts, metadata, priority, queue, + scheduled_at, state_text, tags_json, unique_key_hex, + unique_states_text, created_at, input_order + ) + ), + normalized_job_data AS ( + SELECT + *, + unique_key IS NOT NULL + AND unique_states IS NOT NULL + AND ${stateInBitmask}(unique_states, state::${stateType}) + AS is_unique + FROM raw_job_data + ), + prepared_job_data AS ( + SELECT + *, + nextval($12::regclass) AS proposed_id + FROM normalized_job_data + ), + inserted_jobs AS ( + INSERT INTO ${jobTable} ( + id, args, created_at, kind, max_attempts, metadata, priority, + queue, scheduled_at, state, tags, unique_key, unique_states + ) + SELECT + proposed_id, args, created_at, kind, max_attempts, metadata, + priority, queue, scheduled_at, state::${stateType}, tags, + unique_key, unique_states + FROM prepared_job_data + ORDER BY input_order + ON CONFLICT (unique_key) + WHERE unique_key IS NOT NULL + AND unique_states IS NOT NULL + AND ${stateInBitmask}(unique_states, state) + DO UPDATE SET kind = river_job.kind + RETURNING *, ${uniqueInsertConflictSql(uniqueInsertMode)} AS conflicted + ) + SELECT + inserted_jobs.id::text AS id, + inserted_jobs.args::text AS args, + inserted_jobs.attempt, + ${utcText("inserted_jobs.attempted_at")} AS attempted_at, + inserted_jobs.attempted_by, + ${utcText("inserted_jobs.created_at")} AS created_at, + inserted_jobs.errors::text[] AS errors, + ${utcText("inserted_jobs.finalized_at")} AS finalized_at, + inserted_jobs.kind, + inserted_jobs.max_attempts, + inserted_jobs.metadata::text AS metadata, + inserted_jobs.priority, + inserted_jobs.queue, + ${utcText("inserted_jobs.scheduled_at")} AS scheduled_at, + inserted_jobs.state::text AS state, + inserted_jobs.tags, + encode(inserted_jobs.unique_key, 'hex') AS unique_key, + inserted_jobs.unique_states::text AS unique_states, + inserted_jobs.conflicted AS unique_skipped_as_duplicate + FROM prepared_job_data + JOIN inserted_jobs ON CASE + WHEN prepared_job_data.is_unique THEN + inserted_jobs.unique_key = prepared_job_data.unique_key + AND inserted_jobs.unique_states IS NOT NULL + AND ${stateInBitmask}(inserted_jobs.unique_states, inserted_jobs.state) + ELSE inserted_jobs.id = prepared_job_data.proposed_id + END + ORDER BY prepared_job_data.input_order + `; + + const rows = await this.#options.query( + queryable, + sql, + params.map(({ encodedArgs }) => encodedArgs), + params.map(({ kind }) => kind), + params.map(({ maxAttempts }) => clampSmallint(maxAttempts)), + params.map(({ metadata }, index) => + JSON.stringify( + nonces === null + ? metadata + : { ...metadata, [UNIQUE_INSERT_NONCE_KEY]: nonces[index] } + ) + ), + params.map(({ priority }) => clampSmallint(priority)), + params.map(({ queue }) => queue), + params.map(({ scheduledAt }) => + scheduledAt === undefined ? null : postgresTimestamp(scheduledAt) + ), + params.map(({ state }) => state), + params.map(({ tags }) => JSON.stringify(tags)), + params.map(({ uniqueKey }) => + uniqueKey === null ? null : Buffer.from(uniqueKey).toString("hex") + ), + params.map(({ uniqueStates }) => + uniqueStates === null ? null : uniqueBitmaskFromStates(uniqueStates) + ), + this.#qualifiedJobSequence, + params.map(({ createdAt }) => + createdAt === undefined ? null : postgresTimestamp(createdAt) + ) + ); + if (rows.length !== params.length) { + throw new Error( + `${this.#options.name} returned ${rows.length} rows for ${params.length} River inserts` + ); + } + return rows.map((row, index) => { + const job = toJobRow(row); + const duplicate = + nonces === null + ? row.unique_skipped_as_duplicate + : job.metadata[UNIQUE_INSERT_NONCE_KEY] !== nonces[index]; + return { job, status: duplicate ? "duplicate" : "inserted" }; + }); + } + + /** + * Notify producers of new jobs in each of `queues`. NOTIFY is + * transactional, so in a caller-owned transaction it's delivered only on + * commit. + * + * Part of River's unstable driver interface, which `Client` calls after + * inserting jobs. Its parameter types come from + * `riverqueue/unstable-driver` and may change in any release. + */ + async notifyInsert( + queues: readonly string[], + options?: InsertDriverOptions + ): Promise { + if (queues.length === 0) return; + const queryable = options?.tx; + // A server without LISTEN/NOTIFY, like YugabyteDB by default, gets none. + if (!(await this.#detect(queryable)).supportsListenNotify) return; + // Counting the notifications' rows makes Postgres send each one while + // returning no `void` column for the adapter to decode. + await this.#options.query( + queryable, + ` + WITH notifications AS ( + SELECT pg_notify( + concat(coalesce($1::text, current_schema()), '.', 'river_insert'), + concat('{"queue": ', to_json(queue)::text, '}') + ) + FROM unnest($2::text[]) AS queue + ) + SELECT count(*)::int AS notified FROM notifications + `, + this.#schemaName, + queues + ); + } + + /** Run the operation in the adapter's transaction scope. */ + async operationScope( + tx: Transaction | undefined, + callback: (tx: Transaction) => Promise + ): Promise { + return this.#options.operationScope(tx, callback); + } + + /** + * The server's capabilities, detected with `queryable` the first time. + * Concurrent first callers may each detect; the first result stored wins. + */ + async #detect( + queryable: Transaction | undefined + ): Promise { + if (this.#capabilities !== undefined) return this.#capabilities; + const [row] = await this.#options.query< + { + product: unknown; + version_num: unknown; + yb_listen_notify_enabled: unknown; + }[] + >(queryable, POSTGRES_CAPABILITIES_SQL); + if (row === undefined) { + throw new Error("Postgres returned no server capabilities"); + } + this.#capabilities ??= postgresCapabilitiesFromRow(row); + return this.#capabilities; + } + + #name(value: string): string { + return `${this.#schemaPrefix}${quoteIdentifier(value)}`; + } +} + +/** Transport and transaction callbacks for Postgres insertion adapters. */ +export interface PostgresInserterOptions { + backend: string; + name: string; + operationScope( + tx: Transaction | undefined, + callback: (tx: Transaction) => Promise + ): Promise; + query( + tx: Transaction | undefined, + sql: string, + ...values: unknown[] + ): Promise; + schema?: string; +} + +/** + * Row shape returned by the insert query. JSON, timestamps, IDs, and bytes + * are selected as text so values are decoded exactly by River rather than + * by a database adapter that may parse JSON numbers as doubles. + */ +interface PostgresInsertRow extends Record { + args: string; + attempt: number; + attempted_at: string | null; + attempted_by: string[] | null; + created_at: string; + errors: string[] | null; + finalized_at: string | null; + id: string; + kind: string; + max_attempts: number; + metadata: string; + priority: number; + queue: string; + scheduled_at: string; + state: string; + tags: string[] | null; + unique_key: string | null; + unique_skipped_as_duplicate: boolean; + unique_states: string | null; +} + +/** Parse a timestamp selected through {@link utcText}. */ +function parseUtcInstant(value: string): Temporal.Instant { + return Temporal.Instant.from(value); +} + +/** + * Decode an inserted row exactly. Every value arrives as text chosen by the + * query, so decoding cannot lose precision: an integer beyond 2^53 in args, + * metadata, or errors stays exact instead of making a committed insert + * throw. + */ +function toJobRow(row: PostgresInsertRow): JobRow { + const state = decodeJobState(row.state); + if (!/^-?(0|[1-9]\d*)$/.test(row.id)) { + throw new TypeError(`invalid River job ID: ${JSON.stringify(row.id)}`); + } + + return { + args: parseJsonObject(row.args), + attempt: row.attempt, + attemptedAt: + row.attempted_at === null ? null : parseUtcInstant(row.attempted_at), + attemptedBy: row.attempted_by ?? [], + createdAt: parseUtcInstant(row.created_at), + errors: (row.errors ?? []).map(decodeAttemptError), + finalizedAt: + row.finalized_at === null ? null : parseUtcInstant(row.finalized_at), + id: BigInt(row.id), + kind: row.kind, + maxAttempts: row.max_attempts, + metadata: parseJsonObject(row.metadata), + priority: row.priority, + queue: row.queue, + scheduledAt: parseUtcInstant(row.scheduled_at), + state, + tags: row.tags ?? [], + uniqueKey: + row.unique_key === null + ? null + : new Uint8Array(Buffer.from(row.unique_key, "hex")), + uniqueStates: + row.unique_states === null + ? null + : uniqueBitmaskToStates(Number.parseInt(row.unique_states, 2)), + }; +} + +/** + * Select a timestamptz as UTC ISO 8601 text with microseconds, independent of + * the session's `DateStyle` and `TimeZone`. + */ +function utcText(column: string): string { + return `to_char(${column} AT TIME ZONE 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.US"Z"')`; +} + +function validateSchema(value: string): void { + if (!RIVER_SCHEMA_RE.test(value)) { + throw new TypeError( + "Postgres schema must start with a letter or underscore and contain only letters, numbers, and underscores" + ); + } + if (Buffer.byteLength(value, "utf8") > RIVER_SCHEMA_MAX_BYTES) { + throw new TypeError( + `Postgres schema must not exceed ${RIVER_SCHEMA_MAX_BYTES} bytes so River notification topics remain valid` + ); + } +} + +/** + * Clamps an integer to Postgres's 16-bit `smallint` columns. Like River + * for Go's Postgres drivers, an insert stores a wider `max_attempts` or + * `priority` as 32767 instead of failing; SQLite stores the full value. + */ +function clampSmallint(value: number): number { + return Math.min(value, 32_767); +} diff --git a/js/src/internal/postgres-transaction.ts b/js/src/internal/postgres-transaction.ts new file mode 100644 index 000000000..78d64ac5c --- /dev/null +++ b/js/src/internal/postgres-transaction.ts @@ -0,0 +1,30 @@ +/** + * Query transport for a caller-owned Postgres transaction. First-party ORM + * packages provide these adapters; applications configure them on PgDriver. + * An adapter never begins, commits, rolls back, or releases the transaction. + */ +export interface PostgresTransactionAdapter { + /** Recognize the ORM's transaction handle, including a completed handle. */ + matches(transaction: unknown): transaction is Transaction; + /** Run River's SQL on that exact transaction with the supplied parsers. */ + query( + transaction: unknown, + query: PostgresTransactionQuery + ): Promise>; + /** Reject completed transactions and unsupported database dialects. */ + validate(transaction: unknown): void; +} + +/** River's numbered-parameter SQL and query-scoped Postgres type parsers. */ +export interface PostgresTransactionQuery { + readonly text: string; + /** Driver-owned parsers; the adapter must pass these through unchanged. */ + readonly types?: unknown; + readonly values: readonly unknown[]; +} + +/** Query results used by River, independent of an ORM or pg package. */ +export interface PostgresTransactionResult { + readonly rowCount: number | null; + readonly rows: Row[]; +} diff --git a/js/src/postgres-inserter.integration.test.ts b/js/src/postgres-inserter.integration.test.ts new file mode 100644 index 000000000..93e4f843c --- /dev/null +++ b/js/src/postgres-inserter.integration.test.ts @@ -0,0 +1,338 @@ +import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; +import { Pool } from "pg"; +import knex from "knex"; +import { KnexDriver } from "../driver/knex/src/index.js"; +import type { PoolClient } from "pg"; +import { + Client, + defineJob, + exactJsonNumber, + type InsertClient, + isExactJsonNumber, + type JsonValue, +} from "riverqueue"; +import { PrismaDriver } from "../driver/prisma/src/driver.js"; +import type { PrismaClientLike } from "../driver/prisma/src/driver.js"; + +const TEST_DATABASE_URL = + process.env.TEST_DATABASE_URL ?? + "postgres://localhost:5432/river_test?sslmode=disable"; +const filePrefix = `postgres_insert_${Math.random().toString(36).slice(2, 8)}`; + +class PgPrismaAdapter implements PrismaClientLike { + constructor(private readonly pool: Pool | PoolClient) {} + + async $queryRawUnsafe( + sql: string, + ...values: unknown[] + ): Promise { + const result = await this.pool.query(sql, values); + return result.rows as T; + } +} + +/** A root client whose `$transaction` stands in for Prisma's. */ +class PgPrismaRootAdapter extends PgPrismaAdapter { + constructor(private readonly rootPool: Pool) { + super(rootPool); + } + + async $transaction( + callback: (tx: PrismaClientLike) => Promise + ): Promise { + const client = await this.rootPool.connect(); + try { + await client.query("BEGIN"); + try { + const result = await callback(new PgPrismaAdapter(client)); + await client.query("COMMIT"); + return result; + } catch (error: unknown) { + await client.query("ROLLBACK"); + throw error; + } + } finally { + client.release(); + } + } +} + +function job(suffix: string) { + return defineJob<{ key?: string; n?: number }>()({ + kind: `${filePrefix}_${suffix}`, + }); +} + +interface TransactionHandle { + commit(): Promise; + release(): void; + rollback(): Promise; + tx: Transaction; +} + +function testPostgresInserter( + name: string, + setup: (pool: Pool) => { + begin(): Promise>; + client: InsertClient; + destroy(): Promise; + } +) { + describe(`${name} insertion conformance`, () => { + let pool: Pool; + let client: InsertClient; + let bundle: ReturnType; + + beforeAll(() => { + pool = new Pool({ connectionString: TEST_DATABASE_URL }); + bundle = setup(pool); + client = bundle.client; + }); + + afterAll(async () => { + await bundle.destroy(); + await pool.end(); + }); + + afterEach(async () => { + await pool.query("DELETE FROM river_job WHERE kind LIKE $1", [ + `${filePrefix}%`, + ]); + }); + + it("inserts and decodes an exact job row", async () => { + const definition = job("basic"); + + const result = await client.insert(definition, { key: "value" }); + + expect(result.job.id).toBeGreaterThan(0n); + expect(result.job.kind).toBe(definition.kind); + expect(result.job.args).toEqual({ key: "value" }); + expect(result.job.createdAt).toBeInstanceOf(Temporal.Instant); + expect(result.job.scheduledAt).toBeInstanceOf(Temporal.Instant); + expect(result.status).toBe("inserted"); + }); + + it("inserts with all options", async () => { + const definition = job("opts"); + const future = Temporal.Now.instant() + .add({ hours: 1 }) + .round({ roundingMode: "trunc", smallestUnit: "microsecond" }); + + const result = await client.insert( + definition, + { n: 42 }, + { + maxAttempts: 5, + metadata: { source: "integration" }, + priority: 3, + queue: "high_priority", + scheduledAt: future, + tags: ["tag_one", "tag_two"], + } + ); + + expect(result.job.maxAttempts).toBe(5); + expect(result.job.metadata).toEqual({ source: "integration" }); + expect(result.job.priority).toBe(3); + expect(result.job.queue).toBe("high_priority"); + expect(result.job.scheduledAt.toString()).toBe(future.toString()); + expect(result.job.state).toBe("scheduled"); + }); + + it("clamps a wide maxAttempts to Postgres's smallint like Go", async () => { + const result = await client.insert( + job("clamped"), + {}, + { maxAttempts: 40_000 } + ); + + expect(result.job.maxAttempts).toBe(32_767); + }); + + it("preserves order for heterogeneous batches", async () => { + const a = job("batch_a"); + const b = job("batch_b"); + + const results = await client.insertMany([ + { args: { n: 1 }, job: a }, + { args: { n: 2 }, job: b }, + ]); + + expect(results.map((result) => result.job.kind)).toEqual([ + a.kind, + b.kind, + ]); + expect(new Set(results.map((result) => result.job.id)).size).toBe(2); + }); + + it("returns the existing row for a unique conflict", async () => { + const definition = job("unique"); + const unique = { byArgs: true as const, byQueue: true as const }; + + const first = await client.insert( + definition, + { key: "same" }, + { unique } + ); + const second = await client.insert( + definition, + { key: "same" }, + { unique } + ); + + expect(first.status).toBe("inserted"); + expect(second.status).toBe("duplicate"); + expect(second.job.id).toBe(first.job.id); + }); + + it("keeps the existing job's kind on a unique skip of another kind", async () => { + const a = job("unique_kind_a"); + const b = job("unique_kind_b"); + const options = { unique: { byArgs: true, excludeKind: true } } as const; + + const first = await client.insert(a, { key: "same" }, options); + const single = await client.insert(b, { key: "same" }, options); + const [batched] = await client.insertMany([ + { args: { key: "same" }, job: b, options }, + ]); + + expect(first.status).toBe("inserted"); + for (const result of [single, batched]) { + expect(result.status).toBe("duplicate"); + expect(result.job.id).toBe(first.job.id); + expect(result.job.kind).toBe(a.kind); + } + const stored = await pool.query<{ kind: string }>( + "SELECT kind FROM river_job WHERE kind = ANY($1)", + [[a.kind, b.kind]] + ); + expect(stored.rows).toEqual([{ kind: a.kind }]); + }); + + it("uses the exact caller-owned transaction", async () => { + const definition = job("tx"); + const transaction = await bundle.begin(); + try { + const { tx } = transaction; + await client.insert(definition, {}, { tx }); + await transaction.rollback(); + } finally { + transaction.release(); + } + + const result = await pool.query( + "SELECT count(*)::int AS count FROM river_job WHERE kind = $1", + [definition.kind] + ); + expect(result.rows[0]?.count).toBe(0); + }); + + it("keeps integers beyond JavaScript's safe range exact", async () => { + const definition = defineJob<{ id: JsonValue }>()({ + kind: `${filePrefix}_exact`, + }); + + const result = await client.insert( + definition, + { id: exactJsonNumber("9007199254740993") }, + { metadata: { tenant: exactJsonNumber("9007199254740995") } } + ); + + expect(isExactJsonNumber(result.job.args.id)).toBe(true); + expect(JSON.stringify(result.job.args)).toBe('{"id":9007199254740993}'); + expect(JSON.stringify(result.job.metadata)).toBe( + '{"tenant":9007199254740995}' + ); + }); + + it("notifies producers of available jobs when the transaction commits", async () => { + const available = job("notify_available"); + const scheduled = job("notify_scheduled"); + const queue = `${filePrefix}_notify`; + const listener = await pool.connect(); + const payloads: string[] = []; + listener.on("notification", (message) => { + if (message.payload !== undefined) payloads.push(message.payload); + }); + const transaction = await bundle.begin(); + try { + const schema = await listener.query<{ schema: string }>( + "SELECT current_schema()::text AS schema" + ); + await listener.query(`LISTEN "${schema.rows[0]!.schema}.river_insert"`); + const { tx } = transaction; + await client.insertMany( + [ + { + args: {}, + job: scheduled, + options: { + queue: `${queue}_later`, + scheduledAt: Temporal.Now.instant().add({ hours: 1 }), + }, + }, + { args: {}, job: available, options: { queue } }, + ], + { tx } + ); + // NOTIFY is transactional: nothing is delivered before commit. + await listener.query("SELECT 1"); + expect(payloads).toEqual([]); + await transaction.commit(); + + const deadline = Date.now() + 2_000; + while (payloads.length === 0 && Date.now() < deadline) { + await listener.query("SELECT 1"); + } + expect(payloads.map((payload) => JSON.parse(payload))).toEqual([ + { queue }, + ]); + } finally { + transaction.release(); + await listener.query("UNLISTEN *"); + listener.release(); + } + }); + }); +} + +testPostgresInserter("Knex", () => { + const db = knex({ client: "pg", connection: TEST_DATABASE_URL }); + return { + begin: async () => { + const tx = await db.transaction(); + return { + commit: async () => { + await tx.commit(); + }, + release: () => {}, + rollback: async () => { + await tx.rollback(); + }, + tx, + }; + }, + client: new Client(new KnexDriver(db)), + destroy: () => db.destroy(), + }; +}); + +testPostgresInserter("Prisma", (pool) => ({ + begin: async () => { + const connection = await pool.connect(); + await connection.query("BEGIN"); + return { + commit: async () => { + await connection.query("COMMIT"); + }, + release: () => connection.release(), + rollback: async () => { + await connection.query("ROLLBACK"); + }, + tx: new PgPrismaAdapter(connection), + }; + }, + client: new Client(new PrismaDriver(new PgPrismaRootAdapter(pool))), + destroy: async () => {}, +})); diff --git a/js/src/unstable-driver.ts b/js/src/unstable-driver.ts index 2ff369189..566bb937a 100644 --- a/js/src/unstable-driver.ts +++ b/js/src/unstable-driver.ts @@ -171,3 +171,11 @@ export type { ReadonlyJsonObject, ReadonlyJsonValue, } from "./job-args-transform.js"; + +export { PostgresInserter } from "./internal/postgres-inserter.js"; +export type { PostgresInserterOptions } from "./internal/postgres-inserter.js"; +export type { + PostgresTransactionAdapter, + PostgresTransactionQuery, + PostgresTransactionResult, +} from "./internal/postgres-transaction.js"; diff --git a/js/tsconfig.tests.json b/js/tsconfig.tests.json index 7c29bc586..f93892c2a 100644 --- a/js/tsconfig.tests.json +++ b/js/tsconfig.tests.json @@ -20,6 +20,9 @@ ], "@riverqueue/worker-threads": [ "./worker-threads/src/index.ts" + ], + "@riverqueue/driver-knex": [ + "./driver/knex/src/index.ts" ] } }, diff --git a/js/typedoc.json b/js/typedoc.json index 4a11f0bc1..f49a46f20 100644 --- a/js/typedoc.json +++ b/js/typedoc.json @@ -3,6 +3,7 @@ "entryPoints": [ "src/index.ts", "cli/src/index.ts", + "driver/knex/src/index.ts", "driver/pg/src/index.ts", "driver/prisma/src/index.ts", "driver/sqlite/src/index.ts",