From 5b1bad4da298daec375cdc1b0555717b5aff2384 Mon Sep 17 00:00:00 2001 From: Brandur Date: Sat, 17 Oct 2026 00:39:22 -0400 Subject: [PATCH] JavaScript: Easy Prisma adapter for `completeTx` This one doesn't something similar to a piece of #1516 that we'd added. Currently, you can use Prisma on insertions relatively easily, but workers still run off `PgDriver`, making it somewhat unergonomic to invoke transactional completion (`completeTx`) from within a worker. Here, we add an adapter that makes it very easy. `prismaTransactionAdapter` is configured on `PgDriver`, and from there you've got easy access to a `completeTx` in any worker that'll take a Prisma `$transaction` object: import { PrismaPg } from "@prisma/adapter-pg"; import { PgDriver } from "@riverqueue/driver-pg"; import { prismaTransactionAdapter } from "@riverqueue/driver-prisma"; import { Pool } from "pg"; import { Client, defineJob, Workers } from "riverqueue"; import { PrismaClient, type Prisma } from "./generated/prisma/client.js"; const connection = { connectionString: process.env.DATABASE_URL }; const prisma = new PrismaClient({ adapter: new PrismaPg(connection) }); const syncAccount = defineJob<{ accountId: string }>()({ kind: "sync_account", }); const workers = new Workers().add( syncAccount, async ({ completeTx, job }) => { await prisma.$transaction(async (tx) => { await tx.account.update({ where: { id: job.args.accountId }, data: { synced: true }, }); await completeTx(tx); }); } ); const client = new Client( new PgDriver(new Pool(connection), { transactionAdapters: [prismaTransactionAdapter], }), { queues: { default: { maxWorkers: 10 } }, workers } ); await client.start(); --- js/CHANGELOG.md | 1 + js/docs/README.md | 5 + js/driver/pg/README.md | 7 + js/driver/pg/etc/driver-pg.api.md | 38 +- js/driver/pg/src/database.ts | 22 +- js/driver/pg/src/sql/jobs.ts | 48 +- js/driver/pg/src/sql/leader.ts | 2 + js/driver/pg/src/sql/maintenance.ts | 12 +- js/driver/pg/src/sql/queues.ts | 16 +- js/driver/pg/src/text-query.test.ts | 39 ++ js/driver/pg/src/text-query.ts | 84 +++ js/driver/pg/src/types.ts | 38 +- js/driver/prisma/README.md | 77 ++- js/driver/prisma/etc/driver-prisma.api.md | 36 +- js/driver/prisma/src/driver.ts | 6 +- js/driver/prisma/src/index.ts | 6 +- .../prisma/src/runtime.integration.test.ts | 509 ++++++++++++++++++ js/driver/prisma/src/transaction.test.ts | 143 +++++ js/driver/prisma/src/transaction.ts | 116 ++++ js/etc/riverqueue.api.md | 2 +- js/etc/riverqueue.unstable-driver.api.md | 24 +- js/examples/prisma/README.md | 4 +- js/examples/prisma/package.json | 3 + js/examples/prisma/src/index.ts | 73 ++- js/pnpm-lock.yaml | 9 + js/scripts/check-packages.mjs | 21 +- js/src/internal/postgres-transaction.ts | 20 +- js/src/unstable-driver.ts | 1 + 28 files changed, 1299 insertions(+), 63 deletions(-) create mode 100644 js/driver/pg/src/text-query.test.ts create mode 100644 js/driver/pg/src/text-query.ts create mode 100644 js/driver/prisma/src/runtime.integration.test.ts create mode 100644 js/driver/prisma/src/transaction.test.ts create mode 100644 js/driver/prisma/src/transaction.ts diff --git a/js/CHANGELOG.md b/js/CHANGELOG.md index d3f7f1f91..d9c85785f 100644 --- a/js/CHANGELOG.md +++ b/js/CHANGELOG.md @@ -10,6 +10,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 native Prisma interactive transactions with `completeTx` in workers and `{ tx }` on Postgres client operations. Configure `prismaTransactionAdapter` from `@riverqueue/driver-prisma` on `PgDriver` to commit application writes and job completion together. [PR #1517](https://github.com/riverqueue/river/pull/1517). - 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/docs/README.md b/js/docs/README.md index 6e786fb69..d14c102f4 100644 --- a/js/docs/README.md +++ b/js/docs/README.md @@ -366,6 +366,11 @@ 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`. +For native Prisma transactions, configure `prismaTransactionAdapter` on the +worker's `PgDriver` and use `new Workers()`. See the +[Prisma example](../driver/prisma/README.md#complete-jobs-inside-prisma-transactions). +Installing the Prisma driver alone does not enable its transactions on `PgDriver`. + ## Run workers Configure queues and start the client. `maxWorkers` bounds concurrent diff --git a/js/driver/pg/README.md b/js/driver/pg/README.md index a3940e835..4d2873bb5 100644 --- a/js/driver/pg/README.md +++ b/js/driver/pg/README.md @@ -64,6 +64,13 @@ importing the adapter from `@riverqueue/driver-knex`. This enables both [Knex transaction example](../knex/README.md#complete-jobs-inside-knex-transactions). The adapter is optional; this package has no Knex dependency. +To use native Prisma transactions, configure +`new PgDriver(pool, { transactionAdapters: [prismaTransactionAdapter] })`, +importing the adapter from `@riverqueue/driver-prisma`. This enables both +`completeTx(tx)` in workers and `{ tx }` on client operations. See the +[Prisma transaction example](../prisma/README.md#complete-jobs-inside-prisma-transactions). +The adapter is optional; this package has no Prisma 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 c1e58d8e9..8a1610d9f 100644 --- a/js/driver/pg/etc/driver-pg.api.md +++ b/js/driver/pg/etc/driver-pg.api.md @@ -51,9 +51,10 @@ export interface PgDriverOptions { * 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`. + * The ORM and the pg pool must connect to the same database. Configure the + * same River schema on the worker and insertion clients. Knex and Prisma + * remain optional dependencies, installed with `@riverqueue/driver-knex` + * and `@riverqueue/driver-prisma`, respectively. * * @example * ```ts @@ -85,6 +86,37 @@ export interface PgDriverOptions { * { queues: { default: { maxWorkers: 10 } }, workers } * ); * ``` + * + * @example + * ```ts + * import { Pool } from "pg"; + * import { Client, defineJob, Workers } from "riverqueue"; + * import { prismaTransactionAdapter } from "@riverqueue/driver-prisma"; + * import { PgDriver } from "@riverqueue/driver-pg"; + * import type { Prisma, PrismaClient } from "./generated/prisma/client.js"; + * + * declare const prisma: PrismaClient; // Your generated application client. + * const pool = new Pool({ connectionString: process.env.DATABASE_URL }); + * const syncAccount = defineJob<{ accountId: string }>()({ + * kind: "sync_account", + * }); + * const workers = new Workers().add( + * syncAccount, + * async ({ completeTx, job }) => { + * await prisma.$transaction(async (tx) => { + * await tx.account.update({ + * where: { id: job.args.accountId }, + * data: { synced: true }, + * }); + * await completeTx(tx); + * }); + * } + * ); + * const client = new Client( + * new PgDriver(pool, { transactionAdapters: [prismaTransactionAdapter] }), + * { 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 06e3e91e6..e6221500c 100644 --- a/js/driver/pg/src/database.ts +++ b/js/driver/pg/src/database.ts @@ -10,6 +10,7 @@ import { type PostgresCapabilities, type PostgresTransactionAdapter, type PostgresTransactionResult, + type PostgresTransactionTextQuery, } from "riverqueue/unstable-driver"; import type { Client as PgClient, @@ -27,6 +28,11 @@ import { } from "./errors.js"; import { PG_EXACT_TYPES } from "./exact-types.js"; import { abortablePromise, PgClientLease } from "./lease.js"; +import { + postgresQueryText, + postgresTextQuery, + type PgStatement, +} from "./text-query.js"; import type { PgOperationOptions } from "./types.js"; /** Postgres's maximum identifier length (`NAMEDATALEN - 1`). */ @@ -171,13 +177,15 @@ export class PgDatabase { */ async query( operation: string, - text: string, + text: PgStatement, values: unknown[], options?: PgOperationOptions | InsertDriverOptions ): Promise> { - const queryable = this.resolveQueryable(operation, options); + const queryable = this.resolveQueryable(operation, options, () => + postgresTextQuery(text) + ); const config: QueryConfig = { - text, + text: postgresQueryText(text), types: PG_EXACT_TYPES, values, }; @@ -202,7 +210,7 @@ export class PgDatabase { async queryAfterAcquire( signal: AbortSignal | undefined, operation: string, - text: string, + text: PgStatement, values: unknown[] ): Promise> { return this.withConnection(signal, (options) => @@ -244,7 +252,7 @@ export class PgDatabase { */ async queryAbortable( operation: string, - text: string, + text: PgStatement, values: unknown[], options: { readonly signal?: AbortSignal; readonly tx?: unknown } = {}, cancelPrefix?: string @@ -365,7 +373,8 @@ export class PgDatabase { /** The caller's transaction client when given, else the configured pool or client. */ resolveQueryable( operation: string, - options?: PgOperationOptions | InsertDriverOptions + options?: PgOperationOptions | InsertDriverOptions, + textResult?: () => PostgresTransactionTextQuery ): { query( config: QueryConfig @@ -380,6 +389,7 @@ export class PgDatabase { query: (config) => adapter.query(tx, { text: config.text, + ...(textResult === undefined ? {} : { textResult }), types: config.types, values: config.values ?? [], }), diff --git a/js/driver/pg/src/sql/jobs.ts b/js/driver/pg/src/sql/jobs.ts index 12fac477d..37b91ecb6 100644 --- a/js/driver/pg/src/sql/jobs.ts +++ b/js/driver/pg/src/sql/jobs.ts @@ -69,7 +69,8 @@ export async function jobCancelWithOptions( ): Promise { const jobTable = db.table("river_job"); const { supportsListenNotify } = await db.capabilities(options); - const sql = ` + const sql = { + with: ` WITH locked_job AS ( SELECT id, queue, state, finalized_at FROM ${jobTable} @@ -112,6 +113,8 @@ export async function jobCancelWithOptions( WHERE river_job.id = notification.id RETURNING river_job.* ) + `, + text: ` -- A cancel that lost a race to another committed after this statement -- began updates nothing. Like River for Go, the fallback read locks the -- row so it returns the committed row, not the one this statement first @@ -125,7 +128,8 @@ export async function jobCancelWithOptions( WHERE fallback_job.id NOT IN (SELECT id FROM updated_job) UNION SELECT * FROM updated_job - `; + `, + }; const result = await db.query( "jobCancel", @@ -150,7 +154,8 @@ export async function jobDelete( options?: PgOperationOptions ): Promise { const jobTable = db.table("river_job"); - const sql = ` + const sql = { + with: ` WITH job_to_delete AS ( SELECT id FROM ${jobTable} @@ -164,6 +169,8 @@ export async function jobDelete( AND river_job.state != 'running' RETURNING river_job.* ) + `, + text: ` SELECT *, false AS was_deleted FROM ${jobTable} WHERE id = $1::bigint @@ -171,7 +178,8 @@ export async function jobDelete( UNION SELECT *, true AS was_deleted FROM deleted_job - `; + `, + }; const result = await db.query( "jobDelete", sql, @@ -217,7 +225,8 @@ export async function jobDeleteMany( const jobTable = db.table("river_job"); const result = await db.query( "jobDeleteMany", - ` + { + with: ` WITH jobs_to_delete AS ( SELECT id FROM ${jobTable} WHERE state != 'running' @@ -236,8 +245,11 @@ export async function jobDeleteMany( WHERE river_job.id = jobs_to_delete.id RETURNING river_job.* ) + `, + text: ` SELECT * FROM deleted ORDER BY id ASC - `, + `, + }, [ params.ids.map((id) => id.toString(10)), params.kinds, @@ -381,7 +393,8 @@ export async function jobCompleteMany( const jobTable = db.table("river_job"); const result = await db.queryAbortable( "jobCompleteMany", - `${COMPLETE_MANY_STATEMENT} + { + with: `${COMPLETE_MANY_STATEMENT} WITH job_input AS ( SELECT * FROM unnest( @@ -467,6 +480,8 @@ export async function jobCompleteMany( ) RETURNING river_job.*, job_input.input_order ) + `, + text: ` SELECT updated.*, true AS transition_applied FROM updated UNION ALL @@ -483,7 +498,8 @@ export async function jobCompleteMany( WHERE metadata_updated.id = river_job.id ) ORDER BY input_order - `, + `, + }, [ commands.map(({ id }) => id.toString(10)), commands.map(({ attempt }) => attempt), @@ -659,7 +675,8 @@ export async function jobInsertMany( : null; const jobTable = db.table("river_job"); const stateInBitmask = db.function("river_job_state_in_bitmask"); - const sql = ` + const sql = { + with: ` WITH raw_job_data AS ( SELECT input_order, args, coalesce(created_at, now()) AS created_at, kind, @@ -713,6 +730,8 @@ export async function jobInsertMany( DO UPDATE SET kind = river_job.kind RETURNING *, ${uniqueInsertConflictSql(uniqueInsertMode)} AS conflicted ) + `, + text: ` SELECT inserted_jobs.*, inserted_jobs.conflicted AS unique_skipped_as_duplicate @@ -725,7 +744,8 @@ export async function jobInsertMany( ELSE inserted_jobs.id = prepared_job_data.proposed_id END ORDER BY prepared_job_data.input_order - `; + `, + }; const result = await db.query( "jobInsertMany", @@ -789,7 +809,8 @@ export async function jobRetryWithOptions( options?: PgOperationOptions ): Promise { const jobTable = db.table("river_job"); - const sql = ` + const sql = { + with: ` WITH job_to_update AS ( SELECT id FROM ${jobTable} @@ -816,6 +837,8 @@ export async function jobRetryWithOptions( ) RETURNING river_job.* ) + `, + text: ` -- Like a cancel's, the fallback read locks the row so that a retry that -- lost a race returns the committed row. SELECT * FROM ( @@ -828,7 +851,8 @@ export async function jobRetryWithOptions( UNION SELECT updated_job.* FROM updated_job - `; + `, + }; const result = await db.query( "jobRetry", sql, diff --git a/js/driver/pg/src/sql/leader.ts b/js/driver/pg/src/sql/leader.ts index e13c0dc6b..c5654dc5d 100644 --- a/js/driver/pg/src/sql/leader.ts +++ b/js/driver/pg/src/sql/leader.ts @@ -90,6 +90,7 @@ export async function leaderDeleteExpired( ` DELETE FROM ${db.table("river_leader")} WHERE expires_at < coalesce($1::timestamptz, now()) + RETURNING 1 `, [instantParameter(now)], options @@ -119,6 +120,7 @@ export async function leaderResign( ) END FROM held ) DELETE FROM ${db.table("river_leader")} USING notified + RETURNING 1 `, [ instantParameter(params.electedAt), diff --git a/js/driver/pg/src/sql/maintenance.ts b/js/driver/pg/src/sql/maintenance.ts index f01e101de..300a56e4a 100644 --- a/js/driver/pg/src/sql/maintenance.ts +++ b/js/driver/pg/src/sql/maintenance.ts @@ -67,6 +67,7 @@ export async function jobDeleteBefore( ORDER BY id ASC LIMIT $6::int ) + RETURNING 1 `, [ instantParameter(params.cancelledFinalizedAt), @@ -181,6 +182,7 @@ export async function jobRescueMany( WHERE river_job.id = rescued.id AND river_job.state = 'running' AND river_job.attempted_at < $6::timestamptz + RETURNING 1 `, [ params.items.map(({ id }) => id.toString(10)), @@ -213,7 +215,8 @@ export async function jobSchedule( const inBitmask = db.function("river_job_state_in_bitmask"); const result = await db.query( "jobSchedule", - ` + { + with: ` WITH jobs_to_schedule AS ( SELECT id, unique_key, unique_states, priority, scheduled_at FROM ${jobTable} @@ -269,10 +272,13 @@ export async function jobSchedule( WHERE river_job.id = job_updates.id RETURNING river_job.*, job_updates.conflict_discarded ) + `, + text: ` SELECT updated.* FROM updated ORDER BY priority ASC, scheduled_at ASC, id ASC - `, + `, + }, [ instantParameter(params.scheduledAtHorizon ?? params.now), instantParameter(params.now), @@ -506,7 +512,7 @@ export async function notificationDeleteBefore( WHERE created_at < $1::timestamptz ORDER BY created_at, id LIMIT $2::bigint - )`, + ) RETURNING 1`, [instantParameter(params.createdAtHorizon), params.max], options ); diff --git a/js/driver/pg/src/sql/queues.ts b/js/driver/pg/src/sql/queues.ts index 3d912e3ca..d0dde674f 100644 --- a/js/driver/pg/src/sql/queues.ts +++ b/js/driver/pg/src/sql/queues.ts @@ -175,7 +175,8 @@ async function queueSetPaused( PgQueueDatabaseRow | { [Key in keyof PgQueueDatabaseRow]: null } >( paused ? "queuePause" : "queueResume", - ` + { + with: ` WITH updated AS ( UPDATE ${db.table("river_queue")} SET @@ -205,11 +206,14 @@ async function queueSetPaused( ) END) AS sent WHERE $2::text = '*' OR EXISTS (SELECT 1 FROM updated) ) + `, + text: ` SELECT updated.* FROM notification LEFT JOIN updated ON true ORDER BY updated.name - `, + `, + }, [ instantParameter(params.now), params.name, @@ -239,7 +243,8 @@ export async function queueUpdate( const { supportsListenNotify } = await db.capabilities(options); const result = await db.query( "queueUpdate", - ` + { + with: ` WITH updated AS ( UPDATE ${db.table("river_queue")} SET @@ -254,8 +259,11 @@ export async function queueUpdate( $6::text ) END FROM updated WHERE $1::boolean ) + `, + text: ` SELECT updated.* FROM updated LEFT JOIN notification ON true - `, + `, + }, [ update !== undefined, update?.text ?? "{}", diff --git a/js/driver/pg/src/text-query.test.ts b/js/driver/pg/src/text-query.test.ts new file mode 100644 index 000000000..85a565f25 --- /dev/null +++ b/js/driver/pg/src/text-query.test.ts @@ -0,0 +1,39 @@ +import { Buffer } from "node:buffer"; +import { describe, expect, it } from "vitest"; +import { postgresTextQuery } from "./text-query.js"; + +describe("Postgres text-result decoding", () => { + it("keeps large IDs, JSON numbers, error text, and microsecond timestamps exact", () => { + const row = postgresTextQuery("SELECT 1").decode(`{ + "id": 9223372036854775807, + "args": {"id": 9223372036854775807}, + "created_at": "2026-10-17T12:34:56.123456+02:00", + "errors": [{"attempt": 1, "error": "prior error", "extra": 9223372036854775807}, null], + "finalized_at": null, + "unique_key": "\\\\x00ff", + "unique_states": "11111111" + }`) as Record; + expect(row.id).toBe(9_223_372_036_854_775_807n); + expect(JSON.stringify(row.args)).toBe('{"id":9223372036854775807}'); + expect(row.created_at).toEqual( + Temporal.Instant.from("2026-10-17T10:34:56.123456Z") + ); + expect(row.errors).toEqual([ + '{"attempt":1,"error":"prior error","extra":9223372036854775807}', + null, + ]); + expect(row.finalized_at).toBeNull(); + expect(row.unique_key).toEqual(Buffer.from([0, 255])); + expect(row.unique_states).toBe("11111111"); + }); + + it("preserves explicitly selected text IDs and escape-format bytes", () => { + const row = postgresTextQuery("SELECT 1").decode( + JSON.stringify({ id: "9223372036854775807", unique_key: "\\000\\377" }) + ); + expect(row).toEqual({ + id: "9223372036854775807", + unique_key: Buffer.from([0, 255]), + }); + }); +}); diff --git a/js/driver/pg/src/text-query.ts b/js/driver/pg/src/text-query.ts new file mode 100644 index 000000000..e6384781d --- /dev/null +++ b/js/driver/pg/src/text-query.ts @@ -0,0 +1,84 @@ +import { + isExactJsonNumber, + jsonNumberToBigInt, + parseJsonObject, +} from "riverqueue"; +import type { PostgresTransactionTextQuery } from "riverqueue/unstable-driver"; +import { PG_EXACT_TYPES } from "./exact-types.js"; + +/** + * A query and, when needed, its top-level WITH clause. Data-modifying CTEs + * must stay at the top level when an ORM wraps the result in a text projection. + * Keep that boundary explicit instead of parsing or rewriting SQL strings. + */ +export type PgStatement = + | string + | { + readonly text: string; + /** The entire WITH clause, including any leading statement comment. */ + readonly with: string; + }; + +/** Render the original statement for native Postgres transports. */ +export function postgresQueryText(statement: PgStatement): string { + return typeof statement === "string" + ? statement + : `${statement.with}\n${statement.text}`; +} + +/** + * Return each row as JSON text so an ORM cannot round JSON numbers or + * truncate timestamps. The same CTEs and final query execute once, with the + * original bindings, ordering, row locks, and transaction ownership. + */ +export function postgresTextQuery( + statement: PgStatement +): PostgresTransactionTextQuery { + const prefix = typeof statement === "string" ? "WITH" : `${statement.with},`; + const text = typeof statement === "string" ? statement : statement.text; + return { + decode: decodeTextRow, + text: `${prefix} river_transaction_result AS (${text}) + SELECT row_to_json(river_transaction_result)::text AS river_row + FROM river_transaction_result`, + }; +} + +/** Restore the Postgres types used by River's fixed query projections. */ +function decodeTextRow(text: string): Record { + const json = parseJsonObject(text); + const row: Record = { ...json }; + // An explicitly selected id::text stays text; ordinary bigint IDs are + // JSON numbers, read exactly before conversion to bigint. + if (typeof json.id === "number" || isExactJsonNumber(json.id)) { + row.id = jsonNumberToBigInt(json.id); + } + for (const field of [ + "attempted_at", + "created_at", + "elected_at", + "expires_at", + "finalized_at", + "paused_at", + "scheduled_at", + "updated_at", + ]) { + const value = json[field]; + if (typeof value === "string") row[field] = Temporal.Instant.from(value); + } + if (Array.isArray(json.errors)) { + // The normal pg decoder leaves jsonb[] elements as text for River's + // tolerant attempt-error decoder, including NULL elements. + row.errors = json.errors.map((value) => + value === null ? null : JSON.stringify(value) + ); + } + if (typeof json.unique_key === "string") { + // Honor both Postgres bytea_output formats with River's existing parser. + const parseBytea = PG_EXACT_TYPES.getTypeParser(17, "text") as ( + value: string + ) => unknown; + row.unique_key = parseBytea(json.unique_key); + } + return row; +} diff --git a/js/driver/pg/src/types.ts b/js/driver/pg/src/types.ts index bccf5b8f7..48070a8b2 100644 --- a/js/driver/pg/src/types.ts +++ b/js/driver/pg/src/types.ts @@ -11,9 +11,10 @@ export interface PgDriverOptions { * 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`. + * The ORM and the pg pool must connect to the same database. Configure the + * same River schema on the worker and insertion clients. Knex and Prisma + * remain optional dependencies, installed with `@riverqueue/driver-knex` + * and `@riverqueue/driver-prisma`, respectively. * * @example * ```ts @@ -45,6 +46,37 @@ export interface PgDriverOptions { * { queues: { default: { maxWorkers: 10 } }, workers } * ); * ``` + * + * @example + * ```ts + * import { Pool } from "pg"; + * import { Client, defineJob, Workers } from "riverqueue"; + * import { prismaTransactionAdapter } from "@riverqueue/driver-prisma"; + * import { PgDriver } from "@riverqueue/driver-pg"; + * import type { Prisma, PrismaClient } from "./generated/prisma/client.js"; + * + * declare const prisma: PrismaClient; // Your generated application client. + * const pool = new Pool({ connectionString: process.env.DATABASE_URL }); + * const syncAccount = defineJob<{ accountId: string }>()({ + * kind: "sync_account", + * }); + * const workers = new Workers().add( + * syncAccount, + * async ({ completeTx, job }) => { + * await prisma.$transaction(async (tx) => { + * await tx.account.update({ + * where: { id: job.args.accountId }, + * data: { synced: true }, + * }); + * await completeTx(tx); + * }); + * } + * ); + * const client = new Client( + * new PgDriver(pool, { transactionAdapters: [prismaTransactionAdapter] }), + * { queues: { default: { maxWorkers: 10 } }, workers } + * ); + * ``` */ transactionAdapters?: readonly PostgresTransactionAdapter[]; } diff --git a/js/driver/prisma/README.md b/js/driver/prisma/README.md index e62a75fb5..2cdabda5f 100644 --- a/js/driver/prisma/README.md +++ b/js/driver/prisma/README.md @@ -1,8 +1,8 @@ # `@riverqueue/driver-prisma` -This package inserts River jobs through Prisma, including inside an existing -Prisma transaction. It is intentionally an insertion adapter, not a worker -runtime. Run workers with `@riverqueue/driver-pg` or another complete backend. +This package inserts River jobs through Prisma and adapts native Prisma +transactions for `completeTx` in workers. Use `PrismaDriver` for insertion +and `@riverqueue/driver-pg` with `prismaTransactionAdapter` for workers. ```sh npm install riverqueue @riverqueue/driver-prisma @prisma/client @@ -11,11 +11,19 @@ npm install riverqueue @riverqueue/driver-prisma @prisma/client ```ts import { Client, defineJob } from "riverqueue"; import { z } from "zod"; -import { PrismaDriver, type PrismaClientLike } from "@riverqueue/driver-prisma"; +import { + PrismaDriver, + type PrismaClientLike, + type PrismaTransactionClientLike, +} from "@riverqueue/driver-prisma"; -interface ApplicationTransaction extends PrismaClientLike { +interface ApplicationTransaction extends PrismaTransactionClientLike { account: { create(options: { data: { id: string } }): Promise; + update(options: { + where: { id: string }; + data: { synced: boolean }; + }): Promise; }; } interface ApplicationPrismaClient extends PrismaClientLike { @@ -47,15 +55,14 @@ nor disconnects Prisma. Apply River migrations separately with state. Prisma's transaction object is structurally detected and bound to the operation -that receives it. A transaction cannot provide runtime features such as job -claiming, listening for notifications, leadership, or maintenance. +that receives it. The worker runtime uses a separate `pg` pool for job +claiming, notifications, leadership, and maintenance. Without `{ tx }`, River runs the insertion in an interactive transaction it begins with the root client's `$transaction`: argument validation, insert middleware, hooks, and the write commit together, and an error thrown by any of them rolls the jobs back. Construct the driver with your root `PrismaClient` for -this; a transaction client has no `$transaction`, so a driver built from one -requires `{ tx }` on every insertion. +this. A client without `$transaction` requires `{ tx }` on every insertion. Prisma bounds an interactive transaction with its defaults: a 2 second wait for a connection and a 5 second timeout, which also covers any I/O insert @@ -76,6 +83,58 @@ running workers claim it without waiting for their next poll. Inside a transacti transaction commits. Inserted rows are decoded exactly, so integers beyond JavaScript's safe range in arguments or metadata stay exact. +## Complete jobs inside Prisma transactions + +Workers run through `@riverqueue/driver-pg`. Add that package and `pg`, and +configure `prismaTransactionAdapter` to pass native Prisma interactive +transactions to `completeTx`. Building on the setup above: + +```ts continued +import { Pool } from "pg"; +import { Workers } from "riverqueue"; +import { prismaTransactionAdapter } from "@riverqueue/driver-prisma"; +import { PgDriver } from "@riverqueue/driver-pg"; + +const pool = new Pool({ connectionString: process.env.DATABASE_URL }); +const workers = new Workers().add( + syncAccount, + async ({ completeTx, job }) => { + await prisma.$transaction(async (tx) => { + await tx.account.update({ + where: { id: job.args.accountId }, + data: { synced: true }, + }); + await completeTx(tx); + }); + } +); +const workerClient = new Client( + new PgDriver(pool, { transactionAdapters: [prismaTransactionAdapter] }), + { queues: { default: { maxWorkers: 10 } }, workers } +); +const run = await workerClient.start(); +``` + +The application write and job completion use the same Prisma transaction and +commit together. If the transaction throws, both writes roll back and River +handles the worker's error normally. Await the transaction before returning +from the handler. Prisma owns transaction timeouts and any nested savepoints +supported by your Prisma version. + +Use the same database and River schema for Prisma and the worker's `PgDriver`. +The adapter also enables Prisma `{ tx }` options on the worker client's other +operations, including inserts from `ctx.client`. Pass the transaction client +received by `$transaction`, rather than the root Prisma client. Generated +`Prisma.TransactionClient` types can also be used as the `Workers` type argument. + +River preserves exact JSON numbers and microsecond timestamps by transporting +its query results as text, without changing Prisma's handling of application +queries. Importing the adapter doesn't enable it automatically; configure each +`PgDriver` that needs it. Neither `riverqueue` nor `@riverqueue/driver-pg` depends +on Prisma. + +At shutdown, stop `run`, then disconnect Prisma and close the `pg` pool. + ## Requirements Node.js 24 or newer. River automatically installs a global `Temporal` diff --git a/js/driver/prisma/etc/driver-prisma.api.md b/js/driver/prisma/etc/driver-prisma.api.md index 6bbf14daa..f06d6ae3d 100644 --- a/js/driver/prisma/etc/driver-prisma.api.md +++ b/js/driver/prisma/etc/driver-prisma.api.md @@ -13,9 +13,9 @@ declarations, and file layout are omitted. export interface PrismaClientLike { $queryRawUnsafe(query: string, ...values: unknown[]): Promise; /** - * Prisma's interactive transaction, present on a root client (not on the - * transaction client it passes to the callback). River uses it to run an - * insertion without `{ tx }` in a transaction of its own. + * Prisma's interactive transaction, present on a root client and on + * transaction clients in versions supporting nested transactions. River + * uses it to run an insertion without `{ tx }` in a transaction of its own. */ $transaction?( callback: (tx: PrismaClientLike) => Promise, @@ -70,6 +70,36 @@ export interface PrismaDriverOptions { } ``` +## `prismaTransactionAdapter` + +```ts +/** + * Let a PgDriver use native Prisma interactive transactions, including + * `completeTx(tx)` in worker handlers. Configure `new PgDriver(pool, { + * transactionAdapters: [prismaTransactionAdapter] })`. Prisma owns commit, + * rollback, and transaction timeouts; River uses the supplied transaction only. + */ +export declare const prismaTransactionAdapter: PostgresTransactionAdapter; +``` + +## `PrismaTransactionClientLike` + +```ts +/** + * The raw-query surface of Prisma's interactive-transaction client. It omits + * the root client's connection lifecycle methods. A generated + * `Prisma.TransactionClient` is assignable to this interface, including + * versions that expose `$transaction` for nested savepoints. + */ +export interface PrismaTransactionClientLike extends Pick< + PrismaClientLike, + "$queryRawUnsafe" +> { + readonly $connect?: never; + readonly $disconnect?: never; +} +``` + ## `PrismaTransactionOptions` ```ts diff --git a/js/driver/prisma/src/driver.ts b/js/driver/prisma/src/driver.ts index 0c037b498..d02bacd02 100644 --- a/js/driver/prisma/src/driver.ts +++ b/js/driver/prisma/src/driver.ts @@ -10,9 +10,9 @@ import { export interface PrismaClientLike { $queryRawUnsafe(query: string, ...values: unknown[]): Promise; /** - * Prisma's interactive transaction, present on a root client (not on the - * transaction client it passes to the callback). River uses it to run an - * insertion without `{ tx }` in a transaction of its own. + * Prisma's interactive transaction, present on a root client and on + * transaction clients in versions supporting nested transactions. River + * uses it to run an insertion without `{ tx }` in a transaction of its own. */ $transaction?( callback: (tx: PrismaClientLike) => Promise, diff --git a/js/driver/prisma/src/index.ts b/js/driver/prisma/src/index.ts index 434fe0132..383357630 100644 --- a/js/driver/prisma/src/index.ts +++ b/js/driver/prisma/src/index.ts @@ -1,4 +1,4 @@ -import type { PrismaClientLike } from "./driver.js"; +import type { PrismaTransactionClientLike } from "./transaction.js"; export { PrismaDriver } from "./driver.js"; export type { @@ -6,10 +6,12 @@ export type { PrismaDriverOptions, PrismaTransactionOptions, } from "./driver.js"; +export { prismaTransactionAdapter } from "./transaction.js"; +export type { PrismaTransactionClientLike } from "./transaction.js"; declare module "riverqueue" { interface RiverTransactionRegistry { /** Prisma interactive-transaction clients. */ - "@riverqueue/driver-prisma": PrismaClientLike; + "@riverqueue/driver-prisma": PrismaTransactionClientLike; } } diff --git a/js/driver/prisma/src/runtime.integration.test.ts b/js/driver/prisma/src/runtime.integration.test.ts new file mode 100644 index 000000000..48d74adcc --- /dev/null +++ b/js/driver/prisma/src/runtime.integration.test.ts @@ -0,0 +1,509 @@ +// Exercise native interactive transactions with a generated Prisma client, +// including model writes on the same connection as River's completion. +import { execFile } from "node:child_process"; +import { randomUUID } from "node:crypto"; +import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises"; +import { join } from "node:path"; +import { fileURLToPath } from "node:url"; +import { promisify } from "node:util"; +import { PrismaPg } from "@prisma/adapter-pg"; +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 { + prismaTransactionAdapter, + type PrismaClientLike, + type PrismaTransactionClientLike, +} from "./index.js"; + +const TEST_DATABASE_URL = + process.env.TEST_DATABASE_URL ?? + "postgres://localhost:5432/river_test?sslmode=disable"; +const packageDirectory = fileURLToPath(new URL("..", import.meta.url)); + +interface ApplicationRows { + count(): Promise; + create(options: { data: { value: string } }): Promise<{ value: string }>; + findMany(options?: { + orderBy: { value: "asc" }; + }): Promise<{ value: string }[]>; +} + +class CompanionClient extends PilotClient< + ClientBase | PrismaTransactionClientLike +> {} + +interface GeneratedPrismaClient extends PrismaClientLike { + applicationRow: ApplicationRows; + $disconnect(): Promise; + $transaction( + callback: (transaction: TransactionClient) => Promise + ): Promise; +} + +interface TransactionClient extends PrismaTransactionClientLike { + applicationRow: ApplicationRows; + $executeRawUnsafe(query: string, ...values: unknown[]): Promise; + $transaction( + callback: (transaction: TransactionClient) => Promise + ): Promise; +} + +describe("Prisma transactional completion", () => { + const schema = `prisma_work_${randomUUID().replaceAll("-", "")}`; + const definition = defineJob({ kind: "prisma_work" }); + let generatedDirectory: string; + let pool: Pool; + let prisma: GeneratedPrismaClient; + + beforeAll(async () => { + pool = new Pool({ connectionString: TEST_DATABASE_URL }); + await pool.query(`CREATE SCHEMA "${schema}"`); + await createMigrator({ pool, schema }).migrateUp(); + await pool.query( + `CREATE TABLE "${schema}".application_row (value text PRIMARY KEY)` + ); + const tempDirectory = join(packageDirectory, "temp"); + await mkdir(tempDirectory, { recursive: true }); + generatedDirectory = await mkdtemp(join(tempDirectory, "prisma-work-")); + const schemaFile = join(generatedDirectory, "schema.prisma"); + await writeFile( + schemaFile, + `generator client { + provider = "prisma-client" + output = "./client" + moduleFormat = "esm" + generatedFileExtension = "ts" + importFileExtension = "ts" +} + +datasource db { + provider = "postgresql" + schemas = ["${schema}"] +} + +model ApplicationRow { + value String @id + @@map("application_row") + @@schema("${schema}") +} +` + ); + await promisify(execFile)( + join(packageDirectory, "node_modules", ".bin", "prisma"), + ["generate", "--schema", schemaFile], + { cwd: packageDirectory } + ); + const generated = (await import( + join(generatedDirectory, "client", "client.ts") + )) as { + PrismaClient: new (options: { + adapter: PrismaPg; + }) => GeneratedPrismaClient; + }; + prisma = new generated.PrismaClient({ + adapter: new PrismaPg({ connectionString: TEST_DATABASE_URL }), + }); + }, 60_000); + + afterAll(async () => { + await prisma.$disconnect(); + await pool.query(`DROP SCHEMA "${schema}" CASCADE`); + await pool.end(); + await rm(generatedDirectory, { force: true, recursive: true }); + }); + + afterEach(async () => { + await pool.query( + `TRUNCATE "${schema}".application_row, "${schema}".river_job` + ); + }); + + const runWorker = async ( + handler: WorkHandler, + createPilot?: PilotFactory + ): Promise => { + const driver = new PgDriver(pool, { + schema, + transactionAdapters: [prismaTransactionAdapter], + }); + 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 } + ); + await pool.query( + `UPDATE "${schema}".river_job SET created_at = '2026-10-17T12:34:56.123456Z' WHERE id = $1`, + [inserted.job.id.toString()] + ); + 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" || + 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 model writes and exact completion values together", async () => { + let completed: JobRow | undefined; + const row = await runWorker(async ({ completeTx, job, setMetadata }) => { + setMetadata("custom", "value"); + await prisma.$transaction(async (tx) => { + await tx.applicationRow.create({ + data: { value: "committed ? $1 ' ☃" }, + }); + completed = await completeTx(tx, { + output: { id: exactJsonNumber("9223372036854775807") }, + }); + expect(completed.id).toBe(job.id); + expect(completed.createdAt.toString()).toBe( + "2026-10-17T12:34:56.123456Z" + ); + 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 prisma.applicationRow.count()).toBe(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 prisma.applicationRow.findMany()).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: [prismaTransactionAdapter], + }) + ); + let id: bigint | undefined; + await expect( + prisma.$transaction(async (tx) => { + const { job } = await client.insert( + definition, + { id: exactJsonNumber("9223372036854775807") }, + { tx } + ); + id = job.id; + const fetched = await client.jobs.get(id, { tx }); + expect(fetched).toEqual(job); + expect(await client.jobs.get(id)).toBeNull(); + throw new Error("rollback inserted job"); + }) + ).rejects.toThrow("rollback inserted job"); + expect(id).toBeDefined(); + expect(await client.jobs.get(id!)).toBeNull(); + }); + + it("keeps completion interceptors inside the Prisma transaction", async () => { + let expectedTx: TransactionClient | undefined; + let intercepted = false; + const row = await runWorker( + async ({ completeTx }) => { + await prisma.$transaction(async (tx) => { + expectedTx = tx; + await completeTx(tx); + expect(intercepted).toBe(true); + expect(await tx.applicationRow.findMany()).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.applicationRow.create({ + data: { value: "intercepted" }, + }); + const result = await next(); + expect(await prisma.applicationRow.count()).toBe(0); + return result; + }, + }, + }) + ); + expect(intercepted).toBe(true); + expect(row.state).toBe("discarded"); + expect(await prisma.applicationRow.count()).toBe(0); + }); + + it("keeps Postgres retryability for a blocked transactional write", async () => { + const client = new Client( + new PgDriver(pool, { + schema, + transactionAdapters: [prismaTransactionAdapter], + }) + ); + const { job } = await client.insert(definition, {}); + const holder = await pool.connect(); + try { + await holder.query("BEGIN"); + await holder.query( + `SELECT id FROM "${schema}".river_job WHERE id = $1 FOR UPDATE`, + [job.id.toString()] + ); + await expect( + prisma.$transaction(async (tx) => { + await tx.$executeRawUnsafe("SET LOCAL lock_timeout = '10ms'"); + await client.jobs.update( + job.id, + { metadata: { blocked: true } }, + { tx } + ); + }) + ).rejects.toMatchObject({ + code: "database", + operation: "jobUpdate", + retryable: true, + }); + } finally { + await holder.query("ROLLBACK"); + holder.release(); + } + expect((await client.jobs.get(job.id))?.metadata).not.toHaveProperty( + "blocked" + ); + }); + + it("preserves exact rows across transactional job and queue operations", async () => { + const client = new Client( + new PgDriver(pool, { + schema, + transactionAdapters: [prismaTransactionAdapter], + }) + ); + const exact = exactJsonNumber("9223372036854775807"); + await pool.query( + `INSERT INTO "${schema}".river_queue (name) VALUES ('controls')` + ); + await prisma.$transaction(async (tx) => { + const inserted = await client.insertMany( + [ + { + job: definition, + args: { id: exact }, + options: { unique: { byArgs: true } }, + }, + { job: definition, args: { id: "second" } }, + ], + { tx } + ); + expect(inserted.map(({ status }) => status)).toEqual([ + "inserted", + "inserted", + ]); + const first = inserted[0].job; + const second = inserted[1].job; + const duplicate = await client.insert( + definition, + { id: exact }, + { unique: { byArgs: true }, tx } + ); + expect(duplicate.status).toBe("duplicate"); + expect(duplicate.job).toEqual(first); + expect(first.uniqueKey).toBeInstanceOf(Uint8Array); + expect(first.uniqueStates).not.toBeNull(); + expect((await client.jobs.list({ tx })).jobs.map(({ id }) => id)).toEqual( + [first.id, second.id] + ); + expect( + (await client.jobs.update(first.id, { metadata: { exact } }, { tx })) + ?.metadata.exact + ).toEqual(exact); + expect((await client.jobs.cancel(first.id, { tx }))?.state).toBe( + "cancelled" + ); + expect((await client.jobs.retry(first.id, { tx }))?.state).toBe( + "available" + ); + expect((await client.jobs.delete(first.id, { tx }))?.id).toBe(first.id); + expect( + (await client.jobs.deleteMany({ ids: [second.id], tx })).map( + ({ id }) => id + ) + ).toEqual([second.id]); + expect(await client.jobs.get(first.id, { tx })).toBeNull(); + expect(await client.jobs.delete(first.id, { tx })).toBeNull(); + expect( + ( + await client.queues.update( + "controls", + { metadata: { exact } }, + { tx } + ) + )?.metadata.exact + ).toEqual(exact); + expect( + (await client.queues.get("controls", { tx }))?.createdAt + ).toBeInstanceOf(Temporal.Instant); + expect( + (await client.queues.pause("controls", { tx }))?.pausedAt + ).toBeInstanceOf(Temporal.Instant); + expect( + (await client.queues.resume("controls", { tx }))?.pausedAt + ).toBeNull(); + expect( + (await client.queues.list({ tx })).queues.map(({ name }) => name) + ).toContain("controls"); + expect(await client.queues.pause("missing", { tx })).toBeNull(); + }); + }); + + it("rejects closed, root, and unconfigured transaction clients", async () => { + const client = new Client( + new PgDriver(pool, { + schema, + transactionAdapters: [prismaTransactionAdapter], + }) + ); + // @ts-expect-error A root PrismaClient is not an interactive transaction. + await expect(client.jobs.get(1n, { tx: prisma })).rejects.toMatchObject({ + code: "backend_mismatch", + }); + const unconfigured = new Client(new PgDriver(pool, { schema })); + const tx = await prisma.$transaction(async (tx) => { + // @ts-expect-error Prisma transactions require a configured adapter. + await expect(unconfigured.jobs.get(1n, { tx })).rejects.toMatchObject({ + code: "backend_mismatch", + }); + return tx; + }); + let rejected = false; + const row = await runWorker(async ({ completeTx }) => { + await expect(completeTx(tx)).rejects.toMatchObject({ + cause: { code: "P2028" }, + }); + rejected = true; + throw new Error("transaction closed"); + }); + expect(rejected).toBe(true); + expect(row.state).toBe("discarded"); + }); + + it("rejects completion from an attempt whose identity changed", async () => { + let rejected = false; + const row = await runWorker(async ({ completeTx, job }) => { + await prisma.$transaction(async (tx) => { + await tx.$executeRawUnsafe( + `UPDATE "${schema}".river_job SET attempt = attempt + 1 WHERE id = $1::bigint`, + job.id.toString() + ); + await expect(completeTx(tx)).rejects.toThrow("no longer running"); + rejected = true; + throw new Error("rollback attempt change"); + }); + }); + expect(rejected).toBe(true); + expect(row.state).toBe("discarded"); + expect(row.attempt).toBe(1); + }); + + it("rolls back model writes and completion together", async () => { + let completed = false; + const row = await runWorker(async ({ completeTx }) => { + await prisma.$transaction(async (tx) => { + await tx.applicationRow.create({ data: { value: "rolled back" } }); + await completeTx(tx, { output: { rolledBack: true } }); + completed = true; + throw new Error("rollback both writes"); + }); + }); + expect(row.state).toBe("discarded"); + expect(completed).toBe(true); + expect(row.metadata).not.toHaveProperty("output"); + expect(await prisma.applicationRow.count()).toBe(0); + }); + + it("uses native savepoints without ending the outer transaction", async () => { + const row = await runWorker(async ({ completeTx }) => { + await prisma.$transaction(async (tx) => { + await tx.applicationRow.create({ data: { value: "outer" } }); + await expect( + tx.$transaction(async (nested) => { + await nested.applicationRow.create({ data: { value: "inner" } }); + await completeTx(nested, { output: { inner: true } }); + throw new Error("rollback savepoint"); + }) + ).rejects.toThrow("rollback savepoint"); + await completeTx(tx, { output: { outer: true } }); + await tx.applicationRow.create({ data: { value: "after completion" } }); + }); + }); + expect(row.state).toBe("completed"); + expect(row.metadata.output).toEqual({ outer: true }); + expect( + await prisma.applicationRow.findMany({ orderBy: { value: "asc" } }) + ).toEqual([{ value: "after completion" }, { value: "outer" }]); + }); +}); diff --git a/js/driver/prisma/src/transaction.test.ts b/js/driver/prisma/src/transaction.test.ts new file mode 100644 index 000000000..953d7bba7 --- /dev/null +++ b/js/driver/prisma/src/transaction.test.ts @@ -0,0 +1,143 @@ +import { Pool, type ClientBase } from "pg"; +import { Client } from "riverqueue"; +import { describe, expect, expectTypeOf, it, vi } from "vitest"; +import { PgDriver } from "../../pg/src/index.js"; +import { + prismaTransactionAdapter, + type PrismaTransactionClientLike, +} from "./index.js"; + +describe("prismaTransactionAdapter", () => { + it("adds Prisma transactions to the configured Postgres client's types", async () => { + const pool = new Pool(); + try { + const client = new Client( + new PgDriver(pool, { transactionAdapters: [prismaTransactionAdapter] }) + ); + expectTypeOf(client).toEqualTypeOf< + Client + >(); + const typed = ( + tx: PrismaTransactionClientLike, + root: Pick & { + $connect(): Promise; + } + ) => { + void client.jobs.get(1n, { tx }); + // @ts-expect-error A root PrismaClient is not a transaction. + void client.jobs.get(1n, { tx: root }); + }; + expect(typed).toBeTypeOf("function"); + } finally { + await pool.end(); + } + }); + + it("preserves Postgres error codes across Prisma error formats", async () => { + for (const meta of [ + { code: "55P03" }, + { driverAdapterError: { cause: { code: "55P03", kind: "postgres" } } }, + ]) { + const cause = Object.assign(new Error("lock timeout"), { meta }); + const $queryRawUnsafe = vi.fn().mockRejectedValue(cause); + await expect( + prismaTransactionAdapter.query( + { $queryRawUnsafe }, + { + text: "native SQL", + textResult: () => ({ decode: JSON.parse, text: "exact SQL" }), + values: [], + } + ) + ).rejects.toMatchObject({ + cause, + code: "55P03", + message: "lock timeout", + }); + } + }); + + it("preserves unrelated Prisma failures unchanged", async () => { + const cause = Object.assign(new Error("transaction expired"), { + code: "P2028", + meta: { error: "Transaction already closed" }, + }); + await expect( + prismaTransactionAdapter.query( + { $queryRawUnsafe: vi.fn().mockRejectedValue(cause) }, + { + text: "native SQL", + textResult: () => ({ decode: JSON.parse, text: "exact SQL" }), + values: [], + } + ) + ).rejects.toBe(cause); + }); + + it("recognizes transactions with and without native savepoint support", () => { + const transaction = { $queryRawUnsafe: vi.fn() }; + expect(prismaTransactionAdapter.matches(transaction)).toBe(true); + expect( + prismaTransactionAdapter.matches({ + ...transaction, + $transaction: vi.fn(), + }) + ).toBe(true); + // Prisma proxies can expose a masked property to `in` while returning + // undefined when the method is read. + expect( + prismaTransactionAdapter.matches({ + ...transaction, + $connect: undefined, + $disconnect: undefined, + }) + ).toBe(true); + for (const value of [ + null, + {}, + { ...transaction, $connect: vi.fn() }, + { ...transaction, $disconnect: vi.fn() }, + ]) { + expect(prismaTransactionAdapter.matches(value)).toBe(false); + expect(() => prismaTransactionAdapter.validate(value)).toThrow( + "interactive transaction" + ); + } + }); + + it("requires the exact text transport before issuing a query", async () => { + const $queryRawUnsafe = vi.fn(); + await expect( + prismaTransactionAdapter.query( + { $queryRawUnsafe }, + { text: "SELECT 1", values: [] } + ) + ).rejects.toThrow("exact text-result"); + expect($queryRawUnsafe).not.toHaveBeenCalled(); + }); + + it("uses the exact transport with bound values on the supplied transaction", async () => { + const $queryRawUnsafe = vi.fn(async () => [ + { river_row: "opaque exact row" }, + ]); + const decode = vi.fn((text: string) => ({ text })); + const values = ["? $1 ' ☃", [null, "9223372036854775807"]]; + const result = await prismaTransactionAdapter.query( + { $queryRawUnsafe }, + { + text: "native SQL", + textResult: () => ({ decode, text: "exact SQL $1 $2" }), + values, + } + ); + expect($queryRawUnsafe).toHaveBeenCalledExactlyOnceWith( + "exact SQL $1 $2", + ...values + ); + expect(decode).toHaveBeenCalledExactlyOnceWith("opaque exact row"); + expect(result).toEqual({ + rowCount: 1, + rows: [{ text: "opaque exact row" }], + }); + }); +}); diff --git a/js/driver/prisma/src/transaction.ts b/js/driver/prisma/src/transaction.ts new file mode 100644 index 000000000..35769f282 --- /dev/null +++ b/js/driver/prisma/src/transaction.ts @@ -0,0 +1,116 @@ +import { ConfigurationError } from "riverqueue"; +import type { + PostgresTransactionAdapter, + PostgresTransactionQuery, + PostgresTransactionResult, +} from "riverqueue/unstable-driver"; +import type { PrismaClientLike } from "./driver.js"; + +/** + * The raw-query surface of Prisma's interactive-transaction client. It omits + * the root client's connection lifecycle methods. A generated + * `Prisma.TransactionClient` is assignable to this interface, including + * versions that expose `$transaction` for nested savepoints. + */ +export interface PrismaTransactionClientLike extends Pick< + PrismaClientLike, + "$queryRawUnsafe" +> { + readonly $connect?: never; + readonly $disconnect?: never; +} + +/** + * Let a PgDriver use native Prisma interactive transactions, including + * `completeTx(tx)` in worker handlers. Configure `new PgDriver(pool, { + * transactionAdapters: [prismaTransactionAdapter] })`. Prisma owns commit, + * rollback, and transaction timeouts; River uses the supplied transaction only. + */ +export const prismaTransactionAdapter: PostgresTransactionAdapter = + Object.freeze({ + matches: isPrismaTransaction, + query: queryPrisma, + validate: validatePrismaTransaction, + }); + +function isPrismaTransaction( + value: unknown +): value is PrismaTransactionClientLike { + return ( + typeof value === "object" && + value !== null && + "$queryRawUnsafe" in value && + typeof value.$queryRawUnsafe === "function" && + (!("$connect" in value) || value.$connect === undefined) && + (!("$disconnect" in value) || value.$disconnect === undefined) + ); +} + +function postgresErrorCode(meta: unknown): string | undefined { + if (typeof meta !== "object" || meta === null) return undefined; + if ( + "code" in meta && + typeof meta.code === "string" && + /^[0-9A-Z]{5}$/.test(meta.code) + ) { + return meta.code; + } + if ( + "driverAdapterError" in meta && + typeof meta.driverAdapterError === "object" && + meta.driverAdapterError !== null && + "cause" in meta.driverAdapterError + ) { + return postgresErrorCode(meta.driverAdapterError.cause); + } + return undefined; +} + +async function queryPrisma( + transaction: unknown, + query: PostgresTransactionQuery +): Promise> { + validatePrismaTransaction(transaction); + if (query.textResult === undefined) { + throw new ConfigurationError( + "Prisma transactions require River's exact text-result query transport" + ); + } + const textQuery = query.textResult(); + // Prisma has no query-scoped type parsers: selecting rows as text prevents + // it from rounding JSON numbers or turning microsecond timestamps into Date. + let result: { river_row: string }[]; + try { + result = await transaction.$queryRawUnsafe<{ river_row: string }[]>( + textQuery.text, + ...query.values + ); + } catch (cause: unknown) { + // Prisma puts SQLSTATE in meta.code or, with newer driver adapters, + // meta.driverAdapterError.cause.code. Preserve it for PgDriver's + // retryability classification, retaining the original Prisma error. + if (cause instanceof Error && "meta" in cause) { + const code = postgresErrorCode(cause.meta); + if (code !== undefined) { + throw Object.assign(new Error(cause.message, { cause }), { code }); + } + } + throw cause; + } + return { + rowCount: result.length, + rows: result.map(({ river_row }) => textQuery.decode(river_row) as Row), + }; +} + +function validatePrismaTransaction( + value: unknown +): asserts value is PrismaTransactionClientLike { + if (!isPrismaTransaction(value)) { + throw new ConfigurationError( + "River requires a Prisma interactive transaction for { tx }" + ); + } + // Prisma itself rejects closed/expired interactive transactions before it + // sends a query. Do not inspect private Prisma engine or connection state. +} diff --git a/js/etc/riverqueue.api.md b/js/etc/riverqueue.api.md index 17ceceea5..20cdce6b5 100644 --- a/js/etc/riverqueue.api.md +++ b/js/etc/riverqueue.api.md @@ -2650,7 +2650,7 @@ interface RiverTransactionRegistry { interface RiverTransactionRegistry { /** Prisma interactive-transaction clients. */ - "@riverqueue/driver-prisma": PrismaClientLike; + "@riverqueue/driver-prisma": PrismaTransactionClientLike; } interface RiverTransactionRegistry { diff --git a/js/etc/riverqueue.unstable-driver.api.md b/js/etc/riverqueue.unstable-driver.api.md index 0f7678bcb..abb95301a 100644 --- a/js/etc/riverqueue.unstable-driver.api.md +++ b/js/etc/riverqueue.unstable-driver.api.md @@ -2042,12 +2042,12 @@ export declare function postgresTimestamp(value: Temporal.Instant): string; 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. */ + /** Run River's SQL on that exact transaction, preserving its exact values. */ query( transaction: unknown, query: PostgresTransactionQuery ): Promise>; - /** Reject completed transactions and unsupported database dialects. */ + /** Validate the handle; query execution must also reject closed transactions. */ validate(transaction: unknown): void; } ``` @@ -2058,7 +2058,13 @@ export interface PostgresTransactionAdapter { /** 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. */ + /** + * An equivalent query returning each row as JSON text, with a decoder back + * to the driver's row shape. Used by ORMs that cannot pass through `types`. + * Constructed lazily so native transports keep their existing SQL. + */ + readonly textResult?: () => PostgresTransactionTextQuery; + /** Driver-owned parsers; use these or the exact `textResult` transport. */ readonly types?: unknown; readonly values: readonly unknown[]; } @@ -2074,6 +2080,18 @@ export interface PostgresTransactionResult { } ``` +## `PostgresTransactionTextQuery` + +```ts +/** Exact row transport for ORMs that would otherwise coerce JSON and dates. */ +export interface PostgresTransactionTextQuery { + /** Decode one `river_row` string into the driver's ordinary row shape. */ + decode(row: string): unknown; + /** Returns one text column named `river_row` for each result row. */ + readonly text: string; +} +``` + ## `PreparedInsertParams` ```ts diff --git a/js/examples/prisma/README.md b/js/examples/prisma/README.md index 9d31fe165..39d6e5ea5 100644 --- a/js/examples/prisma/README.md +++ b/js/examples/prisma/README.md @@ -4,7 +4,9 @@ A minimal example demonstrating how to use the [River](https://github.com/riverq The example defines typed jobs and shows single insertion, scheduling, batch insertion, and a caller-owned Prisma interactive transaction that commits an -application row and its River job atomically. +application row and its River job atomically. It then runs a worker through +`PgDriver` with `prismaTransactionAdapter`, committing another application +write and `completeTx(tx)` together inside a native Prisma transaction. ## Prerequisites diff --git a/js/examples/prisma/package.json b/js/examples/prisma/package.json index af67caa8c..38c031324 100644 --- a/js/examples/prisma/package.json +++ b/js/examples/prisma/package.json @@ -15,12 +15,15 @@ "dependencies": { "@prisma/adapter-pg": "7.9.1", "@prisma/client": "7.9.1", + "@riverqueue/driver-pg": "workspace:*", "@riverqueue/driver-prisma": "workspace:*", + "pg": "^8.22.0", "riverqueue": "workspace:*", "zod": "^4.6.5" }, "devDependencies": { "@types/node": "^26.1.1", + "@types/pg": "^8.20.0", "prisma": "7.9.1", "typescript": "^6.0.3" } diff --git a/js/examples/prisma/src/index.ts b/js/examples/prisma/src/index.ts index 363062ca8..4139cdc2e 100644 --- a/js/examples/prisma/src/index.ts +++ b/js/examples/prisma/src/index.ts @@ -1,9 +1,15 @@ import { randomUUID } from "node:crypto"; +import assert from "node:assert/strict"; import { PrismaPg } from "@prisma/adapter-pg"; -import { PrismaDriver } from "@riverqueue/driver-prisma"; -import { Client, defineJob } from "riverqueue"; -import { PrismaClient } from "./generated/prisma/client.js"; +import { PgDriver } from "@riverqueue/driver-pg"; +import { + PrismaDriver, + prismaTransactionAdapter, +} from "@riverqueue/driver-prisma"; +import { Pool } from "pg"; +import { Client, defineJob, Workers } from "riverqueue"; +import { PrismaClient, type Prisma } from "./generated/prisma/client.js"; import { z } from "zod"; const sort = defineJob<{ strings: string[] }>()({ kind: "sort" }); @@ -71,15 +77,70 @@ async function main() { ) `; const accountId = `prisma_${randomUUID()}`; - await prisma.$transaction(async (tx) => { + const queue = `prisma_${randomUUID()}`; + const inserted = await prisma.$transaction(async (tx) => { await tx.$executeRaw` INSERT INTO riverqueue_example_account (id) VALUES (${accountId}) `; - await client.insert(syncAccount, { accountId }, { tx }); + return client.insert(syncAccount, { accountId }, { queue, tx }); }); console.log(`Committed account ${accountId} and its job atomically`); - await prisma.$disconnect(); + await prisma.$executeRaw` + CREATE TABLE IF NOT EXISTS riverqueue_example_synced_account ( + id text PRIMARY KEY + ) + `; + const pool = new Pool({ connectionString }); + const workers = new Workers().add( + syncAccount, + async ({ completeTx, job }) => { + await prisma.$transaction(async (tx) => { + await tx.$executeRaw` + INSERT INTO riverqueue_example_synced_account (id) + VALUES (${job.args.accountId}) + `; + await completeTx(tx); + }); + } + ); + const workerClient = new Client( + new PgDriver(pool, { transactionAdapters: [prismaTransactionAdapter] }), + { queues: { [queue]: { maxWorkers: 1 } }, workers } + ); + const events = workerClient.subscribe({ + kinds: ["job_completed", "job_failed"], + signal: AbortSignal.timeout(10_000), + }); + try { + const run = await workerClient.start(); + try { + for await (const event of events) { + if ( + event.kind === "subscription_lag" || + event.job.id !== inserted.job.id + ) + continue; + assert.equal(event.kind, "job_completed"); + break; + } + assert.equal( + (await workerClient.jobs.get(inserted.job.id))?.state, + "completed" + ); + const rows = await prisma.$queryRaw<{ id: string }[]>` + SELECT id FROM riverqueue_example_synced_account WHERE id = ${accountId} + `; + assert.deepEqual(rows, [{ id: accountId }]); + console.log(`Committed account sync and job completion atomically`); + } finally { + await run.stop(); + } + } finally { + events.close(); + await prisma.$disconnect(); + await pool.end(); + } } main().catch((error: unknown) => { diff --git a/js/pnpm-lock.yaml b/js/pnpm-lock.yaml index ef651a8f0..800267190 100644 --- a/js/pnpm-lock.yaml +++ b/js/pnpm-lock.yaml @@ -308,9 +308,15 @@ importers: '@prisma/client': specifier: 7.9.1 version: 7.9.1(prisma@7.9.1(@types/node@26.1.1)(@types/react@19.2.18)(magicast@0.5.5)(react-dom@19.2.8(react@19.2.8))(react@19.2.8)(typescript@6.0.3))(typescript@6.0.3) + '@riverqueue/driver-pg': + specifier: workspace:* + version: link:../../driver/pg '@riverqueue/driver-prisma': specifier: workspace:* version: link:../../driver/prisma + pg: + specifier: ^8.22.0 + version: 8.22.0 riverqueue: specifier: workspace:* version: link:../.. @@ -321,6 +327,9 @@ importers: '@types/node': specifier: ^26.1.1 version: 26.1.1 + '@types/pg': + specifier: ^8.20.0 + version: 8.20.0 prisma: specifier: 7.9.1 version: 7.9.1(@types/node@26.1.1)(@types/react@19.2.18)(magicast@0.5.5)(react-dom@19.2.8(react@19.2.8))(react@19.2.8)(typescript@6.0.3) diff --git a/js/scripts/check-packages.mjs b/js/scripts/check-packages.mjs index 0867fe1a3..8449ac200 100644 --- a/js/scripts/check-packages.mjs +++ b/js/scripts/check-packages.mjs @@ -571,7 +571,7 @@ 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 { PrismaDriver, prismaTransactionAdapter, type PrismaClientLike, type PrismaTransactionClientLike } from "@riverqueue/driver-prisma"; import { SqliteDriver, transaction, @@ -621,6 +621,25 @@ void knexRuntimeClient.jobs.get(1n, { tx: pgTx }); void knexRuntimeClient.jobs.get(1n, { tx: knex }); // @ts-expect-error Knex transactions require a configured adapter. void pgClient.jobs.get(1n, { tx: knexTx }); + +declare const prismaTx: PrismaTransactionClientLike; +declare const prismaRoot: PrismaClientLike & { $disconnect(): Promise }; +const prismaWorkers = new Workers().add(emailJob, async ({ completeTx, client }) => { + await completeTx(prismaTx); + await client.insert(emailJob, { address: "followup@example.com" }, { tx: prismaTx }); + // @ts-expect-error A root PrismaClient is not a transaction. + await completeTx(prismaRoot); +}); +const prismaRuntimeClient = new Client( + new PgDriver(pgPool, { transactionAdapters: [prismaTransactionAdapter] }), + { workers: prismaWorkers }, +); +void prismaRuntimeClient.jobs.get(1n, { tx: prismaTx }); +void prismaRuntimeClient.jobs.get(1n, { tx: pgTx }); +// @ts-expect-error A root PrismaClient is not a transaction. +void prismaRuntimeClient.jobs.get(1n, { tx: prismaRoot }); +// @ts-expect-error Prisma transactions require a configured adapter. +void pgClient.jobs.get(1n, { tx: prismaTx }); const sqliteDriver = new SqliteDriver(sqliteDatabase); const sqliteClient = new Client(sqliteDriver); const migrators: Migrator[] = [ diff --git a/js/src/internal/postgres-transaction.ts b/js/src/internal/postgres-transaction.ts index 78d64ac5c..a92220574 100644 --- a/js/src/internal/postgres-transaction.ts +++ b/js/src/internal/postgres-transaction.ts @@ -6,19 +6,25 @@ 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. */ + /** Run River's SQL on that exact transaction, preserving its exact values. */ query( transaction: unknown, query: PostgresTransactionQuery ): Promise>; - /** Reject completed transactions and unsupported database dialects. */ + /** Validate the handle; query execution must also reject closed transactions. */ 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. */ + /** + * An equivalent query returning each row as JSON text, with a decoder back + * to the driver's row shape. Used by ORMs that cannot pass through `types`. + * Constructed lazily so native transports keep their existing SQL. + */ + readonly textResult?: () => PostgresTransactionTextQuery; + /** Driver-owned parsers; use these or the exact `textResult` transport. */ readonly types?: unknown; readonly values: readonly unknown[]; } @@ -28,3 +34,11 @@ export interface PostgresTransactionResult { readonly rowCount: number | null; readonly rows: Row[]; } + +/** Exact row transport for ORMs that would otherwise coerce JSON and dates. */ +export interface PostgresTransactionTextQuery { + /** Decode one `river_row` string into the driver's ordinary row shape. */ + decode(row: string): unknown; + /** Returns one text column named `river_row` for each result row. */ + readonly text: string; +} diff --git a/js/src/unstable-driver.ts b/js/src/unstable-driver.ts index 566bb937a..39acc3144 100644 --- a/js/src/unstable-driver.ts +++ b/js/src/unstable-driver.ts @@ -178,4 +178,5 @@ export type { PostgresTransactionAdapter, PostgresTransactionQuery, PostgresTransactionResult, + PostgresTransactionTextQuery, } from "./internal/postgres-transaction.js";