import { ModuleDatabaseError, type ModuleDatabaseExecutor, type ModuleDatabaseParameter, type ModuleDatabaseQuery, type ModuleDatabaseTransaction, type ModuleDatabaseValue, } from "@ekklesia/module-database"; /** * A Module Database ABI v1 executor double (platform ADR 0050), for proving Signage's production * store without PostgreSQL. * * It holds Signage to the rules the platform's real executor enforces, so a store change that would * break against Core breaks here first: * * - parameters must be ABI values — a `Date` or a plain object is refused, exactly as Core refuses * one, so an instant or a JSON payload the adapter failed to encode cannot slip through; * - transactions and savepoints are callback scopes: committed only when the callback resolves with * no work outstanding, rolled back when it rejects, and revoked once it settles; * - only the innermost open scope may run statements; * - after a failed statement the transaction is aborted, as in PostgreSQL, and cannot commit. * * Results are what Core sends: positional rows of the ABI's raw forms (`int8`, timestamps, and JSON * as text). A responder answers each statement with rows keyed by column name; the double orders * them by the statement's own select or returning list. */ export interface RecordedStatement { readonly text: string; readonly values: readonly ModuleDatabaseParameter[]; readonly rowMode: string; /** 0 for an autocommit statement, 1 inside a transaction, 2 or more inside savepoints. */ readonly depth: number; /** `select signage_timelines`, `insert signage_channels`, and so on. */ readonly summary: string; } export type AbiRow = Readonly>; export type AbiResponder = (statement: RecordedStatement) => readonly AbiRow[] | undefined; export interface FakeAbiExecutor { readonly executor: ModuleDatabaseExecutor; /** Every statement, in order, plus `BEGIN`, `COMMIT`, `ROLLBACK`, `SAVEPOINT`, `RELEASE`, and * `ROLLBACK TO SAVEPOINT` markers. */ readonly events: string[]; readonly statements: RecordedStatement[]; } function isAbiParameter(value: unknown): value is ModuleDatabaseParameter { if (value === null) { return true; } if (Array.isArray(value)) { return value.every(isAbiParameter); } return ( typeof value === "boolean" || typeof value === "number" || typeof value === "bigint" || typeof value === "string" || value instanceof Uint8Array ); } function summarize(text: string): string { const verb = text.trimStart().split(/\s/u, 1)[0]?.toLowerCase() ?? ""; const target = verb === "select" ? /\bfrom\s+"([^"]+)"/iu.exec(text) : verb === "insert" ? /^\s*insert\s+into\s+"([^"]+)"/iu.exec(text) : verb === "update" ? /^\s*update\s+"([^"]+)"/iu.exec(text) : /^\s*delete\s+from\s+"([^"]+)"/iu.exec(text); return `${verb} ${target?.[1] ?? "?"}`; } /** The column names a statement returns, in order: its select list, or its `returning` list. */ function resultColumns(text: string): string[] { const list = /^\s*select\s+(.*?)\s+from\s/isu.exec(text)?.[1] ?? /\sreturning\s+(.*)$/isu.exec(text)?.[1]; if (list === undefined) { return []; } return list.split(",").map((column) => { const names = [...column.matchAll(/"([^"]+)"/gu)].map((match) => match[1] ?? ""); return names.at(-1) ?? column.trim(); }); } interface Scope { closed: boolean; active: boolean; pending: number; aborted: boolean; } export function createFakeAbiExecutor(respond: AbiResponder = () => []): FakeAbiExecutor { const events: string[] = []; const statements: RecordedStatement[] = []; async function run( scope: Scope, transaction: Scope | undefined, depth: number, request: ModuleDatabaseQuery, ) { if (scope.closed) { throw new ModuleDatabaseError("MODULE_DATABASE_SCOPE_CLOSED"); } if (!scope.active) { throw new ModuleDatabaseError("MODULE_DATABASE_SCOPE_INACTIVE"); } const values = request.values ?? []; if (!values.every(isAbiParameter)) { throw new ModuleDatabaseError("MODULE_DATABASE_PARAMETER_INVALID"); } if (transaction?.aborted) { // PostgreSQL's in_failed_sql_transaction. throw new ModuleDatabaseError("MODULE_DATABASE_QUERY_FAILED", { sqlState: "25P02" }); } const statement: RecordedStatement = { depth, rowMode: request.rowMode ?? "object", summary: summarize(request.text), text: request.text, values, }; statements.push(statement); events.push(statement.summary); scope.pending += 1; try { // Settle on a later event-loop turn, as a real round trip does, so a statement a callback // forgot to await is still outstanding when that callback returns. await new Promise((resolve) => setImmediate(resolve)); let rows: readonly AbiRow[]; try { rows = respond(statement) ?? []; } catch (error) { if (transaction) { transaction.aborted = true; } throw error; } const columns = resultColumns(request.text); return { command: statement.summary.split(" ")[0]?.toUpperCase() ?? "", fields: columns.map((name) => ({ name, typeOid: 25 })), rowCount: rows.length, rows: statement.rowMode === "array" ? rows.map((row) => columns.map((column) => row[column] ?? null)) : rows, }; } finally { scope.pending -= 1; } } async function runScope( parent: Scope, transaction: Scope | undefined, depth: number, callback: (scope: ModuleDatabaseTransaction) => Promise | T, ): Promise { const isTransaction = transaction === undefined; const scope: Scope = { aborted: false, active: true, closed: false, pending: 0 }; const owner = transaction ?? scope; const executor = Object.freeze({ abiVersion: 1 as const, query: (request: ModuleDatabaseQuery) => run(scope, owner, depth, request), savepoint: (inner: (savepoint: ModuleDatabaseTransaction) => Promise | S) => { if (scope.closed) { return Promise.reject(new ModuleDatabaseError("MODULE_DATABASE_SCOPE_CLOSED")); } if (!scope.active) { return Promise.reject(new ModuleDatabaseError("MODULE_DATABASE_SCOPE_INACTIVE")); } return runScope(scope, owner, depth + 1, inner); }, }) as unknown as ModuleDatabaseTransaction; parent.active = false; events.push(isTransaction ? "BEGIN" : "SAVEPOINT"); const abort = isTransaction ? "ROLLBACK" : "ROLLBACK TO SAVEPOINT"; try { let value: T; try { value = await callback(executor); } catch (error) { events.push(abort); throw error; } if (scope.pending > 0) { events.push(abort); throw new ModuleDatabaseError("MODULE_DATABASE_SCOPE_WORK_OUTSTANDING"); } if (isTransaction && owner.aborted) { events.push(abort); throw new ModuleDatabaseError("MODULE_DATABASE_TRANSACTION_FAILED", { sqlState: "25P02" }); } events.push(isTransaction ? "COMMIT" : "RELEASE"); return value; } finally { if (!isTransaction && events.at(-1) === abort) { // A savepoint that rolled back leaves its transaction usable, as in PostgreSQL. owner.aborted = false; } scope.closed = true; parent.active = true; } } const root: Scope = { aborted: false, active: true, closed: false, pending: 0 }; const executor = Object.freeze({ abiVersion: 1 as const, query: (request: ModuleDatabaseQuery) => run(root, undefined, 0, request), transaction: (callback: (transaction: ModuleDatabaseTransaction) => Promise | T) => { // Top-level statements are autocommit and may run on other pooled connections, so the root // executor stays usable while a transaction is open. const detached: Scope = { aborted: false, active: true, closed: false, pending: 0 }; return runScope(detached, undefined, 1, callback); }, }) as unknown as ModuleDatabaseExecutor; return { events, executor, statements }; }