import { MODULE_DATABASE_ABI_VERSION, type ModuleDatabaseArrayRow, ModuleDatabaseError, type ModuleDatabaseErrorCode, type ModuleDatabaseExecutor, type ModuleDatabaseField, type ModuleDatabaseObjectRow, type ModuleDatabaseQueryable, type ModuleDatabaseQueryResult, type ModuleDatabaseTransaction, type ModuleId, moduleIdSchema, } from "@ekklesia/contracts"; import { DatabaseError, type Pool, type PoolClient, type QueryArrayConfig } from "pg"; import { decodeModuleDatabaseRow, encodeModuleDatabaseParameters, rawTextTypes, } from "./database-abi-values.js"; /** * Core's Module Database ABI v1 executor (ADR 0050). * * A module hands Core SQL text, ordered parameter values, and a row mode; Core hands back decoded * rows, column metadata, and stable errors. No ORM, driver, pool, client, or connection object ever * crosses in either direction. The executor is built over the one pool the API already owns * (ADR 0016) and never closes it. * * This is a correctness and compatibility boundary, **not a security sandbox**: a trusted module can * still send any SQL its database role permits. Nothing here inspects table names. * * ### Connections and session state * * Every ABI operation, a top-level statement or a whole transaction, borrows a connection from the * pool, pins the session settings the ABI's decoding depends on, does its work, and resets the * session with `DISCARD ALL` before returning the connection. So no setting, temporary table, * listener, advisory lock, prepared statement, or role change a module makes survives the operation * that made it, and nothing the ABI pinned reaches the pool's other users. A connection that cannot * be reset cleanly, or that is not idle when the operation ends, is destroyed rather than returned. * `DISCARD ALL` also deallocates named prepared statements; Core uses none on the shared pool. * * ### One statement per `query` * * Every ABI statement is sent with PostgreSQL's extended query protocol, forced with the driver's * `queryMode: "extended"` even when there are no parameters. PostgreSQL itself refuses a Parse * message holding more than one statement (SQLSTATE 42601), using its own SQL grammar, so comments, * quoted literals, and semicolons inside them are never mistaken for a second statement. Core does * not parse SQL. A statement text that holds no statement at all is refused after PostgreSQL reports * it empty. * * ### Transaction control belongs to the callbacks * * Transaction-control SQL (`BEGIN`, `START TRANSACTION`, `COMMIT`, `ROLLBACK`, `SAVEPOINT`, * `RELEASE SAVEPOINT`, `PREPARE TRANSACTION`, and their aliases) is forbidden through `query`: a * conformance rule for trusted module code, not a security guarantee. Core cannot refuse such a * statement beforehand without parsing SQL, which it does not do, and it cannot undo one PostgreSQL * has already executed: a `COMMIT` sent through a transaction's `query` has committed by the time * Core sees it. What Core does is detect the violation afterwards, from what PostgreSQL itself * reports (the statement's command tag, and the protocol's transaction status), and fail safely: * the call rejects, an enclosing transaction is never committed by Core and rejects too, and the * connection is destroyed rather than returned to the pool. The commit and rollback guarantees of * the transaction API apply to conforming use. */ /** The settings the ABI's decoding and determinism depend on, pinned for every ABI operation. */ const pinSessionStatement = "SET TimeZone TO 'UTC'; SET DateStyle TO 'ISO, MDY'; SET IntervalStyle TO 'postgres'; " + "SET bytea_output TO 'hex'; SET extra_float_digits TO 1"; const resetSessionStatement = "DISCARD ALL"; const transactionControlCommands: ReadonlySet = new Set([ "BEGIN", "COMMIT", "PREPARE", "RELEASE", "ROLLBACK", "SAVEPOINT", "START", ]); const isolationStatements: ReadonlyMap = new Map([ ["read committed", "READ COMMITTED"], ["repeatable read", "REPEATABLE READ"], ["serializable", "SERIALIZABLE"], ]); /** `pg` 8.22 reports the protocol's ReadyForQuery transaction status through this method, which * its type declarations omit. `acquire` refuses a client without it rather than guessing. */ type AbiClient = PoolClient & { getTransactionStatus(): string | null }; interface PreparedQuery { readonly config: QueryArrayConfig; readonly rowMode: "array" | "object"; } interface RawResult { readonly command: string | null; readonly fields: readonly { readonly dataTypeID: number; readonly name: string }[]; readonly rowCount: number | null; readonly rows: readonly unknown[][]; } function databaseError(code: ModuleDatabaseErrorCode, cause?: unknown): ModuleDatabaseError { // Only PostgreSQL's own SQLSTATE survives. The driver's message, detail, hint, position, and // object names, and every non-PostgreSQL error, are dropped here. if (cause instanceof DatabaseError && typeof cause.code === "string") { return new ModuleDatabaseError(code, { sqlState: cause.code }); } return new ModuleDatabaseError(code); } /** A PostgreSQL error leaves the connection usable; anything else (a dropped connection, a * client-side timeout) may leave it mid-protocol, so it must be destroyed. */ function connectionSurvives(error: unknown): boolean { return error instanceof DatabaseError; } function prepareQuery(request: unknown): PreparedQuery { if (typeof request !== "object" || request === null || Array.isArray(request)) { throw new ModuleDatabaseError("MODULE_DATABASE_REQUEST_INVALID"); } for (const key of Object.keys(request)) { if (key !== "text" && key !== "values" && key !== "rowMode") { throw new ModuleDatabaseError("MODULE_DATABASE_REQUEST_INVALID"); } } const { rowMode, text, values } = request as Record; if (typeof text !== "string" || text.length === 0) { throw new ModuleDatabaseError("MODULE_DATABASE_REQUEST_INVALID"); } if (rowMode !== undefined && rowMode !== "object" && rowMode !== "array") { throw new ModuleDatabaseError("MODULE_DATABASE_REQUEST_INVALID"); } const config = { queryMode: "extended", rowMode: "array", text, types: rawTextTypes, values: encodeModuleDatabaseParameters(values), }; return { config: config as unknown as QueryArrayConfig, rowMode: rowMode ?? "object" }; } function beginStatement(options: unknown): string { if (options === undefined) { return "BEGIN ISOLATION LEVEL READ COMMITTED READ WRITE"; } if (typeof options !== "object" || options === null || Array.isArray(options)) { throw new ModuleDatabaseError("MODULE_DATABASE_TRANSACTION_OPTION_INVALID"); } for (const key of Object.keys(options)) { if (key !== "isolation" && key !== "readOnly") { throw new ModuleDatabaseError("MODULE_DATABASE_TRANSACTION_OPTION_INVALID"); } } const { isolation, readOnly } = options as Record; const level = isolation === undefined ? "READ COMMITTED" : isolationStatements.get(isolation); if (level === undefined || (readOnly !== undefined && typeof readOnly !== "boolean")) { throw new ModuleDatabaseError("MODULE_DATABASE_TRANSACTION_OPTION_INVALID"); } return `BEGIN ISOLATION LEVEL ${level} ${readOnly === true ? "READ ONLY" : "READ WRITE"}`; } function buildResult( raw: RawResult, rowMode: "array" | "object", ): ModuleDatabaseQueryResult { if (typeof raw.command !== "string" || raw.command === "") { // PostgreSQL answered with EmptyQueryResponse: the text held no statement. throw new ModuleDatabaseError("MODULE_DATABASE_REQUEST_INVALID"); } const fields: readonly ModuleDatabaseField[] = Object.freeze( raw.fields.map((field) => Object.freeze({ name: field.name, typeOid: field.dataTypeID })), ); const decoded = raw.rows.map((row) => decodeModuleDatabaseRow(fields, row)); let rows: readonly (ModuleDatabaseArrayRow | ModuleDatabaseObjectRow)[]; if (rowMode === "array") { rows = decoded; } else { if (new Set(fields.map((field) => field.name)).size !== fields.length) { // Object rows cannot hold two columns with one name; refusing beats dropping one. throw new ModuleDatabaseError("MODULE_DATABASE_REQUEST_INVALID"); } // `Object.fromEntries` defines properties, so a column named `__proto__` stays a column. rows = decoded.map((row) => Object.fromEntries(fields.map((field, index) => [field.name, row[index] ?? null])), ); } return Object.freeze({ command: raw.command, fields, rowCount: raw.rowCount ?? null, rows: Object.freeze(rows), }); } /** A borrowed client's `error` events are already reported through the statement they interrupt. * The pool only listens to idle clients, so without this a connection the server terminates while * borrowed would surface as an unhandled `error` event. */ function ignoreClientError(): void {} function destroy(client: AbiClient): void { client.removeListener("error", ignoreClientError); client.release(true); } async function acquire(pool: Pool, failure: ModuleDatabaseErrorCode): Promise { let client: AbiClient; try { client = (await pool.connect()) as AbiClient; } catch (error) { throw databaseError(failure, error); } client.on("error", ignoreClientError); if (typeof client.getTransactionStatus !== "function") { destroy(client); throw new ModuleDatabaseError(failure); } try { await client.query(pinSessionStatement); } catch (error) { destroy(client); throw databaseError(failure, error); } return client; } /** Returns a connection to the pool only when it is idle and resets cleanly; destroys it * otherwise. Never throws. */ async function relinquish(client: AbiClient, healthy: boolean): Promise { if (!healthy || client.getTransactionStatus() !== "I") { destroy(client); return; } try { await client.query(resetSessionStatement); } catch { destroy(client); return; } client.removeListener("error", ignoreClientError); client.release(); } /** One open scope: a transaction or a savepoint inside it. */ interface Scope { activeChild: Scope | undefined; readonly children: Set; closed: boolean; /** Operations started through this scope's executor that have not settled yet. */ pending: number; } interface TransactionState { readonly client: AbiClient; healthy: boolean; /** Set when the transaction can no longer be committed: transaction control sent through * `query`, or a savepoint that could not be undone. */ failure: ModuleDatabaseError | undefined; finished: boolean; /** Settled-never-rejecting shadows of every statement sent on the connection. */ readonly inflight: Set>; savepoints: number; } function newScope(parent?: Scope): Scope { const scope: Scope = { activeChild: undefined, children: new Set(), closed: false, pending: 0 }; parent?.children.add(scope); return scope; } function closeScope(scope: Scope): void { scope.closed = true; for (const child of scope.children) { closeScope(child); } } function assertActive(scope: Scope): void { if (scope.closed) { throw new ModuleDatabaseError("MODULE_DATABASE_SCOPE_CLOSED"); } if (scope.activeChild !== undefined) { throw new ModuleDatabaseError("MODULE_DATABASE_SCOPE_INACTIVE"); } } /** Once a transaction can no longer be committed, its scopes send nothing more. After a * transaction-control violation in particular, a later statement would otherwise run outside any * transaction, as its own autocommitted statement. */ function assertUsable(state: TransactionState): void { if (state.failure !== undefined) { throw new ModuleDatabaseError(state.failure.code); } } function track(state: TransactionState, promise: Promise): Promise { const shadow = promise.then( () => undefined, () => undefined, ); state.inflight.add(shadow); void shadow.then(() => state.inflight.delete(shadow)); return promise; } /** Waits until every statement already sent has settled, including any sent while waiting. */ async function drain(state: TransactionState): Promise { while (state.inflight.size > 0) { await Promise.all([...state.inflight]); } } /** Runs one of Core's own transaction-control statements. Refuses, without sending anything, once * the transaction has finished. */ async function control(state: TransactionState, statement: string): Promise { if (state.finished) { throw new ModuleDatabaseError("MODULE_DATABASE_SCOPE_CLOSED"); } try { await track(state, state.client.query(statement)); } catch (error) { if (!connectionSurvives(error)) { state.healthy = false; } throw databaseError("MODULE_DATABASE_TRANSACTION_FAILED", error); } } async function rollback(state: TransactionState): Promise { try { await state.client.query("ROLLBACK"); } catch { state.healthy = false; } } type Outcome = { readonly value: T } | { readonly error: unknown }; async function settle(run: () => Promise | T): Promise> { try { return { value: await run() }; } catch (error) { return { error }; } } const moduleByExecutor = new WeakMap(); /** Which module an ABI executor, or a transaction or savepoint executor derived from it, belongs * to. For Core's own logging, diagnostics, and attribution; never exposed to modules. */ export function moduleDatabaseExecutorModuleId(executor: object): ModuleId | undefined { return moduleByExecutor.get(executor); } export interface ModuleDatabaseExecutorOptions { /** The module this executor serves. Carried for attribution only: the ABI enforces no table * ownership at run time (ADR 0050). */ readonly moduleId: ModuleId; } /** * Creates one module's ABI v1 executor over the API's existing pool. Synchronous and free of I/O: * nothing touches the database until the module calls it. The returned object, and every * transaction and savepoint executor it hands out, is frozen and holds exactly the ABI's members. */ export function createModuleDatabaseExecutor( pool: Pool, options: ModuleDatabaseExecutorOptions, ): ModuleDatabaseExecutor { const parsedModuleId = moduleIdSchema.safeParse(options.moduleId); if (!parsedModuleId.success) { throw new TypeError("A module database executor needs a valid module ID."); } const moduleId = parsedModuleId.data; async function query( request: unknown, ): Promise> { const prepared = prepareQuery(request); const client = await acquire(pool, "MODULE_DATABASE_QUERY_FAILED"); let healthy = true; try { let raw: RawResult; try { raw = (await client.query(prepared.config)) as unknown as RawResult; } catch (error) { healthy = connectionSurvives(error); throw databaseError("MODULE_DATABASE_QUERY_FAILED", error); } if ( (raw.command !== null && transactionControlCommands.has(raw.command)) || client.getTransactionStatus() !== "I" ) { // A transaction-control violation. PostgreSQL has already run it; the connection is // destroyed below rather than returned in a state Core did not choose. healthy = false; throw new ModuleDatabaseError("MODULE_DATABASE_REQUEST_INVALID"); } return buildResult(raw, prepared.rowMode); } finally { await relinquish(client, healthy); } } function scopeExecutor(state: TransactionState, scope: Scope): ModuleDatabaseTransaction { async function scopeQuery( request: unknown, ): Promise> { assertActive(scope); assertUsable(state); const prepared = prepareQuery(request); scope.pending += 1; try { let raw: RawResult; try { raw = (await track(state, state.client.query(prepared.config))) as unknown as RawResult; } catch (error) { if (!connectionSurvives(error)) { state.healthy = false; } throw databaseError("MODULE_DATABASE_QUERY_FAILED", error); } if ( (raw.command !== null && transactionControlCommands.has(raw.command)) || state.client.getTransactionStatus() === "I" ) { // A transaction-control violation. PostgreSQL has already run it, and Core cannot undo // that; it can only never commit this transaction itself, and destroy the connection. state.failure ??= new ModuleDatabaseError("MODULE_DATABASE_REQUEST_INVALID"); state.healthy = false; throw new ModuleDatabaseError("MODULE_DATABASE_REQUEST_INVALID"); } return buildResult(raw, prepared.rowMode); } finally { scope.pending -= 1; } } async function savepoint( callback: (savepoint: ModuleDatabaseTransaction) => Promise | T, ): Promise { assertActive(scope); assertUsable(state); if (typeof callback !== "function") { throw new ModuleDatabaseError("MODULE_DATABASE_REQUEST_INVALID"); } scope.pending += 1; const child = newScope(scope); scope.activeChild = child; state.savepoints += 1; // Core's own identifier: a fixed prefix and a counter, never anything a module supplied. const name = `ekklesia_savepoint_${state.savepoints}`; try { await control(state, `SAVEPOINT ${name}`); if (child.closed) { throw new ModuleDatabaseError("MODULE_DATABASE_SCOPE_CLOSED"); } const outcome = await settle(() => callback(scopeExecutor(state, child))); const outstanding = child.pending > 0; closeScope(child); await drain(state); if (scope.closed) { // The enclosing transaction finished first; it rolls back, and nothing more is sent. throw new ModuleDatabaseError("MODULE_DATABASE_SCOPE_CLOSED"); } if ("error" in outcome) { // A rejected callback's own error stays the error, even if sibling work it started was // still settling: that work has been drained, and is undone here with everything else. try { await control(state, `ROLLBACK TO SAVEPOINT ${name}`); await control(state, `RELEASE SAVEPOINT ${name}`); } catch (error) { state.failure ??= new ModuleDatabaseError("MODULE_DATABASE_TRANSACTION_FAILED"); throw error; } throw outcome.error; } if (outstanding) { // A callback that resolved with work outstanding forgot an await. Roll back to the // savepoint and never release it. try { await control(state, `ROLLBACK TO SAVEPOINT ${name}`); } catch (error) { state.failure ??= new ModuleDatabaseError("MODULE_DATABASE_TRANSACTION_FAILED"); throw error; } throw new ModuleDatabaseError("MODULE_DATABASE_SCOPE_WORK_OUTSTANDING"); } try { await control(state, `RELEASE SAVEPOINT ${name}`); } catch (releaseError) { // Typically a failed statement left the transaction aborted. Undo the savepoint so the // transaction stays usable, and report that this savepoint did not complete. try { await control(state, `ROLLBACK TO SAVEPOINT ${name}`); await control(state, `RELEASE SAVEPOINT ${name}`); } catch { state.failure ??= new ModuleDatabaseError("MODULE_DATABASE_TRANSACTION_FAILED"); } throw releaseError; } return (outcome as { readonly value: T }).value; } finally { closeScope(child); if (scope.activeChild === child) { scope.activeChild = undefined; } scope.pending -= 1; } } const executor = Object.freeze({ abiVersion: MODULE_DATABASE_ABI_VERSION, query: scopeQuery as ModuleDatabaseQueryable["query"], savepoint, }); moduleByExecutor.set(executor, moduleId); return executor; } async function transaction( callback: (transaction: ModuleDatabaseTransaction) => Promise | T, options?: unknown, ): Promise { if (typeof callback !== "function") { throw new ModuleDatabaseError("MODULE_DATABASE_REQUEST_INVALID"); } const begin = beginStatement(options); const client = await acquire(pool, "MODULE_DATABASE_TRANSACTION_FAILED"); const state: TransactionState = { client, failure: undefined, finished: false, healthy: true, inflight: new Set(), savepoints: 0, }; try { try { await client.query(begin); } catch (error) { state.healthy = connectionSurvives(error); throw databaseError("MODULE_DATABASE_TRANSACTION_FAILED", error); } const scope = newScope(); const outcome = await settle(() => callback(scopeExecutor(state, scope))); const outstanding = scope.pending > 0; closeScope(scope); await drain(state); state.finished = true; if ("error" in outcome) { // The callback's own error stays primary, even when sibling work it started (say, the // other half of a `Promise.all`) was still settling. That work has been drained, and rolls // back with the rest; outstanding work never leads to a commit. await rollback(state); throw outcome.error; } if (outstanding) { // The callback resolved with work it never awaited: a forgotten `await`. await rollback(state); throw new ModuleDatabaseError("MODULE_DATABASE_SCOPE_WORK_OUTSTANDING"); } if (state.failure !== undefined) { await rollback(state); throw state.failure; } let committed: { readonly command: string | null }; try { committed = await client.query("COMMIT"); } catch (error) { state.healthy = connectionSurvives(error); throw databaseError("MODULE_DATABASE_TRANSACTION_FAILED", error); } if (committed.command !== "COMMIT") { // PostgreSQL answers COMMIT with ROLLBACK when a failed statement had aborted the // transaction. Nothing was committed, and PostgreSQL supplied no SQLSTATE to report. throw new ModuleDatabaseError("MODULE_DATABASE_TRANSACTION_FAILED"); } return outcome.value; } finally { state.finished = true; await relinquish(client, state.healthy); } } const executor = Object.freeze({ abiVersion: MODULE_DATABASE_ABI_VERSION, query: query as ModuleDatabaseQueryable["query"], transaction: transaction as ModuleDatabaseExecutor["transaction"], }); moduleByExecutor.set(executor, moduleId); return executor; }