fix(db): honor the prepared-statement opt-in for ORM queries (#17673)

Drizzle calls Postgres.js unsafe queries, which disable preparation unless
per-query options enable it. The existing client-level opt-in therefore
does not prepare ORM queries.

Honor explicit true across root and child clients. Preserve defaults,
per-query options, lazy execution, and driver error handling. Document the
pooler capability requirement.

Test with 92 selected database and connection checks, workspace typechecks,
server and database builds, and a read-only hosted ORM protocol probe.

Co-Authored-By: Paperclip <noreply@paperclip.ing>

doc/DATABASE.md

@@ -458,14 +458,14 @@ DATABASE_URL=postgres://postgres.[PROJECT-REF]:[PASSWORD]@aws-0-[REGION].pooler.
 DATABASE_MIGRATION_URL=postgres://postgres.[PROJECT-REF]:[PASSWORD]@aws-0-[REGION].pooler.supabase.com:5432/postgres
 ```
 
-If your hosted database requires transaction-pooling-only connections (pgbouncer transaction mode, Supavisor port 6543, Neon `-pooler` endpoints), set `DATABASE_PREPARED_STATEMENTS=false` so the client does not rely on session-scoped prepared statements, and keep `DATABASE_MIGRATION_URL` on a direct connection. Do not edit database client source files as part of deployment setup.
+If your transaction-mode pooler does not support protocol-level named prepared statements, set `DATABASE_PREPARED_STATEMENTS=false`. If your pooler supports them, you can opt in with `DATABASE_PREPARED_STATEMENTS=true`. Check your provider's pooler configuration before you enable this setting. Keep `DATABASE_MIGRATION_URL` on a direct connection. Do not edit database client source files as part of deployment setup.
 
 ### Client tuning (optional)
 
 All of these are optional; when unset, the driver defaults apply and behavior is unchanged — typical self-hosted setups need none of them:
 
 ```sh
-DATABASE_PREPARED_STATEMENTS=false   # required for transaction-mode poolers; default: enabled
+DATABASE_PREPARED_STATEMENTS=false   # use false when the pooler does not support named prepared statements
 DATABASE_POOL_MAX=25                 # connection pool size; default: 10
 DATABASE_IDLE_TIMEOUT_SECONDS=60     # close idle pooled connections; default: 60 (0 = keep open)
 DATABASE_CONNECT_TIMEOUT_SECONDS=10  # default: 30
@@ -473,6 +473,14 @@ DATABASE_MAX_LIFETIME_SECONDS=1800   # recycle a pooled connection after this lo
 DATABASE_APPLICATION_NAME=paperclip  # application_name in pg_stat_activity; default: paperclip
 ```
 
+An explicit `DATABASE_PREPARED_STATEMENTS=true` enables named prepared statements
+for parameterized ORM queries, transactions, reserved connections, and dedicated
+clients. Postgres.js manages the statement cache. This does not cache query
+results. Bound values and returned data stay fresh. Explicit per-query options
+still apply, including `prepare: false` and simple-protocol execution. With the
+setting unset, ORM queries keep their existing unprepared behavior and tagged
+SQL uses the driver default. With `false`, preparation stays disabled.
+
 ### Bounded response diagnostics
 
 `PAPERCLIP_DB_RESPONSE_CAPTURE_WINDOW=<start>/<end>` enables one explicit UTC

docs/deploy/database.md

@@ -56,12 +56,19 @@ For production, use a hosted provider like [Supabase](https://supabase.com/).
 
 Use the **direct connection** (port 5432) for migrations and the **pooled connection** (port 6543) for the application.
 
-If using connection pooling (transaction mode), disable prepared statements via the environment — no source edits needed:
+If your transaction-mode pooler does not support protocol-level named prepared statements, disable them via the environment:
 
 ```sh
 DATABASE_PREPARED_STATEMENTS=false
 ```
 
+For a direct connection or a pooler that supports named prepared statements,
+`DATABASE_PREPARED_STATEMENTS=true` also prepares parameterized ORM queries and
+queries inside transactions and reserved or dedicated connections. Check the
+provider's configuration before you enable it. Preparation caches statement
+metadata, not query results. Unset preserves the existing ORM behavior and the
+driver default for tagged SQL. Explicit per-query options still apply.
+
 Related optional client tuning: `DATABASE_POOL_MAX`, `DATABASE_IDLE_TIMEOUT_SECONDS`, `DATABASE_CONNECT_TIMEOUT_SECONDS`, `DATABASE_MAX_LIFETIME_SECONDS`, `DATABASE_APPLICATION_NAME`. Driver defaults apply when unset, except that idle pooled connections close after 60 seconds (`DATABASE_IDLE_TIMEOUT_SECONDS=0` keeps them open) and the pool reports `application_name=paperclip`. See [Connection pool settings](#connection-pool-settings).
 
 ## Connection Pool Settings
@@ -75,7 +82,7 @@ The server opens one postgres.js pool for its own queries (and a second one when
 | `DATABASE_CONNECT_TIMEOUT_SECONDS` | `30` (driver) | Give up on a connection attempt after this long. |
 | `DATABASE_MAX_LIFETIME_SECONDS` | 30–60 min, randomized (driver) | Recycle a pooled connection once it is this old. |
 | `DATABASE_APPLICATION_NAME` | `paperclip` | Value of `application_name` in `pg_stat_activity`, so you can find Paperclip's backends: `SELECT * FROM pg_stat_activity WHERE application_name = 'paperclip';` |
-| `DATABASE_PREPARED_STATEMENTS` | `true` (driver) | Set `false` behind a transaction-mode pooler (see above). |
+| `DATABASE_PREPARED_STATEMENTS` | Unset | `true` opts parameterized ORM queries into preparation. `false` disables preparation. Use `false` when the pooler does not support named statements. Unset preserves existing behavior. |
 
 The server ends its pools during shutdown (SIGINT/SIGTERM) and when startup fails after the pool was opened, so a restarting server does not leave idle backends behind. Size `max_connections` on the PostgreSQL side for at least `DATABASE_POOL_MAX` per server process plus your other clients. Each server process also opens up to 2 connections (`application_name=paperclip-controller-lease`) for agent-run lease renewals, so they never queue behind slow application queries; count those too.
 

packages/db/src/client.ts

@@ -5,6 +5,7 @@ import { fileURLToPath } from "node:url";
 import postgres from "postgres";
 import { installDatabaseWorkSignals } from "./work-signals.js";
 import { installDatabaseResponseCapture } from "./response-capture.js";
+import { installPreparedQueries } from "./prepared-queries.js";
 import * as schema from "./schema/index.js";
 
 const MIGRATIONS_FOLDER = fileURLToPath(new URL("./migrations", import.meta.url));
@@ -129,13 +130,6 @@ export type MigrationState =
     };
 
 export interface DatabaseClientOptions {
-  /**
-   * postgres.js `prepare`. Set false when connecting through a
-   * transaction-mode pooler (pgbouncer / Neon `-pooler` endpoints /
-   * Supabase Supavisor transaction ports) so the client does not rely on
-   * session-scoped prepared statements. Defaults to the driver default
-   * (enabled), preserving existing behavior on direct connections.
-   */
   prepare?: boolean;
   /** postgres.js `max` — connection pool size (driver default: 10). */
   maxConnections?: number;
@@ -301,6 +295,7 @@ export function createDb(url: string, options?: DatabaseClientOptions) {
 
 function createDbClient(url: string, resolved: DatabaseClientOptions, owner?: object) {
   const sql = postgres(url, postgresJsOptions(resolved));
+  if (resolved.prepare === true) installPreparedQueries(sql);
   const key = hostPortKeyOrNull(url);
   if (key) registerClient(key, sql);
   // A disconnect can lose the response after a statement has committed.

packages/db/src/prepared-queries-postgres.test.ts

@@ -0,0 +1,175 @@
+import { eq, sql } from "drizzle-orm";
+import { integer, jsonb, pgTable, text } from "drizzle-orm/pg-core";
+import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it } from "vitest";
+import { closeLivenessDb, createDb, databaseClientOptionsFromEnv, livenessDb, withDedicatedDbConnection, type DatabaseClientOptions, type Db } from "./client.js";
+import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./test-embedded-postgres.js";
+import { runEventPayload } from "./run-event-payload.js";
+
+const records = pgTable("prepared_query_fixture", { id: text("id").primaryKey(), companyId: text("company_id").notNull(), value: integer("value").notNull(), payload: jsonb("payload") });
+const encoded = pgTable("prepared_encoded_fixture", { id: text("id").primaryKey(), payload: runEventPayload("payload") });
+const support = await getEmbeddedPostgresTestSupport();
+
+describe.skipIf(!support.supported)("explicit ORM prepared statements on PostgreSQL", () => {
+  let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
+  let setup: Db;
+  const clients: Db[] = [];
+  beforeAll(async () => {
+    temporary = await startEmbeddedPostgresTestDatabase("prepared-orm-");
+    setup = createDb(temporary.connectionString, { prepare: false, maxConnections: 1 });
+    await setup.execute(sql`create table prepared_query_fixture(id text primary key, company_id text not null, value integer not null, payload jsonb)`);
+    await setup.execute(sql`create table prepared_encoded_fixture(id text primary key, payload jsonb)`);
+    await setup.execute(sql`create table prepared_schema_change_fixture(id text primary key)`);
+    await setup.execute(sql`insert into prepared_schema_change_fixture(id) values ('retained')`);
+  }, 90_000);
+  beforeEach(async () => {
+    await setup.execute(sql`truncate prepared_query_fixture, prepared_encoded_fixture`);
+    await setup.insert(records).values([
+      { id: "a", companyId: "company-a", value: 1, payload: { owner: "a" } },
+      { id: "b", companyId: "company-b", value: 2, payload: { owner: "b" } },
+    ]);
+  });
+  afterEach(async () => {
+    for (const db of clients.splice(0)) {
+      await closeLivenessDb(db, 2);
+      await db.$client.end();
+    }
+  });
+  afterAll(async () => { await setup?.$client.end(); await temporary?.cleanup(); });
+  function fixture(options: DatabaseClientOptions = {}) {
+    const db = createDb(temporary.connectionString, { ...options, maxConnections: 1 });
+    clients.push(db); return db;
+  }
+  async function statements(db: Db, pattern = '%"prepared_query_fixture"%') {
+    return db.$client.unsafe("select name,statement from pg_prepared_statements where statement like $1", [pattern]);
+  }
+
+  it.each([undefined, false])("keeps the existing ORM mode when prepare is %s", async prepare => {
+    const db = fixture({ prepare });
+    expect(await db.select().from(records).where(eq(records.companyId, "company-a"))).toEqual([
+      { id: "a", companyId: "company-a", value: 1, payload: { owner: "a" } },
+    ]);
+    expect(await statements(db)).toHaveLength(0);
+  });
+
+  it("honors the explicit environment opt-in for ORM queries with fresh company bindings", async () => {
+    const db = fixture(databaseClientOptionsFromEnv({ DATABASE_PREPARED_STATEMENTS: "true" }));
+    for (let i = 0; i < 12; i++) {
+      const companyId = i % 2 ? "company-b" : "company-a";
+      const rows = await db.select().from(records).where(eq(records.companyId, companyId));
+      expect(rows).toEqual([{ id: i % 2 ? "b" : "a", companyId, value: i % 2 ? 2 : 1, payload: { owner: i % 2 ? "b" : "a" } }]);
+    }
+    expect(await statements(db)).toHaveLength(1);
+  });
+
+  it("preserves explicit per-query opt-out", async () => {
+    const db = fixture({ prepare: true });
+    expect((await db.$client.unsafe("select $1::integer as explicit_opt_out_marker", [7], { prepare: false }))[0].explicit_opt_out_marker).toBe(7);
+    expect(await statements(db, "%explicit_opt_out_marker%")).toHaveLength(0);
+  });
+
+  it("preserves simple queries and the two-argument options overload", async () => {
+    const db = fixture({ prepare: true });
+    const simpleOptions = { prepare: true, simple: true };
+    const simple = await db.$client.unsafe("select 1 as first; select 2 as second", [], simpleOptions);
+    expect(simple).toEqual([[{ first: 1 }], [{ second: 2 }]]);
+    const unsafe = db.$client.unsafe as unknown as (query: string, options: { simple: boolean }) => Promise<unknown>;
+    expect(await unsafe("select 3 as third; select 4 as fourth", { simple: true }))
+      .toEqual([[{ third: 3 }], [{ fourth: 4 }]]);
+  });
+
+  it("does not replay a rejected prepared write", async () => {
+    const db = fixture({ prepare: true });
+    await expect(db.insert(records).values({ id: "a", companyId: "company-b", value: 9 }))
+      .rejects.toMatchObject({ cause: { code: "23505" } });
+    expect(await db.select({ companyId: records.companyId, value: records.value }).from(records).where(eq(records.id, "a")))
+      .toEqual([{ companyId: "company-a", value: 1 }]);
+  });
+
+  it("keeps write queries lazy and repeated awaits from repeating writes", async () => {
+    const db = fixture({ prepare: true });
+    const pending = db.$client.unsafe("insert into prepared_query_fixture(id,company_id,value) values ($1,$2,$3) returning id", ["lazy", "company-a", 3]);
+    expect(typeof pending.values).toBe("function");
+    expect(await setup.select().from(records).where(eq(records.id, "lazy"))).toEqual([]);
+    expect(await pending).toEqual([{ id: "lazy" }]);
+    expect(await pending).toEqual([{ id: "lazy" }]);
+    expect(await setup.select({ id: records.id }).from(records).where(eq(records.id, "lazy"))).toEqual([{ id: "lazy" }]);
+    expect(await statements(db, "%insert into prepared_query_fixture%")).toHaveLength(1);
+  });
+
+  it("preserves transaction commits, nested savepoint rollback, and scoped reads", async () => {
+    const db = fixture({ prepare: true });
+    await db.transaction(async tx => {
+      expect(await tx.select({ value: records.value }).from(records).where(eq(records.companyId, "company-a"))).toEqual([{ value: 1 }]);
+      await expect(tx.transaction(async nested => {
+        await nested.update(records).set({ value: 99 }).where(eq(records.companyId, "company-a"));
+        throw new Error("rollback savepoint");
+      })).rejects.toThrow("rollback savepoint");
+      expect(await tx.select({ value: records.value }).from(records).where(eq(records.companyId, "company-a"))).toEqual([{ value: 1 }]);
+      await tx.insert(records).values({ id: "committed", companyId: "company-b", value: 3 });
+      const names = await tx.execute(sql`select statement from pg_prepared_statements`);
+      expect(names.some(row => String(row.statement).includes('"prepared_query_fixture"'))).toBe(true);
+    }, { isolationLevel: "serializable" });
+    expect(await db.select({ value: records.value }).from(records).where(eq(records.id, "committed"))).toEqual([{ value: 3 }]);
+  });
+
+  it("rolls back failed outer transactions without replay", async () => {
+    const db = fixture({ prepare: true });
+    await expect(db.transaction(async tx => {
+      await tx.update(records).set({ value: 42 }).where(eq(records.id, "a"));
+      throw new Error("rollback outer transaction");
+    })).rejects.toThrow("rollback outer transaction");
+    expect(await db.select({ value: records.value }).from(records).where(eq(records.id, "a"))).toEqual([{ value: 1 }]);
+  });
+
+  it("prepares reserved-client queries and keeps their bound values fresh", async () => {
+    const db = fixture({ prepare: true });
+    const reserved = await db.$client.reserve();
+    try {
+      for (const marker of [1, 2, 3]) {
+        expect((await reserved.unsafe("select $1::integer as reserved_prepared_marker", [marker]))[0].reserved_prepared_marker).toBe(marker);
+      }
+      const names = await reserved.unsafe("select statement from pg_prepared_statements where statement like $1", ["%reserved_prepared_marker%"]);
+      expect(names).toHaveLength(1);
+    } finally { reserved.release(); }
+  });
+
+  it("preserves object and array results for the same prepared shape", async () => {
+    const db = fixture({ prepare: true });
+    expect(await db.$client.unsafe("select $1::integer as result_mode_marker", [41]).values()).toEqual([[41]]);
+    expect(await db.$client.unsafe("select $1::integer as result_mode_marker", [42])).toEqual([{ result_mode_marker: 42 }]);
+  });
+
+  it("keeps custom stored-payload decoding fresh", async () => {
+    const db = fixture({ prepare: true });
+    const first = { text: "encoded first payload🙂".repeat(2048) };
+    const second = { text: "encoded second payload🙂".repeat(2048) };
+    await setup.insert(encoded).values({ id: "encoded", payload: first });
+    expect((await db.select().from(encoded).where(eq(encoded.id, "encoded")))[0].payload).toEqual(first);
+    await setup.update(encoded).set({ payload: second }).where(eq(encoded.id, "encoded"));
+    expect((await db.select().from(encoded).where(eq(encoded.id, "encoded")))[0].payload).toEqual(second);
+    expect(await statements(db, '%"prepared_encoded_fixture"%')).toHaveLength(1);
+  });
+
+  it("inherits the opt-in in dedicated and liveness clients", async () => {
+    const db = fixture({ prepare: true });
+    await withDedicatedDbConnection(db, async dedicated => {
+      expect(await dedicated.select({ value: records.value }).from(records).where(eq(records.companyId, "company-b"))).toEqual([{ value: 2 }]);
+      expect(await statements(dedicated)).toHaveLength(1);
+    });
+    const live = livenessDb(db);
+    expect(await live.select({ value: records.value }).from(records).where(eq(records.companyId, "company-a"))).toEqual([{ value: 1 }]);
+    expect(await statements(live)).toHaveLength(1);
+  });
+
+  it("refreshes prepared result metadata after a schema change", async () => {
+    const db = fixture({ prepare: true });
+    const query = "select * from prepared_schema_change_fixture where id=$1";
+    expect(await db.$client.unsafe(query, ["retained"])).toEqual([{ id: "retained" }]);
+    try {
+      await setup.execute(sql`alter table prepared_schema_change_fixture add column extra text`);
+      expect(await db.$client.unsafe(query, ["retained"])).toEqual([{ id: "retained", extra: null }]);
+    } finally {
+      await setup.execute(sql`alter table prepared_schema_change_fixture drop column if exists extra`);
+    }
+  });
+});

packages/db/src/prepared-queries.ts

@@ -0,0 +1,47 @@
+type Method = (this: unknown, ...args: unknown[]) => unknown;
+
+export function installPreparedQueries(client: unknown): void {
+  const visited = new WeakSet<object>();
+
+  function visit(value: unknown): void {
+    if ((typeof value !== "object" || value === null) && typeof value !== "function") return;
+    if (visited.has(value)) return;
+    visited.add(value);
+    const target = value as Record<string, unknown>;
+    const unsafe = target.unsafe;
+    if (typeof unsafe === "function") {
+      target.unsafe = function (this: unknown, ...args: unknown[]) {
+        const index = args.length === 2 && !Array.isArray(args[1]) ? 1 : 2;
+        const options = args[index];
+        if (options !== undefined && (options === null || typeof options !== "object")) {
+          return Reflect.apply(unsafe, this, args);
+        }
+        if (index === 2 && args.length < 2) args[1] = [];
+        args[index] = { prepare: true, ...options };
+        return Reflect.apply(unsafe, this, args);
+      };
+    }
+    for (const method of ["begin", "savepoint"]) {
+      const original = target[method];
+      if (typeof original !== "function") continue;
+      target[method] = function (this: unknown, ...args: unknown[]) {
+        const wrapped = args.map(arg => typeof arg !== "function" ? arg : function (this: unknown, ...inner: unknown[]) {
+          visit(inner[0]);
+          return Reflect.apply(arg, this, inner);
+        });
+        return Reflect.apply(original, this, wrapped);
+      } satisfies Method;
+    }
+    const reserve = target.reserve;
+    if (typeof reserve === "function") {
+      target.reserve = function (this: unknown, ...args: unknown[]) {
+        return Promise.resolve(Reflect.apply(reserve, this, args)).then(reserved => {
+          visit(reserved);
+          return reserved;
+        });
+      } satisfies Method;
+    }
+  }
+
+  visit(client);
+}

packages/db/src/response-capture-postgres.test.ts

@@ -31,11 +31,11 @@ describe.skipIf(!support.supported)("database response capture PostgreSQL behavi
     vi.restoreAllMocks(); vi.unstubAllEnvs();
   });
   afterAll(async () => { await temporary?.cleanup(); });
-  function fixture() {
+  function fixture(prepare?: boolean) {
     const events: Array<Record<string, unknown>> = [];
     const now = Date.now();
     const capture = createDatabaseResponseCapture({ start: now - 100, end: now + 59_000 }, event => events.push(event));
-    const db = createDb(temporary.connectionString);
+    const db = createDb(temporary.connectionString, { prepare });
     clients.push(db); captures.push(capture);
     expect(installDatabaseResponseCapture(db, { table, encodedTable }, capture)).toBe(db);
     expect(installDatabaseResponseCapture(db, { table, encodedTable }, capture)).toBe(db);
@@ -43,6 +43,18 @@ describe.skipIf(!support.supported)("database response capture PostgreSQL behavi
   }
   function summary(events: Array<Record<string, unknown>>) { return events.find(event => event.event === "paperclip.db_response_capture.summary.v1")!; }
 
+  it("retains capture counts when ORM preparation is enabled", async () => {
+    const f = fixture(true), id = randomUUID(), payload = "prepared-capture-value";
+    await f.db.insert(table).values({ id, payload });
+    await f.db.transaction(async tx => {
+      expect(await tx.select().from(table).where(eq(table.id, id))).toEqual([{ id, payload }]);
+    });
+    f.capture.flush();
+    expect(summary(f.events)).toMatchObject({ startedQueries: 2, completedQueries: 2 });
+    expect(JSON.stringify(f.events)).not.toContain(payload);
+    expect(JSON.stringify(f.events)).not.toContain(id);
+  });
+
   it("keeps rows and builder identity through lazy select and write returning", async () => {
     const f = fixture(), id = randomUUID(), payload = "sentinel-result-and-parameter-value🙂\n\"";
     const insert = f.db.insert(table).values({ id, payload });
@@ -173,11 +185,11 @@ describe.skipIf(!support.supported)("database response capture PostgreSQL behavi
     f.capture.flush();
     expect(summary(f.events).completedQueries).toBeGreaterThan(4);
   });
-  it("covers automatically created dedicated pools and retains owner identity", async () => {
+  it.each([undefined, true])("covers automatically created dedicated pools with prepare %s and retains owner identity", async prepare => {
     const logs: string[] = [], now = Date.now();
     vi.spyOn(console, "info").mockImplementation(value => logs.push(String(value)));
     vi.stubEnv("PAPERCLIP_DB_RESPONSE_CAPTURE_WINDOW", `${new Date(now - 100).toISOString()}/${new Date(now + 59_000).toISOString()}`);
-    const db = createDb(temporary.connectionString); clients.push(db);
+    const db = createDb(temporary.connectionString, { prepare }); clients.push(db);
     const capture = databaseResponseCaptureFromEnv()!; captures.push(capture);
     expect(await withDedicatedDbConnection(db, async dedicated => dedicated.execute(sql`select ${"dedicated-value"}::text as value`)))
       .toEqual([{ value: "dedicated-value" }]);