diff --git a/js/CHANGELOG.md b/js/CHANGELOG.md index 9d8e2d797..2ca8f71f6 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 +- 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 91c650d97..a573607bb 100644 --- a/js/docs/README.md +++ b/js/docs/README.md @@ -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 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 82a97ea5c..d5ba3bad3 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 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 a99311601..4738ace16 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. + * + * Prisma and the pg pool must connect to the same database. Configure the + * same River schema on the worker and insertion clients. Prisma remains an + * optional dependency, installed with `@riverqueue/driver-prisma`. + * + * @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 8d272772e..e6221500c 100644 --- a/js/driver/pg/src/database.ts +++ b/js/driver/pg/src/database.ts @@ -8,14 +8,15 @@ import { postgresCapabilitiesFromRow, quoteIdentifier, type PostgresCapabilities, + type PostgresTransactionAdapter, + type PostgresTransactionResult, + type PostgresTransactionTextQuery, } from "riverqueue/unstable-driver"; import type { Client as PgClient, - ClientBase, Pool, PoolClient, QueryConfig, - QueryResult, QueryResultRow, } from "pg"; import { RiverError } from "riverqueue"; @@ -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`). */ @@ -55,9 +61,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 +84,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<{ @@ -165,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); + options?: PgOperationOptions | InsertDriverOptions + ): Promise> { + const queryable = this.resolveQueryable(operation, options, () => + postgresTextQuery(text) + ); const config: QueryConfig = { - text, + text: postgresQueryText(text), types: PG_EXACT_TYPES, values, }; @@ -196,9 +210,9 @@ export class PgDatabase { async queryAfterAcquire( signal: AbortSignal | undefined, operation: string, - text: string, + text: PgStatement, values: unknown[] - ): Promise> { + ): Promise> { return this.withConnection(signal, (options) => this.query(operation, text, values, options) ); @@ -212,7 +226,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(); @@ -238,11 +252,11 @@ export class PgDatabase { */ async queryAbortable( operation: string, - text: string, + text: PgStatement, 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 +373,32 @@ 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, + textResult?: () => PostgresTransactionTextQuery + ): { + 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, + ...(textResult === undefined ? {} : { textResult }), + 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 +423,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..37b91ecb6 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 { @@ -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, @@ -296,7 +308,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 +323,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 +385,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); @@ -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), @@ -629,7 +645,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 +662,7 @@ export async function jobInsert( export async function jobInsertMany( db: PgDatabase, params: readonly JobInsertParams[], - options?: InsertDriverOptions + options?: InsertDriverOptions ): Promise { if (params.length === 0) return []; @@ -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, @@ -861,7 +885,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..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), @@ -133,7 +135,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..300a56e4a 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, @@ -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), @@ -391,9 +397,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 +455,7 @@ export function maintenanceTransaction( db: PgDatabase, leader: RuntimeLeader, batch: RuntimeMaintenanceBatch, - callback: (tx: ClientBase) => Promise + callback: (tx: PoolClient) => Promise ): Promise { return withMaintenanceLeader( db, @@ -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 ); @@ -684,7 +690,7 @@ async function holdLeaderFence( db: PgDatabase, operation: string, leader: RuntimeLeader, - tx: ClientBase + tx: unknown ): Promise { const held = await db.query( `${operation}LeaderFence`, @@ -711,8 +717,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/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/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/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 e3673d45b..2ebb563e5 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. + * + * Prisma and the pg pool must connect to the same database. Configure the + * same River schema on the worker and insertion clients. Prisma remains an + * optional dependency, installed with `@riverqueue/driver-prisma`. + * + * @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[]; } /** 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/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 c7785f599..db7a49dd3 100644 --- a/js/driver/prisma/src/driver.ts +++ b/js/driver/prisma/src/driver.ts @@ -31,9 +31,9 @@ const RIVER_SCHEMA_RE = /^[A-Za-z_][A-Za-z0-9_]*$/; 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 8418e3a83..c27b60f9d 100644 --- a/js/etc/riverqueue.api.md +++ b/js/etc/riverqueue.api.md @@ -2645,7 +2645,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 272ff2394..a96db5184 100644 --- a/js/etc/riverqueue.unstable-driver.api.md +++ b/js/etc/riverqueue.unstable-driver.api.md @@ -1953,6 +1953,67 @@ 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, preserving its exact values. */ + query( + transaction: unknown, + query: PostgresTransactionQuery + ): Promise>; + /** Validate the handle; query execution must also reject closed transactions. */ + validate(transaction: unknown): void; +} +``` + +## `PostgresTransactionQuery` + +```ts +/** River's numbered-parameter SQL and query-scoped Postgres type parsers. */ +export interface PostgresTransactionQuery { + readonly text: string; + /** + * 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[]; +} +``` + +## `PostgresTransactionResult` + +```ts +/** Query results used by River, independent of an ORM or pg package. */ +export interface PostgresTransactionResult { + readonly rowCount: number | null; + readonly rows: Row[]; +} +``` + +## `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 742e4fed1..6c197364f 100644 --- a/js/pnpm-lock.yaml +++ b/js/pnpm-lock.yaml @@ -290,9 +290,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:../.. @@ -303,6 +309,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 d97cbf098..28539e257 100644 --- a/js/scripts/check-packages.mjs +++ b/js/scripts/check-packages.mjs @@ -566,7 +566,7 @@ import type { } from "riverqueue"; import { run as runCli } from "@riverqueue/cli"; 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, @@ -592,6 +592,24 @@ declare const sqliteDatabase: DatabaseSync; const client = new Client(driver); const pgDriver = new PgDriver(pgPool); const pgClient = new Client(pgDriver); +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 new file mode 100644 index 000000000..a92220574 --- /dev/null +++ b/js/src/internal/postgres-transaction.ts @@ -0,0 +1,44 @@ +/** + * 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, preserving its exact values. */ + query( + transaction: unknown, + query: PostgresTransactionQuery + ): Promise>; + /** 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; + /** + * 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[]; +} + +/** Query results used by River, independent of an ORM or pg package. */ +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 2ff369189..b1ff7dd89 100644 --- a/js/src/unstable-driver.ts +++ b/js/src/unstable-driver.ts @@ -171,3 +171,10 @@ export type { ReadonlyJsonObject, ReadonlyJsonValue, } from "./job-args-transform.js"; + +export type { + PostgresTransactionAdapter, + PostgresTransactionQuery, + PostgresTransactionResult, + PostgresTransactionTextQuery, +} from "./internal/postgres-transaction.js";