Ekklesia/core/src/modules/database-abi.ts

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<string> = new Set([
  "BEGIN",
  "COMMIT",
  "PREPARE",
  "RELEASE",
  "ROLLBACK",
  "SAVEPOINT",
  "START",
]);

const isolationStatements: ReadonlyMap<unknown, string> = 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<string, unknown>;
  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<string, unknown>;
  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<ModuleDatabaseArrayRow | ModuleDatabaseObjectRow> {
  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<AbiClient> {
  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<void> {
  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<Scope>;
  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<Promise<void>>;
  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<T>(state: TransactionState, promise: Promise<T>): Promise<T> {
  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<void> {
  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<void> {
  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<void> {
  try {
    await state.client.query("ROLLBACK");
  } catch {
    state.healthy = false;
  }
}

type Outcome<T> = { readonly value: T } | { readonly error: unknown };

async function settle<T>(run: () => Promise<T> | T): Promise<Outcome<T>> {
  try {
    return { value: await run() };
  } catch (error) {
    return { error };
  }
}

const moduleByExecutor = new WeakMap<object, ModuleId>();

/** 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<ModuleDatabaseQueryResult<ModuleDatabaseArrayRow | ModuleDatabaseObjectRow>> {
    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<ModuleDatabaseQueryResult<ModuleDatabaseArrayRow | ModuleDatabaseObjectRow>> {
      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<T>(
      callback: (savepoint: ModuleDatabaseTransaction) => Promise<T> | T,
    ): Promise<T> {
      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<T>(
    callback: (transaction: ModuleDatabaseTransaction) => Promise<T> | T,
    options?: unknown,
  ): Promise<T> {
    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;
}