diff --git a/doc/DATABASE.md b/doc/DATABASE.md index 13424c8fad6..bc1292bbe94 100644 --- a/doc/DATABASE.md +++ b/doc/DATABASE.md @@ -320,6 +320,31 @@ DATABASE_MAX_LIFETIME_SECONDS=1800 # recycle a pooled connection after this lo DATABASE_APPLICATION_NAME=paperclip # application_name in pg_stat_activity; default: paperclip ``` +### Bounded response diagnostics + +`PAPERCLIP_DB_RESPONSE_CAPTURE_WINDOW=/` enables one explicit UTC +window. Both timestamps use ISO format ending in `Z`. The window lasts at most +60 seconds and starts within 30 minutes. Unset, invalid, and expired windows +leave database construction unchanged. An expired window stays inactive after a +restart; remove the setting after the investigation. + +The diagnostic measures awaited Postgres.js `unsafe` responses before Drizzle +column decoding. It retains the Drizzle builder caller for normal, prepared, +transaction, and dedicated-pool execution. It issues no queries. +It logs the public server caller locations, static schema table, operation, +query/row counts, and response value sizes. It emits at most 128 aggregate +groups and a summary at the deadline. It never logs SQL, parameters, result +values, connection strings, or query errors. In-flight queries continue normally. + +Look for `paperclip.db_response_capture.entry.v1` and +`paperclip.db_response_capture.summary.v1` in the process logs. Tagged raw-driver calls, streams, unawaited raw-driver +dispatch, and protocol overhead are not measured. Relational queries use a driver-level caller fallback. +Queries still in flight at the deadline are reported separately. Sizes describe +driver values before column decoding, not billed network bytes. Complex parsed +values use JSON size estimates, counted in `estimatedValues`. Reconcile them with +provider metering and socket measurements. Serialization adds temporary CPU and memory work; +measure the final transfer rate after the window expires. + ### Push the schema ```sh diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index 5149c1c9b27..f1b8079db45 100644 --- a/packages/db/src/client.ts +++ b/packages/db/src/client.ts @@ -4,6 +4,7 @@ import { readFile, readdir } from "node:fs/promises"; import { fileURLToPath } from "node:url"; import postgres from "postgres"; import { installDatabaseWorkSignals } from "./work-signals.js"; +import { installDatabaseResponseCapture } from "./response-capture.js"; import * as schema from "./schema/index.js"; const MIGRATIONS_FOLDER = fileURLToPath(new URL("./migrations", import.meta.url)); @@ -307,6 +308,7 @@ function createDbClient(url: string, resolved: DatabaseClientOptions, owner?: ob // message cannot establish that replay is safe. Leave retries to callers // that know the complete operation is idempotent. const db = installDatabaseWorkSignals(drizzlePg(sql, { schema }), owner); + installDatabaseResponseCapture(db, schema); dedicatedDbFactories.set(db, () => createDbClient(url, { ...resolved, maxConnections: 1, applicationName: "paperclip-workspace-finalization-lock", }, db)); diff --git a/packages/db/src/response-capture-postgres.test.ts b/packages/db/src/response-capture-postgres.test.ts new file mode 100644 index 00000000000..370d738d056 --- /dev/null +++ b/packages/db/src/response-capture-postgres.test.ts @@ -0,0 +1,190 @@ +import { randomUUID } from "node:crypto"; +import { compileFunction } from "node:vm"; +import { eq, sql } from "drizzle-orm"; +import { pgTable, text } from "drizzle-orm/pg-core"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; +import { createDb, withDedicatedDbConnection } from "./client.js"; +import { createDatabaseResponseCapture, databaseResponseCaptureFromEnv, installDatabaseResponseCapture } from "./response-capture.js"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./test-embedded-postgres.js"; +import { encodeRunEventPayload, runEventPayload } from "./run-event-payload.js"; +import { databaseWorkPending, reconcileDatabaseWork, signalDatabaseWork, subscribeDatabaseWork } from "./work-signals.js"; + +const table = pgTable("response_capture_fixture", { id: text("id").primaryKey(), payload: text("payload").notNull() }); +const encodedTable = pgTable("response_capture_encoded_fixture", { id: text("id").primaryKey(), payload: runEventPayload("payload") }); +const support = await getEmbeddedPostgresTestSupport(); +describe.skipIf(!support.supported)("database response capture PostgreSQL behavior", () => { + let temporary: Awaited>; + const clients: Array> = []; + const captures: Array> = []; + beforeAll(async () => { + temporary = await startEmbeddedPostgresTestDatabase("response-capture-"); + const db = createDb(temporary.connectionString); + try { + await db.execute(sql`create table response_capture_fixture (id text primary key, payload text not null)`); + await db.execute(sql`create table response_capture_encoded_fixture (id text primary key, payload jsonb not null)`); + } + finally { await db.$client.end(); } + }, 90_000); + afterEach(async () => { + for (const capture of captures.splice(0)) capture.flush(); + for (const client of clients.splice(0)) await client.$client.end(); + vi.restoreAllMocks(); vi.unstubAllEnvs(); + }); + afterAll(async () => { await temporary?.cleanup(); }); + function fixture() { + const events: Array> = []; + const now = Date.now(); + const capture = createDatabaseResponseCapture({ start: now - 100, end: now + 59_000 }, event => events.push(event)); + const db = createDb(temporary.connectionString); + clients.push(db); captures.push(capture); + expect(installDatabaseResponseCapture(db, { table, encodedTable }, capture)).toBe(db); + expect(installDatabaseResponseCapture(db, { table, encodedTable }, capture)).toBe(db); + return { db, capture, events }; + } + function summary(events: Array>) { return events.find(event => event.event === "paperclip.db_response_capture.summary.v1")!; } + + 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 }); + expect(insert.returning()).toBe(insert); + expect(await insert).toEqual([{ id, payload }]); + const query = f.db.select().from(table); + expect(query.where(eq(table.id, id))).toBe(query); + expect(await query).toEqual([{ id, payload }]); + const updated = await f.db.update(table).set({ payload: "changed" }).where(eq(table.id, id)).returning(); + expect(updated).toEqual([{ id, payload: "changed" }]); + expect(await f.db.delete(table).where(eq(table.id, id)).returning()).toEqual([{ id, payload: "changed" }]); + f.capture.flush(); + expect(summary(f.events)).toMatchObject({ startedQueries: 4, completedQueries: 4 }); + expect(f.events.filter(event => event.event === "paperclip.db_response_capture.entry.v1").map(event => event.operation).sort()) + .toEqual(["delete", "insert", "select", "update"]); + expect(JSON.stringify(f.events)).not.toContain(payload); + expect(JSON.stringify(f.events)).not.toContain(id); + }); + it("counts the stored encoding before a custom column decoder changes the value", async () => { + const setup = createDb(temporary.connectionString); clients.push(setup); + const id = randomUUID(), payload = { text: "codec-payload-sentinel\u0000value".repeat(512) }; + await setup.insert(encodedTable).values({ id, payload }); + const f = fixture(); + expect(await f.db.select().from(encodedTable).where(eq(encodedTable.id, id))).toEqual([{ id, payload }]); + f.capture.flush(); + const entry = f.events.find(event => event.table === "response_capture_encoded_fixture")!; + expect(entry.responseValueBytes).toBe(Buffer.byteLength(id) + Buffer.byteLength(encodeRunEventPayload(payload))); + expect(entry.responseValueBytes).toBeGreaterThan(Buffer.byteLength(JSON.stringify({ id, payload }))); + expect(JSON.stringify(f.events)).not.toContain("codec-payload-sentinel"); + }); + it("keeps a raw driver query lazy and counts repeated awaits without replay", async () => { + const f = fixture(), id = randomUUID(); + const observer = createDb(temporary.connectionString); clients.push(observer); + const pending = f.db.$client.unsafe("insert into response_capture_fixture (id, payload) values ($1, $2) returning id", [id, "raw-lazy-sentinel"]); + expect(await observer.select().from(table).where(eq(table.id, id))).toEqual([]); + const first = await pending; + expect(await pending).toBe(first); + expect(first).toEqual([{ id }]); + expect(await observer.select().from(table).where(eq(table.id, id))).toEqual([{ id, payload: "raw-lazy-sentinel" }]); + f.capture.flush(); + expect(summary(f.events)).toMatchObject({ startedQueries: 1, completedQueries: 1 }); + expect(JSON.stringify(f.events)).not.toContain("raw-lazy-sentinel"); + expect(JSON.stringify(f.events)).not.toContain(id); + }); + it("counts prepared executions once and retains their parameter bindings", async () => { + const f = fixture(), id = randomUUID(); + await f.db.insert(table).values({ id, payload: "prepared-value" }); + const prepared = f.db.select().from(table).where(eq(table.id, sql.placeholder("id"))).prepare("response_capture_prepared"); + expect(await prepared.execute({ id })).toEqual([{ id, payload: "prepared-value" }]); + expect(await prepared.execute({ id: randomUUID() })).toEqual([]); + f.capture.flush(); + expect(summary(f.events)).toMatchObject({ startedQueries: 3, completedQueries: 3 }); + expect(JSON.stringify(f.events)).not.toContain("prepared-value"); + }); + it("retains distinct builder callers through normal, prepared, and transaction execution", async () => { + const f = fixture(), id = randomUUID(); + const setup = createDb(temporary.connectionString); clients.push(setup); + await setup.insert(table).values({ id, payload: "caller-sentinel" }); + const shape = f.db.select().from(table); + const caller = (name: string) => compileFunction( + "return db.select().from(table).where(eq(table.id, id))", ["db", "id"], + { filename: `/fixture/server/src/services/${name}.ts`, contextExtensions: [{ table, eq }] }, + ) as (db: Pick, id: string) => typeof shape; + const fromA = caller("capture-caller-a"), fromB = caller("capture-caller-b"); + const normalA = fromA(f.db, id), normalB = fromB(f.db, id); + expect(await normalA).toEqual([{ id, payload: "caller-sentinel" }]); + expect(await normalB).toEqual([{ id, payload: "caller-sentinel" }]); + const prepared = fromA(f.db, id).prepare("capture_caller_prepared"); + expect(await prepared.execute()).toEqual([{ id, payload: "caller-sentinel" }]); + await f.db.transaction(async tx => { + expect(await fromB(tx, id)).toEqual([{ id, payload: "caller-sentinel" }]); + }); + f.capture.flush(); + const entries = f.events.filter(event => event.event === "paperclip.db_response_capture.entry.v1"); + expect(entries).toHaveLength(2); + for (const name of ["a", "b"]) { + expect(entries.find(event => (event.callers as string[]).some(location => location.startsWith(`services/capture-caller-${name}.ts:`)))) + .toMatchObject({ operation: "select", table: "response_capture_fixture", calls: 2, rows: 2 }); + } + expect(summary(f.events)).toMatchObject({ startedQueries: 4, completedQueries: 4 }); + expect(JSON.stringify(f.events)).not.toContain("/fixture/"); + expect(JSON.stringify(f.events)).not.toContain("caller-sentinel"); + expect(JSON.stringify(f.events)).not.toContain(id); + }); + it("records the executed CTE rather than its unexecuted inner select", async () => { + const f = fixture(), id = randomUUID(); + await f.db.insert(table).values({ id, payload: "cte-value" }); + const cte = f.db.$with("capture_cte").as(f.db.select().from(table).where(eq(table.id, id))); + expect(await f.db.with(cte).select().from(cte)).toEqual([{ id, payload: "cte-value" }]); + f.capture.flush(); + expect(summary(f.events)).toMatchObject({ startedQueries: 2, completedQueries: 2 }); + }); + it("keeps raw execute results and rejection behavior without logging SQL or errors", async () => { + const f = fixture(); + expect(await f.db.execute(sql`select ${"sentinel-sql-parameter"}::text as value`)).toEqual([{ value: "sentinel-sql-parameter" }]); + await expect(f.db.execute(sql`select 1/0`)).rejects.toThrow(); + f.capture.flush(); + expect(summary(f.events)).toMatchObject({ startedQueries: 2, completedQueries: 2 }); + expect(f.events.find(event => event.operation === "execute")).toMatchObject({ queryErrors: 1 }); + expect(JSON.stringify(f.events)).not.toContain("sentinel-sql-parameter"); + expect(JSON.stringify(f.events)).not.toContain("select 1/0"); + }); + it("preserves nested commit, rollback, and durable work-signal boundaries", async () => { + const f = fixture(), id = randomUUID(), events: string[] = []; + const unsubscribe = subscribeDatabaseWork(f.db, "capture-topic", event => events.push(event)); + try { + expect(await f.db.transaction(async tx => { + await tx.transaction(async nested => { + await signalDatabaseWork(nested, "capture-topic"); + await nested.insert(table).values({ id, payload: "committed" }); + }); + expect(events).toEqual(["intent"]); expect(databaseWorkPending(f.db, "capture-topic")).toBe(true); + return 42; + })).toBe(42); + expect(events).toEqual(["intent", "settled"]); + expect(databaseWorkPending(f.db, "capture-topic")).toBe(false); + const rolledBack = randomUUID(); + await expect(f.db.transaction(async tx => { + await signalDatabaseWork(tx, "capture-topic"); + await tx.insert(table).values({ id: rolledBack, payload: "rolled-back" }); + throw new Error("rollback fixture"); + })).rejects.toThrow("rollback fixture"); + await reconcileDatabaseWork(f.db, "capture-topic"); + expect(databaseWorkPending(f.db, "capture-topic")).toBe(false); + expect(await f.db.select().from(table).where(eq(table.id, rolledBack))).toEqual([]); + expect(await f.db.select().from(table).where(eq(table.id, id))).toEqual([{ id, payload: "committed" }]); + } finally { unsubscribe(); } + f.capture.flush(); + expect(summary(f.events).completedQueries).toBeGreaterThan(4); + }); + it("covers automatically created dedicated pools and retains owner identity", async () => { + 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 capture = databaseResponseCaptureFromEnv()!; captures.push(capture); + expect(await withDedicatedDbConnection(db, async dedicated => dedicated.execute(sql`select ${"dedicated-value"}::text as value`))) + .toEqual([{ value: "dedicated-value" }]); + capture.flush(); + const events = logs.map(line => JSON.parse(line) as Record); + expect(summary(events)).toMatchObject({ startedQueries: 1, completedQueries: 1 }); + expect(logs.join("\n")).not.toContain("dedicated-value"); + expect(logs.join("\n")).not.toContain(temporary.connectionString); + }); +}); diff --git a/packages/db/src/response-capture.test.ts b/packages/db/src/response-capture.test.ts new file mode 100644 index 00000000000..65f58d10883 --- /dev/null +++ b/packages/db/src/response-capture.test.ts @@ -0,0 +1,133 @@ +import { integer, pgTable } from "drizzle-orm/pg-core"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { createDatabaseResponseCapture, installDatabaseResponseCapture, publicDatabaseCallers, responseCaptureWindow } from "./response-capture.js"; + +const epoch = Date.parse("2020-01-01T00:00:00Z"); +const metadata = { operation: "select", table: "capture_fixture", callers: ["services/fixture.ts:4:2"] }; +const table = pgTable("capture_fixture", { id: integer("id") }); +const captures: Array> = []; +afterEach(() => { for (const capture of captures.splice(0)) capture.flush(); vi.unstubAllEnvs(); }); +function fixture(emitOverride?: (event: Record) => void) { + let clock = epoch; + const events: Array> = []; + const capture = createDatabaseResponseCapture({ start: epoch, end: epoch + 60_000 }, emitOverride ?? (event => events.push(event)), () => clock); + captures.push(capture); + return { capture, events, now: (value: number) => { clock = value; } }; +} + +describe("bounded database response capture", () => { + it.each([undefined, "", "sentinel-invalid-input", "2020-01-01T00:00:00Z", "2020-01-01T00:00:00Z/2020-01-01T00:00:00Z", + "2020-01-01T00:00:00Z/2020-01-01T00:01:01Z", "2020-01-01T01:00:00Z/2020-01-01T01:01:00Z", + "2020-01-01T00:00:00Z/2020-01-01T00:01:00Z/extra", + ])("disables absent, malformed, expired, or unbounded windows: %s", value => { + expect(responseCaptureWindow(value, epoch + 60_000)).toBeUndefined(); + }); + it("accepts a fixed future window of at most one minute", () => { + expect(responseCaptureWindow("2020-01-01T00:01:00Z/2020-01-01T00:02:00Z", epoch)) + .toEqual({ start: epoch + 60_000, end: epoch + 120_000 }); + expect(() => createDatabaseResponseCapture({ start: epoch, end: epoch + 60_001 }, () => {})).toThrow("Invalid"); + }); + it("keeps only public server file locations from stacks", () => { + const stack = ["Error: sentinel-message", "at f (/private/fixture/server/src/services/example.ts:4:2)", + "at f (/private/fixture/server/src/services/example.ts:4:2)", "at f (/private/fixture/server/dist/routes/example.js:5:3)", + "at f (/private/fixture/customer-file.js:7:9)", "at f (/app/server/src/services/../../customer-file.ts:8:1)"].join("\n"); + expect(publicDatabaseCallers(stack)).toEqual(["services/example.ts:4:2", "routes/example.js:5:3"]); + }); + it("leaves methods and identity unchanged when disabled", () => { + vi.stubEnv("PAPERCLIP_DB_RESPONSE_CAPTURE_WINDOW", undefined); + const db = { select: vi.fn(), transaction: vi.fn() }, select = db.select, transaction = db.transaction; + expect(installDatabaseResponseCapture(db, { table })).toBe(db); + expect(db.select).toBe(select); expect(db.transaction).toBe(transaction); + }); + it("preserves exact values and only emits counts, sizes, and caller metadata", async () => { + const f = fixture(), rows = [{ text: "sentinel-result-value🙂\n\"", n: 4n }]; + expect(await f.capture.execute(() => Promise.resolve(rows), metadata)).toBe(rows); + f.capture.flush(); + const entry = f.events.find(event => event.event === "paperclip.db_response_capture.entry.v1")!; + expect(entry).toMatchObject({ calls: 1, rows: 1, queryErrors: 0, serializationErrors: 0, + responseValueBytes: Buffer.byteLength(JSON.stringify(rows, (_, value) => typeof value === "bigint" ? value.toString() : value)) }); + expect(JSON.stringify(f.events)).not.toContain("sentinel-result-value"); + expect(f.events.at(-1)).toMatchObject({ startedQueries: 1, completedQueries: 1, inFlightAtEnd: 0, + rowValuesLogged: false, sqlOrParametersLogged: false, wireBytesMeasured: false }); + }); + it("counts driver text values without adding JSON escaping or column names", async () => { + const f = fixture(), text = "sentinel-driver-text\n\""; + const rows = Object.assign([[text, true, 42n, null]], { columns: [ + { name: "sentinel-column-name" }, { name: "b" }, { name: "n" }, { name: "empty" } ] }); + expect(await f.capture.execute(() => Promise.resolve(rows), metadata)).toBe(rows); + f.capture.flush(); + expect(f.events[0]).toMatchObject({ responseValueBytes: Buffer.byteLength(text) + 1 + 2, estimatedValues: 0 }); + expect(JSON.stringify(f.events)).not.toContain("sentinel-driver-text"); + expect(JSON.stringify(f.events)).not.toContain("sentinel-column-name"); + }); + it("does not alter synchronous or asynchronous errors or retry them", async () => { + const f = fixture(), error = new Error("sentinel-private-error"); + const sync = vi.fn(() => { throw error; }), asyncFailure = vi.fn(() => Promise.reject(error)); + expect(() => f.capture.execute(sync, metadata)).toThrow(error); + await expect(f.capture.execute(asyncFailure, metadata)).rejects.toBe(error); + expect(sync).toHaveBeenCalledTimes(1); expect(asyncFailure).toHaveBeenCalledTimes(1); + f.capture.flush(); + expect(f.events[0]).toMatchObject({ calls: 2, queryErrors: 2, responseValueBytes: 0 }); + expect(JSON.stringify(f.events)).not.toContain("sentinel-private-error"); + }); + it("keeps unserializable results and logging failures out of the query outcome", () => { + const f = fixture(), circular: Record = {}; circular.self = circular; + expect(f.capture.execute(() => circular, metadata)).toBe(circular); + f.capture.flush(); + expect(f.events[0]).toMatchObject({ calls: 1, serializationErrors: 1 }); + const brokenLogger = fixture(() => { throw new Error("logger fixture"); }); + expect(brokenLogger.capture.execute(() => circular, metadata)).toBe(circular); + expect(() => brokenLogger.capture.flush()).not.toThrow(); + }); + it("stops accounting at the deadline and leaves in-flight work untouched", async () => { + const f = fixture(); + let resolve!: (rows: unknown[]) => void; + const pending = new Promise(done => { resolve = done; }); + const result = f.capture.execute(() => pending, metadata); + f.now(epoch + 60_000); f.capture.flush(); + expect(f.events.at(-1)).toMatchObject({ startedQueries: 1, completedQueries: 0, inFlightAtEnd: 1 }); + const rows = [{ marker: "late-result-value" }]; resolve(rows); + expect(await result).toBe(rows); + const direct = Promise.resolve(rows); + expect(f.capture.execute(() => direct, metadata)).toBe(direct); + expect(f.events).toHaveLength(1); + }); + it("retains the deadline count when a query resolves before a delayed flush", async () => { + const f = fixture(); + let resolve!: (rows: unknown[]) => void; + const pending = new Promise(done => { resolve = done; }); + const result = f.capture.execute(() => pending, metadata); + f.now(epoch + 60_001); + const rows = [{ marker: "late-before-flush-sentinel" }]; resolve(rows); + expect(await result).toBe(rows); + f.capture.flush(); + expect(f.events).toHaveLength(1); + expect(f.events[0]).toMatchObject({ startedQueries: 1, completedQueries: 0, inFlightAtEnd: 1, responseValueBytes: 0 }); + expect(JSON.stringify(f.events)).not.toContain("late-before-flush-sentinel"); + }); + it("bounds cardinality without dropping response-byte totals", () => { + const f = fixture(); + for (let i = 0; i < 200; i += 1) f.capture.execute(() => [i], { ...metadata, callers: [`services/f${i}.ts:1:1`] }); + f.capture.flush(); + const summary = f.events.at(-1)!; + expect(summary).toMatchObject({ startedQueries: 200, completedQueries: 200, groups: 128, overflowCalls: 73 }); + expect(summary.responseValueBytes).toBe(Array.from({ length: 200 }, (_, i) => Buffer.byteLength(JSON.stringify([i]))).reduce((a, b) => a + b, 0)); + }); + it("does not execute builders or prepared statements until requested", async () => { + const f = fixture(), rows = [{ id: 1 }], execute = vi.fn(async () => rows); + const query = { execute, prepare: () => ({ execute }) }; + const builder = { from: () => query }; + const db = { select: () => builder }; + expect(installDatabaseResponseCapture(db, { table }, f.capture)).toBe(db); + expect(db.select()).toBe(builder); + expect(db.select().from()).toBe(query); + expect(execute).not.toHaveBeenCalled(); + const prepared = query.prepare(); + expect(execute).not.toHaveBeenCalled(); + expect(await prepared.execute()).toBe(rows); + expect(await query.execute()).toBe(rows); + expect(execute).toHaveBeenCalledTimes(2); + f.capture.flush(); + expect(f.events.at(-1)).toMatchObject({ completedQueries: 0 }); + }); +}); diff --git a/packages/db/src/response-capture.ts b/packages/db/src/response-capture.ts new file mode 100644 index 00000000000..8f738043f4d --- /dev/null +++ b/packages/db/src/response-capture.ts @@ -0,0 +1,273 @@ +import { AsyncLocalStorage } from "node:async_hooks"; +import { randomUUID } from "node:crypto"; +import { getTableName, is, Table } from "drizzle-orm"; + +type Metadata = { operation: string; table: string; callers: string[] }; +type Entry = Metadata & { calls: number; rows: number; bytes: number; maxBytes: number; queryErrors: number; serializationErrors: number; estimatedValues: number }; +type Event = Record; +type Method = (this: unknown, ...args: unknown[]) => unknown; +export type ResponseCaptureWindow = { start: number; end: number }; + +export function responseCaptureWindow(value: string | undefined, now = Date.now()): ResponseCaptureWindow | undefined { + if (!value) return; + const parts = value.split("/"); + if (parts.length !== 2 || parts.some(part => !/^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d{3})?Z$/.test(part))) return; + const start = Date.parse(parts[0]!), end = Date.parse(parts[1]!); + if (!Number.isFinite(start) || !Number.isFinite(end) || end <= start || end - start > 60_000 || end <= now || start > now + 30 * 60_000) return; + return { start, end }; +} + +export function publicDatabaseCallers(stack: string | undefined): string[] { + const callers: string[] = []; + for (const line of (stack ?? "").split("\n")) { + const match = line.match(/[/\\]server[/\\](?:src|dist)[/\\]((?:services|routes|modules|middleware)[/\\][A-Za-z0-9_./\\-]+\.[cm]?[jt]s):(\d+):(\d+)/); + if (!match) continue; + const relative = match[1]!.replaceAll("\\", "/"); + if (relative.split("/").includes("..")) continue; + const caller = `${relative}:${match[2]}:${match[3]}`; + if (!callers.includes(caller)) callers.push(caller); + if (callers.length === 4) break; + } + return callers; +} + +function responseValueSize(value: unknown): { bytes: number; estimatedValues: number } { + const columns = value && typeof value === "object" ? (value as { columns?: unknown }).columns : undefined; + if (!Array.isArray(value) || !Array.isArray(columns)) { + const serialized = JSON.stringify(value, (_, item: unknown) => typeof item === "bigint" ? item.toString() : item); + return { bytes: serialized === undefined ? 0 : Buffer.byteLength(serialized), estimatedValues: 1 }; + } + let bytes = 0, estimatedValues = 0; + for (const row of value) { + const cells = Array.isArray(row) ? row : columns.map(column => row[column.name]); + for (const cell of cells) { + if (cell === null || cell === undefined) continue; + if (typeof cell === "string") bytes += Buffer.byteLength(cell); + else if (typeof cell === "boolean") bytes += 1; + else if (typeof cell === "number" || typeof cell === "bigint") bytes += Buffer.byteLength(String(cell)); + else { + estimatedValues += 1; + const serialized = JSON.stringify(cell, (_, item: unknown) => typeof item === "bigint" ? item.toString() : item); + if (serialized !== undefined) bytes += Buffer.byteLength(serialized); + } + } + } + return { bytes, estimatedValues }; +} + +export function createDatabaseResponseCapture( + window: ResponseCaptureWindow, + emit: (event: Event) => void, + now: () => number = Date.now, +) { + if (!Number.isFinite(new Date(window.start).getTime()) || !Number.isFinite(new Date(window.end).getTime()) || + window.end <= window.start || window.end - window.start > 60_000) throw new Error("Invalid database response capture window"); + const id = randomUUID(), entries = new Map(); + const roots = new WeakSet(), queries = new WeakSet(), factories = new WeakSet(); + const clients = new WeakSet(), driverQueries = new WeakSet(); + const scope = new AsyncLocalStorage(); + let closed = false, inFlight = 0, started = 0, completed = 0, overflowCalls = 0; + const send = (event: Event) => { try { emit(event); } catch {} }; + const expired = () => closed || now() >= window.end; + const active = () => !expired() && now() >= window.start; + const flush = () => { + if (closed) return; + closed = true; + clearTimeout(timer); + scope.disable(); + const ordered = [...entries.values()].sort((a, b) => b.bytes - a.bytes); + for (const entry of ordered) send({ event: "paperclip.db_response_capture.entry.v1", captureId: id, + operation: entry.operation, table: entry.table, callers: entry.callers, calls: entry.calls, rows: entry.rows, + responseValueBytes: entry.bytes, maxResponseValueBytes: entry.maxBytes, estimatedValues: entry.estimatedValues, + queryErrors: entry.queryErrors, serializationErrors: entry.serializationErrors }); + send({ event: "paperclip.db_response_capture.summary.v1", captureId: id, + windowStart: new Date(window.start).toISOString(), windowEnd: new Date(window.end).toISOString(), + closedAt: new Date(now()).toISOString(), + startedQueries: started, completedQueries: completed, inFlightAtEnd: inFlight, groups: entries.size, + overflowCalls, responseValueBytes: ordered.reduce((sum, entry) => sum + entry.bytes, 0), + rowValuesLogged: false, sqlOrParametersLogged: false, wireBytesMeasured: false, + measurementBoundary: "postgres-js results before Drizzle column decoding", + relationalQueriesMeasured: true, unsafeDriverQueriesMeasured: true, + taggedDriverQueriesMeasured: false, streamingQueriesMeasured: false }); + entries.clear(); + }; + const timer = setTimeout(flush, Math.max(0, window.end - now())); + timer.unref(); + function finish(metadata: Metadata, value: unknown, failed: boolean) { + if (!active()) return; + inFlight -= 1; + completed += 1; + try { + let key = JSON.stringify(metadata); + if (!entries.has(key) && entries.size >= 127) { + overflowCalls += 1; + metadata = { operation: "other", table: "other", callers: [] }; + key = "overflow"; + } + let entry = entries.get(key); + if (!entry) { + entry = { ...metadata, calls: 0, rows: 0, bytes: 0, maxBytes: 0, queryErrors: 0, serializationErrors: 0, estimatedValues: 0 }; + entries.set(key, entry); + } + entry.calls += 1; + if (failed) { entry.queryErrors += 1; return; } + entry.rows += Array.isArray(value) ? value.length : 0; + try { + const size = responseValueSize(value); + entry.bytes += size.bytes; + entry.maxBytes = Math.max(entry.maxBytes, size.bytes); + entry.estimatedValues += size.estimatedValues; + } catch { entry.serializationErrors += 1; } + } catch {} + } + function execute(action: () => unknown, metadata: Metadata) { + if (!active()) return action(); + inFlight += 1; started += 1; + let result: unknown; + try { result = action(); } + catch (error) { finish(metadata, undefined, true); throw error; } + let asynchronous = false; + try { asynchronous = Boolean(result && typeof (result as { then?: unknown }).then === "function"); } catch {} + if (asynchronous) { + return Promise.resolve(result).then(value => { finish(metadata, value, false); return value; }, + error => { finish(metadata, undefined, true); throw error; }); + } + finish(metadata, result, false); + return result; + } + return { window, roots, queries, factories, clients, driverQueries, scope, active, expired, execute, flush }; +} + +type Capture = ReturnType; +let environmentCapture: { key: string; capture: Capture } | undefined; +export function databaseResponseCaptureFromEnv(env: NodeJS.ProcessEnv = process.env): Capture | undefined { + const key = env.PAPERCLIP_DB_RESPONSE_CAPTURE_WINDOW; + const window = responseCaptureWindow(key); + if (!key || !window) return; + if (environmentCapture?.key === key) return environmentCapture.capture; + const capture = createDatabaseResponseCapture(window, event => console.info(JSON.stringify(event))); + environmentCapture = { key, capture }; + return capture; +} + +export function installDatabaseResponseCapture(db: T, schema: Record, capture = databaseResponseCaptureFromEnv()): T { + if (!capture || capture.expired()) return db; + const tables = new Set(Object.values(schema).filter(value => is(value, Table)).map(table => getTableName(table))); + const object = (value: unknown): value is Record => value !== null && typeof value === "object"; + function replace(target: Record, name: string, replacement: (original: Method) => Method) { + const original = target[name]; + if (typeof original !== "function") return; + try { target[name] = replacement(original as Method); } catch {} + } + function tableName(value: unknown) { + try { if (is(value, Table)) { const name = getTableName(value); if (tables.has(name)) return name; } } catch {} + return "other"; + } + function query(value: unknown, metadata: Metadata) { + if (!object(value) || capture!.queries.has(value)) return; + capture!.queries.add(value); + replace(value, "execute", original => function (this: unknown, ...args: unknown[]) { + if (capture!.expired()) return Reflect.apply(original, this, args); + return capture!.scope.run(metadata, () => Reflect.apply(original, this, args)); + }); + replace(value, "prepare", original => function (this: unknown, ...args: unknown[]) { + const prepared = Reflect.apply(original, this, args); + try { query(prepared, metadata); } catch {} + return prepared; + }); + } + function factory(value: unknown, step: string, metadata: Metadata, from = false) { + if (!object(value) || capture!.factories.has(value)) return; + capture!.factories.add(value); + replace(value, step, original => function (this: unknown, ...args: unknown[]) { + const result = Reflect.apply(original, this, args); + try { query(result, { ...metadata, table: from ? tableName(args[0]) : metadata.table }); } catch {} + return result; + }); + } + function driverQuery(value: unknown, metadata: Metadata) { + if (!object(value) || capture!.driverQueries.has(value)) return; + capture!.driverQueries.add(value); + let consumed = false; + replace(value, "then", original => function (this: unknown, ...handlers: unknown[]) { + if (consumed || !capture!.active()) { + consumed = true; + return Reflect.apply(original, this, handlers); + } + consumed = true; + const result = capture!.execute(() => Reflect.apply(original, this, []), metadata) as Promise; + return Reflect.apply(result.then, result, handlers); + }); + for (const method of ["execute", "cursor", "forEach", "readable", "writable"]) { + replace(value, method, original => function (this: unknown, ...args: unknown[]) { + consumed = true; + return Reflect.apply(original, this, args); + }); + } + } + function sqlMetadata(value: unknown): Metadata { + const scoped = capture!.scope.getStore(); + const text = typeof value === "string" ? value : ""; + let table = "other"; + for (const match of text.matchAll(/\b(?:from|update|into)\s+"([A-Za-z0-9_]+)"/gi)) { + if (tables.has(match[1]!)) { table = match[1]!; break; } + } + return scoped ? { ...scoped, table: scoped.table === "other" ? table : scoped.table } + : { operation: text.match(/^\s*(select|insert|update|delete)\b/i)?.[1]?.toLowerCase() ?? "execute", + table, callers: publicDatabaseCallers(new Error().stack) }; + } + function client(value: unknown) { + if ((!object(value) && typeof value !== "function") || capture!.clients.has(value)) return; + capture!.clients.add(value); + const target = value as Record; + replace(target, "unsafe", original => function (this: unknown, ...args: unknown[]) { + const result = Reflect.apply(original, this, args); + if (!capture!.expired()) { try { driverQuery(result, sqlMetadata(args[0])); } catch {} } + return result; + }); + for (const method of ["begin", "savepoint"]) { + replace(target, method, original => function (this: unknown, ...args: unknown[]) { + const wrapped = args.map(arg => typeof arg !== "function" ? arg : function (this: unknown, ...inner: unknown[]) { + try { client(inner[0]); } catch {} + return Reflect.apply(arg, this, inner); + }); + return Reflect.apply(original, this, wrapped); + }); + } + replace(target, "reserve", original => function (this: unknown, ...args: unknown[]) { + return Promise.resolve(Reflect.apply(original, this, args)).then(reserved => { + try { client(reserved); } catch {} + return reserved; + }); + }); + } + function root(value: unknown) { + if (!object(value) || capture!.roots.has(value)) return; + capture!.roots.add(value); + for (const method of ["select", "selectDistinct", "selectDistinctOn", "insert", "update", "delete", "execute", "with"] as const) { + replace(value, method, original => function (this: unknown, ...args: unknown[]) { + const result = Reflect.apply(original, this, args); + if (capture!.expired()) return result; + try { + if (method === "with") { root(result); return result; } + const operation = method.startsWith("select") ? "select" : method; + const metadata = { operation, table: operation === "select" || method === "execute" ? "other" : tableName(args[0]), callers: publicDatabaseCallers(new Error().stack) }; + if (operation === "select") factory(result, "from", metadata, true); + else if (method === "insert") factory(result, "values", metadata); + else if (method === "update") factory(result, "set", metadata); + else query(result, metadata); + } catch {} + return result; + }); + } + replace(value, "transaction", original => function (this: unknown, callback: unknown, ...options: unknown[]) { + if (typeof callback !== "function" || capture!.expired()) return Reflect.apply(original, this, [callback, ...options]); + return Reflect.apply(original, this, [function (this: unknown, ...args: unknown[]) { + try { root(args[0]); } catch {} + return Reflect.apply(callback, this, args); + }, ...options]); + }); + } + try { client((db as Record).$client); root(db); } catch {} + return db; +}