JavaScript: Easy Prisma adapter for completeTx

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

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<Prisma.TransactionClient>()`. 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

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.
 

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<Transaction = never> {
   /**
    * 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<Transaction>
+  );
 }
 ```
 
 ## `PgDriverOptions`
 
-```ts
+````ts
 /** Construction options owned by the Postgres backend. */
-export interface PgDriverOptions {
+export interface PgDriverOptions<Transaction = never> {
   /** 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<Prisma.TransactionClient>().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<Transaction>[];
 }
-```
+````

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<ClientBase>
+    options?: PgOperationOptions | InsertDriverOptions
   ): Promise<PostgresCapabilities> {
     if (this.#capabilities !== undefined) return this.#capabilities;
     const result = await this.query<{
@@ -165,13 +177,15 @@ export class PgDatabase {
    */
   async query<Row extends QueryResultRow = QueryResultRow>(
     operation: string,
-    text: string,
+    text: PgStatement,
     values: unknown[],
-    options?: PgOperationOptions | InsertDriverOptions<ClientBase>
-  ): Promise<QueryResult<Row>> {
-    const queryable = this.resolveQueryable(operation, options);
+    options?: PgOperationOptions | InsertDriverOptions
+  ): Promise<PostgresTransactionResult<Row>> {
+    const queryable = this.resolveQueryable(operation, options, () =>
+      postgresTextQuery(text)
+    );
     const config: QueryConfig<unknown[]> = {
-      text,
+      text: postgresQueryText(text),
       types: PG_EXACT_TYPES,
       values,
     };
@@ -196,9 +210,9 @@ export class PgDatabase {
   async queryAfterAcquire<Row extends QueryResultRow = QueryResultRow>(
     signal: AbortSignal | undefined,
     operation: string,
-    text: string,
+    text: PgStatement,
     values: unknown[]
-  ): Promise<QueryResult<Row>> {
+  ): Promise<PostgresTransactionResult<Row>> {
     return this.withConnection(signal, (options) =>
       this.query<Row>(operation, text, values, options)
     );
@@ -212,7 +226,7 @@ export class PgDatabase {
    */
   async withConnection<T>(
     signal: AbortSignal | undefined,
-    run: (options: { readonly tx?: ClientBase }) => Promise<T>
+    run: (options: { readonly tx?: unknown }) => Promise<T>
   ): Promise<T> {
     if (signal === undefined || this.pool === null) return run({});
     const acquiring = this.pool.connect();
@@ -238,11 +252,11 @@ export class PgDatabase {
    */
   async queryAbortable<Row extends QueryResultRow = QueryResultRow>(
     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<QueryResult<Row>> {
+  ): Promise<PostgresTransactionResult<Row>> {
     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<ClientBase>
-  ): PgQueryable {
+    options?: PgOperationOptions | InsertDriverOptions,
+    textResult?: () => PostgresTransactionTextQuery
+  ): {
+    query<Row extends QueryResultRow>(
+      config: QueryConfig<unknown[]>
+    ): Promise<PostgresTransactionResult<Row>>;
+  } {
     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"