JavaScript: Easy Prisma adapter for completeTx

js/CHANGELOG.md

@@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Added
 
 - Add `@riverqueue/driver-knex` for inserting jobs on Postgres inside Knex transactions, including batches, unique jobs, and nested savepoints. Configure its `knexTransactionAdapter` on `PgDriver` to use native Knex transactions with `completeTx` in workers. [PR #1516](https://github.com/riverqueue/river/pull/1516).
+- Support native Prisma interactive transactions with `completeTx` in workers and `{ tx }` on Postgres client operations. Configure `prismaTransactionAdapter` from `@riverqueue/driver-prisma` on `PgDriver` to commit application writes and job completion together. [PR #1517](https://github.com/riverqueue/river/pull/1517).
 - Support Node.js 24 across all packages. River automatically installs a `Temporal` polyfill when the runtime has no native implementation, preserving native `Temporal` on Node.js 26.
 
 ### Changed

js/docs/README.md

@@ -366,6 +366,11 @@ worker's `PgDriver` and use `new Workers<Knex.Transaction>()`. See the
 [Knex example](../driver/knex/README.md#complete-jobs-inside-knex-transactions).
 Installing the Knex driver alone does not enable its transactions on `PgDriver`.
 
+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

@@ -64,6 +64,13 @@ importing the adapter from `@riverqueue/driver-knex`. This enables both
 [Knex transaction example](../knex/README.md#complete-jobs-inside-knex-transactions).
 The adapter is optional; this package has no Knex dependency.
 
+To use native Prisma transactions, configure
+`new PgDriver(pool, { transactionAdapters: [prismaTransactionAdapter] })`,
+importing the adapter from `@riverqueue/driver-prisma`. This enables both
+`completeTx(tx)` in workers and `{ tx }` on client operations. See the
+[Prisma transaction example](../prisma/README.md#complete-jobs-inside-prisma-transactions).
+The adapter is optional; this package has no Prisma dependency.
+
 Apply migrations explicitly with `@riverqueue/migrate` or
 `@riverqueue/cli` before starting workers.
 

js/driver/pg/etc/driver-pg.api.md

@@ -51,9 +51,10 @@ export interface PgDriverOptions<Transaction = never> {
    * and configure it on each `PgDriver` that needs it. The ORM owns commit,
    * rollback, and savepoints; River uses the supplied transaction only.
    *
-   * Knex and the pg pool must connect to the same database. Configure the
-   * same River schema on the worker and insertion clients. Knex remains an
-   * optional dependency, installed with `@riverqueue/driver-knex`.
+   * The ORM and the pg pool must connect to the same database. Configure the
+   * same River schema on the worker and insertion clients. Knex and Prisma
+   * remain optional dependencies, installed with `@riverqueue/driver-knex`
+   * and `@riverqueue/driver-prisma`, respectively.
    *
    * @example
    * ```ts
@@ -85,6 +86,37 @@ export interface PgDriverOptions<Transaction = never> {
    *   { queues: { default: { maxWorkers: 10 } }, workers }
    * );
    * ```
+   *
+   * @example
+   * ```ts
+   * import { Pool } from "pg";
+   * import { Client, defineJob, Workers } from "riverqueue";
+   * import { prismaTransactionAdapter } from "@riverqueue/driver-prisma";
+   * import { PgDriver } from "@riverqueue/driver-pg";
+   * import type { Prisma, PrismaClient } from "./generated/prisma/client.js";
+   *
+   * declare const prisma: PrismaClient; // Your generated application client.
+   * const pool = new Pool({ connectionString: process.env.DATABASE_URL });
+   * const syncAccount = defineJob<{ accountId: string }>()({
+   *   kind: "sync_account",
+   * });
+   * const workers = new Workers<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

@@ -10,6 +10,7 @@ import {
   type PostgresCapabilities,
   type PostgresTransactionAdapter,
   type PostgresTransactionResult,
+  type PostgresTransactionTextQuery,
 } from "riverqueue/unstable-driver";
 import type {
   Client as PgClient,
@@ -27,6 +28,11 @@ import {
 } from "./errors.js";
 import { PG_EXACT_TYPES } from "./exact-types.js";
 import { abortablePromise, PgClientLease } from "./lease.js";
+import {
+  postgresQueryText,
+  postgresTextQuery,
+  type PgStatement,
+} from "./text-query.js";
 import type { PgOperationOptions } from "./types.js";
 
 /** Postgres's maximum identifier length (`NAMEDATALEN - 1`). */
@@ -171,13 +177,15 @@ export class PgDatabase {
    */
   async query<Row extends QueryResultRow = QueryResultRow>(
     operation: string,
-    text: string,
+    text: PgStatement,
     values: unknown[],
     options?: PgOperationOptions | InsertDriverOptions
   ): Promise<PostgresTransactionResult<Row>> {
-    const queryable = this.resolveQueryable(operation, options);
+    const queryable = this.resolveQueryable(operation, options, () =>
+      postgresTextQuery(text)
+    );
     const config: QueryConfig<unknown[]> = {
-      text,
+      text: postgresQueryText(text),
       types: PG_EXACT_TYPES,
       values,
     };
@@ -202,7 +210,7 @@ export class PgDatabase {
   async queryAfterAcquire<Row extends QueryResultRow = QueryResultRow>(
     signal: AbortSignal | undefined,
     operation: string,
-    text: string,
+    text: PgStatement,
     values: unknown[]
   ): Promise<PostgresTransactionResult<Row>> {
     return this.withConnection(signal, (options) =>
@@ -244,7 +252,7 @@ export class PgDatabase {
    */
   async queryAbortable<Row extends QueryResultRow = QueryResultRow>(
     operation: string,
-    text: string,
+    text: PgStatement,
     values: unknown[],
     options: { readonly signal?: AbortSignal; readonly tx?: unknown } = {},
     cancelPrefix?: string
@@ -365,7 +373,8 @@ export class PgDatabase {
   /** The caller's transaction client when given, else the configured pool or client. */
   resolveQueryable(
     operation: string,
-    options?: PgOperationOptions | InsertDriverOptions
+    options?: PgOperationOptions | InsertDriverOptions,
+    textResult?: () => PostgresTransactionTextQuery
   ): {
     query<Row extends QueryResultRow>(
       config: QueryConfig<unknown[]>
@@ -380,6 +389,7 @@ export class PgDatabase {
         query: (config) =>
           adapter.query(tx, {
             text: config.text,
+            ...(textResult === undefined ? {} : { textResult }),
             types: config.types,
             values: config.values ?? [],
           }),

js/driver/pg/src/sql/jobs.ts

@@ -69,7 +69,8 @@ export async function jobCancelWithOptions(
 ): Promise<JobRow | null> {
   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<PgJobRow>(
     "jobCancel",
@@ -150,7 +154,8 @@ export async function jobDelete(
   options?: PgOperationOptions
 ): Promise<JobDeleteResult> {
   const jobTable = db.table("river_job");
-  const sql = `
+  const sql = {
+    with: `
     WITH job_to_delete AS (
       SELECT id
       FROM ${jobTable}
@@ -164,14 +169,17 @@ export async function jobDelete(
         AND river_job.state != 'running'
       RETURNING river_job.*
     )
+    `,
+    text: `
     SELECT *, false AS was_deleted
     FROM ${jobTable}
     WHERE id = $1::bigint
       AND id NOT IN (SELECT id FROM deleted_job)
     UNION
     SELECT *, true AS was_deleted
     FROM deleted_job
-  `;
+    `,
+  };
   const result = await db.query<PgJobRow & { was_deleted: boolean }>(
     "jobDelete",
     sql,
@@ -217,7 +225,8 @@ export async function jobDeleteMany(
   const jobTable = db.table("river_job");
   const result = await db.query<PgJobRow>(
     "jobDeleteMany",
-    `
+    {
+      with: `
       WITH jobs_to_delete AS (
         SELECT id FROM ${jobTable}
         WHERE state != 'running'
@@ -236,8 +245,11 @@ export async function jobDeleteMany(
         WHERE river_job.id = jobs_to_delete.id
         RETURNING river_job.*
       )
+      `,
+      text: `
       SELECT * FROM deleted ORDER BY id ASC
-    `,
+      `,
+    },
     [
       params.ids.map((id) => id.toString(10)),
       params.kinds,
@@ -381,7 +393,8 @@ export async function jobCompleteMany(
   const jobTable = db.table("river_job");
   const result = await db.queryAbortable<PgCompletionRow>(
     "jobCompleteMany",
-    `${COMPLETE_MANY_STATEMENT}
+    {
+      with: `${COMPLETE_MANY_STATEMENT}
       WITH job_input AS (
         SELECT *
         FROM unnest(
@@ -467,6 +480,8 @@ export async function jobCompleteMany(
           )
         RETURNING river_job.*, job_input.input_order
       )
+      `,
+      text: `
       SELECT updated.*, true AS transition_applied
       FROM updated
       UNION ALL
@@ -483,7 +498,8 @@ export async function jobCompleteMany(
           WHERE metadata_updated.id = river_job.id
         )
       ORDER BY input_order
-    `,
+      `,
+    },
     [
       commands.map(({ id }) => id.toString(10)),
       commands.map(({ attempt }) => attempt),
@@ -659,7 +675,8 @@ export async function jobInsertMany(
       : null;
   const jobTable = db.table("river_job");
   const stateInBitmask = db.function("river_job_state_in_bitmask");
-  const sql = `
+  const sql = {
+    with: `
     WITH raw_job_data AS (
       SELECT
         input_order, args, coalesce(created_at, now()) AS created_at, kind,
@@ -713,6 +730,8 @@ export async function jobInsertMany(
       DO UPDATE SET kind = river_job.kind
       RETURNING *, ${uniqueInsertConflictSql(uniqueInsertMode)} AS conflicted
     )
+    `,
+    text: `
     SELECT
       inserted_jobs.*,
       inserted_jobs.conflicted AS unique_skipped_as_duplicate
@@ -725,7 +744,8 @@ export async function jobInsertMany(
       ELSE inserted_jobs.id = prepared_job_data.proposed_id
     END
     ORDER BY prepared_job_data.input_order
-  `;
+    `,
+  };
 
   const result = await db.query<PgInsertRow>(
     "jobInsertMany",
@@ -789,7 +809,8 @@ export async function jobRetryWithOptions(
   options?: PgOperationOptions
 ): Promise<JobRow | null> {
   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<PgJobRow>(
     "jobRetry",
     sql,

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),

js/driver/pg/src/sql/maintenance.ts

@@ -67,6 +67,7 @@ export async function jobDeleteBefore(
         ORDER BY id ASC
         LIMIT $6::int
       )
+      RETURNING 1
     `,
     [
       instantParameter(params.cancelledFinalizedAt),
@@ -181,6 +182,7 @@ export async function jobRescueMany(
       WHERE river_job.id = rescued.id
         AND river_job.state = 'running'
         AND river_job.attempted_at < $6::timestamptz
+      RETURNING 1
     `,
     [
       params.items.map(({ id }) => id.toString(10)),
@@ -213,7 +215,8 @@ export async function jobSchedule(
   const inBitmask = db.function("river_job_state_in_bitmask");
   const result = await db.query<PgScheduleRow>(
     "jobSchedule",
-    `
+    {
+      with: `
       WITH jobs_to_schedule AS (
         SELECT id, unique_key, unique_states, priority, scheduled_at
         FROM ${jobTable}
@@ -269,10 +272,13 @@ export async function jobSchedule(
         WHERE river_job.id = job_updates.id
         RETURNING river_job.*, job_updates.conflict_discarded
       )
+      `,
+      text: `
       SELECT updated.*
       FROM updated
       ORDER BY priority ASC, scheduled_at ASC, id ASC
-    `,
+      `,
+    },
     [
       instantParameter(params.scheduledAtHorizon ?? params.now),
       instantParameter(params.now),
@@ -506,7 +512,7 @@ export async function notificationDeleteBefore(
        WHERE created_at < $1::timestamptz
        ORDER BY created_at, id
        LIMIT $2::bigint
-     )`,
+     ) RETURNING 1`,
     [instantParameter(params.createdAtHorizon), params.max],
     options
   );