Ekklesia/core/src/migrations/postgres-database.ts

import { Client, type ClientConfig, type QueryResultRow } from "pg";
import {
  type DatabaseTransportPolicy,
  databaseTlsConnectionOptions,
} from "../database/transport.js";
import type {
  AppliedMigrationRecord,
  MigrationDatabase,
  MigrationDatabaseSession,
  MigrationHistoryInsert,
  MigrationTransaction,
} from "./runner.js";

const historyTableSchema = "public";
const historyTableName = "ekklesia_core_migrations";
const advisoryLockNamespace = 1_162_562_380;
const advisoryLockKey = 1_296_648_018;
const advisoryLockPollIntervalMs = 50;
const maximumPostgresTimeoutMs = 2_147_483_647;

const createHistoryTableSql = `
  CREATE TABLE IF NOT EXISTS "public"."ekklesia_core_migrations" (
    "applied_order" bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    "identity" text NOT NULL UNIQUE,
    "module_id" text NOT NULL,
    "migration_tag" text NOT NULL,
    "migration_index" integer NOT NULL,
    "checksum" text NOT NULL,
    "applied_at" timestamptz NOT NULL DEFAULT clock_timestamp(),
    UNIQUE ("module_id", "migration_tag"),
    UNIQUE ("module_id", "migration_index")
  )
`;

const readHistorySql = `
  SELECT
    "applied_order",
    "identity",
    "module_id",
    "migration_tag",
    "migration_index",
    "checksum",
    "applied_at"
  FROM "public"."ekklesia_core_migrations"
  ORDER BY "applied_order" ASC
`;

const insertHistorySql = `
  INSERT INTO "public"."ekklesia_core_migrations" (
    "identity",
    "module_id",
    "migration_tag",
    "migration_index",
    "checksum"
  )
  VALUES ($1, $2, $3, $4, $5)
`;

export type PostgresMigrationDatabaseErrorCode =
  | "POSTGRES_MIGRATION_CONFIGURATION_INVALID"
  | "POSTGRES_MIGRATION_CONNECTION_FAILED"
  | "POSTGRES_MIGRATION_EXECUTION_FAILED"
  | "POSTGRES_MIGRATION_HISTORY_READ_FAILED"
  | "POSTGRES_MIGRATION_HISTORY_WRITE_FAILED"
  | "POSTGRES_MIGRATION_LOCK_FAILED"
  | "POSTGRES_MIGRATION_SESSION_CLOSE_FAILED"
  | "POSTGRES_MIGRATION_TRANSACTION_FAILED";

export class PostgresMigrationDatabaseError extends Error {
  readonly code: PostgresMigrationDatabaseErrorCode;

  constructor(code: PostgresMigrationDatabaseErrorCode, message: string) {
    super(message);
    this.name = "PostgresMigrationDatabaseError";
    this.code = code;
    this.stack = `${this.name}: ${this.message}`;
  }
}

/**
 * The platform's fixed migration connection deadline, applied when a caller does not narrow it.
 * Lives here, at the one construction boundary every migration command shares, so a future CLI
 * call site cannot silently omit it.
 */
export const MIGRATION_CONNECTION_TIMEOUT_MS = 10_000;

/**
 * The platform's fixed migration operation deadline: both `pg`'s client-side `query_timeout` and
 * PostgreSQL's own server-side `statement_timeout`. The client-side value bounds a query whose
 * response never arrives; the server-side value bounds work the server is actually doing. Applied
 * to every statement the session issues — history reads, advisory-lock queries, transaction
 * control, migration SQL, and history writes.
 */
export const MIGRATION_OPERATION_TIMEOUT_MS = 10_000;

/**
 * Connection-string parameters that must not reach the pinned driver, established by measuring the
 * installed `[email protected]`/`[email protected]` against a real PostgreSQL and a real
 * blackhole listener rather than by reading the merge and assuming.
 *
 * `ConnectionParameters` merges as `Object.assign({}, config, parse(config.connectionString))` and
 * `pg-connection-string` retains *arbitrary* query keys, so any of these lands in the driver's
 * configuration. What each one actually does differs, and only the first genuinely weakens policy:
 *
 * - `statement_timeout` — **wins outright.** `?statement_timeout=0` produced a server-reported
 *   `statement_timeout` of `0` (disabled) and `?statement_timeout=900000` produced `15min`, against
 *   an explicit `statement_timeout: 10000`. This is the one real downgrade.
 * - `query_timeout` — overrides the explicit field, but does not weaken the bound; it breaks the
 *   client. `client.query` computes its read timer as
 *   `queryConfig.query_timeout || connectionParameters.query_timeout`, and the parsed URL value is
 *   the *string* `"0"` — truthy — so every query gets a 0 ms read timer and fails
 *   nondeterministically with `Query read timeout`. Rejected so an operator gets a fixed
 *   configuration error instead of intermittent, unexplained query failures.
 * - `connectionTimeoutMillis` — pollutes `ConnectionParameters.connect_timeout` (10 → `0` or `900`
 *   seconds), but that field is only consumed by `getLibpqConnectionString()`, which the pure-JS
 *   client never calls. Measured against a blackhole listener, the effective connect timer is
 *   **unchanged**: an explicit 500 ms still expired at ~500 ms with `?connectionTimeoutMillis=0`
 *   and with `=900000`. Rejected as unsupported and as defense in depth — no legitimate
 *   `DATABASE_URL` carries a driver option, and the polluted value would take effect on the native
 *   binding — not because it currently defeats the deadline.
 * - `connect_timeout` — no effect at all: `ConnectionParameters` recomputes it from
 *   `connectionTimeoutMillis`, so `?connect_timeout=0` left it at the derived `10`. Rejected as
 *   unsupported; earlier revisions of this comment wrongly claimed it overrode the explicit field.
 * - `options` carrying `statement_timeout` — measured **not** to override the named startup
 *   parameter: `?options=-c statement_timeout=0`, `=900s`, and a backslash-escaped spelling all
 *   still produced `10s`, as did an inherited `PGOPTIONS`. Rejected as unsupported rather than
 *   because it wins.
 *
 * Deliberately narrow: unrelated parameters (`application_name`, and anything else that does not
 * name a timeout this boundary selects) are untouched. `PGCONNECT_TIMEOUT` needs no handling — the
 * driver consults it only when `connectionTimeoutMillis` is `undefined`, and this boundary always
 * sets it; an inherited `PGCONNECT_TIMEOUT=0` was measured to change nothing.
 */
const CONFLICTING_TIMEOUT_PARAMETERS = Object.freeze([
  "connect_timeout",
  "connectiontimeoutmillis",
  "query_timeout",
  "statement_timeout",
]);

function assertNoConflictingTimeoutParameters(connectionString: string): void {
  let search: URLSearchParams;
  try {
    search = new URL(connectionString).searchParams;
  } catch {
    throw databaseError(
      "POSTGRES_MIGRATION_CONFIGURATION_INVALID",
      "The PostgreSQL database URL is invalid.",
    );
  }
  for (const [rawKey, rawValue] of search.entries()) {
    const key = rawKey.toLowerCase();
    if (CONFLICTING_TIMEOUT_PARAMETERS.includes(key)) {
      throw databaseError(
        "POSTGRES_MIGRATION_CONFIGURATION_INVALID",
        "The PostgreSQL database URL must not set a connection or statement timeout.",
      );
    }
    if (key === "options" && rawValue.toLowerCase().includes("statement_timeout")) {
      throw databaseError(
        "POSTGRES_MIGRATION_CONFIGURATION_INVALID",
        "The PostgreSQL database URL must not set a connection or statement timeout.",
      );
    }
  }
}

export interface PostgresMigrationDatabaseOptions {
  /** Omitted means `MIGRATION_CONNECTION_TIMEOUT_MS`. Tests may narrow it; nothing disables it. */
  readonly connectionTimeoutMs?: number;
  /** Omitted means `MIGRATION_OPERATION_TIMEOUT_MS`. Tests may narrow it; nothing disables it. */
  readonly operationTimeoutMs?: number;
  /** The validated transport policy (`core/src/database/transport.js`). Every production executable
   * that plans or applies migrations passes one; omitting it keeps the pre-existing plaintext
   * behavior used by the disposable loopback PostgreSQL harness. */
  readonly transport?: DatabaseTransportPolicy;
}

interface HistoryExistenceRow extends QueryResultRow {
  readonly relation_kind: string;
}

interface HistoryRow extends QueryResultRow {
  readonly applied_at: Date | string;
  readonly applied_order: number | string;
  readonly checksum: string;
  readonly identity: string;
  readonly migration_index: number;
  readonly migration_tag: string;
  readonly module_id: string;
}

interface AdvisoryLockRow extends QueryResultRow {
  readonly acquired: boolean;
}

interface AdvisoryUnlockRow extends QueryResultRow {
  readonly released: boolean;
}

function databaseError(
  code: PostgresMigrationDatabaseErrorCode,
  message: string,
): PostgresMigrationDatabaseError {
  return new PostgresMigrationDatabaseError(code, message);
}

function validateConnectionString(connectionString: string | undefined): string {
  if (typeof connectionString !== "string" || connectionString.trim() === "") {
    throw databaseError(
      "POSTGRES_MIGRATION_CONFIGURATION_INVALID",
      "A PostgreSQL database URL is required.",
    );
  }

  try {
    const url = new URL(connectionString);
    if (url.protocol !== "postgres:" && url.protocol !== "postgresql:") {
      throw new Error("unsupported protocol");
    }
  } catch {
    throw databaseError(
      "POSTGRES_MIGRATION_CONFIGURATION_INVALID",
      "The PostgreSQL database URL is invalid.",
    );
  }

  return connectionString;
}

function invalidConfiguration(message: string): PostgresMigrationDatabaseError {
  return databaseError("POSTGRES_MIGRATION_CONFIGURATION_INVALID", message);
}

function validateTimeout(value: unknown, message: string): number | undefined {
  if (value === undefined) {
    return undefined;
  }
  if (
    typeof value !== "number" ||
    !Number.isSafeInteger(value) ||
    value <= 0 ||
    value > maximumPostgresTimeoutMs
  ) {
    throw invalidConfiguration(message);
  }
  return value;
}

function createClientConfig(
  connectionString: string,
  options: PostgresMigrationDatabaseOptions | undefined,
): ClientConfig {
  if (
    options !== undefined &&
    (typeof options !== "object" || options === null || Array.isArray(options))
  ) {
    throw invalidConfiguration("The PostgreSQL migration database options are invalid.");
  }

  let connectionTimeoutMs: unknown;
  let operationTimeoutMs: unknown;
  try {
    connectionTimeoutMs = options?.connectionTimeoutMs;
    operationTimeoutMs = options?.operationTimeoutMs;
  } catch {
    throw invalidConfiguration("The PostgreSQL migration database options are invalid.");
  }

  // Defaults are applied here, at the shared construction boundary, so every migration command
  // inherits a finite deadline without having to remember to ask for one.
  const validatedConnectionTimeoutMs =
    validateTimeout(connectionTimeoutMs, "The PostgreSQL connection timeout is invalid.") ??
    MIGRATION_CONNECTION_TIMEOUT_MS;
  const validatedOperationTimeoutMs =
    validateTimeout(operationTimeoutMs, "The PostgreSQL operation timeout is invalid.") ??
    MIGRATION_OPERATION_TIMEOUT_MS;

  // The URL must not be able to weaken what was just selected.
  assertNoConflictingTimeoutParameters(connectionString);

  // Also rejects a URL carrying its own TLS parameters, so `pg`'s connection-string precedence
  // can never silently replace this policy with a weaker one.
  const ssl =
    options?.transport === undefined
      ? undefined
      : databaseTlsConnectionOptions(connectionString, options.transport);

  return {
    connectionString,
    connectionTimeoutMillis: validatedConnectionTimeoutMs,
    query_timeout: validatedOperationTimeoutMs,
    ...(ssl === undefined ? {} : { ssl }),
    statement_timeout: validatedOperationTimeoutMs,
  };
}

function isPostgresMigrationDatabaseError(error: unknown): error is PostgresMigrationDatabaseError {
  return error instanceof PostgresMigrationDatabaseError;
}

function parseAppliedOrder(value: number | string): number | undefined {
  const parsed = typeof value === "number" ? value : Number(value);
  return Number.isSafeInteger(parsed) && parsed > 0 ? parsed : undefined;
}

function parseAppliedAt(value: Date | string): string | undefined {
  const parsed = value instanceof Date ? value : new Date(value);
  return Number.isNaN(parsed.getTime()) ? undefined : parsed.toISOString();
}

function toAppliedMigrationRecord(row: HistoryRow): AppliedMigrationRecord {
  const appliedOrder = parseAppliedOrder(row.applied_order);
  const appliedAt = parseAppliedAt(row.applied_at);
  if (
    appliedOrder === undefined ||
    appliedAt === undefined ||
    typeof row.identity !== "string" ||
    typeof row.module_id !== "string" ||
    typeof row.migration_tag !== "string" ||
    !Number.isSafeInteger(row.migration_index) ||
    typeof row.checksum !== "string"
  ) {
    throw databaseError(
      "POSTGRES_MIGRATION_HISTORY_READ_FAILED",
      "The Core migration history could not be read.",
    );
  }

  return Object.freeze({
    appliedAt,
    appliedOrder,
    checksum: row.checksum,
    identity: row.identity,
    index: row.migration_index,
    moduleId: row.module_id,
    tag: row.migration_tag,
  });
}

async function pause(milliseconds: number): Promise<void> {
  await new Promise<void>((resolve) => setTimeout(resolve, milliseconds));
}

class PostgresMigrationDatabaseSession implements MigrationDatabaseSession {
  readonly #client: Client;
  #closed = false;
  #lockHeld = false;
  #transactionActive = false;

  constructor(client: Client) {
    this.#client = client;
  }

  async readAppliedMigrations(): Promise<readonly AppliedMigrationRecord[]> {
    this.#assertOpen("POSTGRES_MIGRATION_HISTORY_READ_FAILED");

    try {
      const existence = await this.#client.query<HistoryExistenceRow>(
        `
          SELECT relation.relkind AS relation_kind
          FROM pg_catalog.pg_class AS relation
          INNER JOIN pg_catalog.pg_namespace AS namespace
            ON namespace.oid = relation.relnamespace
          WHERE namespace.nspname = $1 AND relation.relname = $2
        `,
        [historyTableSchema, historyTableName],
      );
      const relationKind = existence.rows[0]?.relation_kind;
      if (relationKind === undefined) {
        return Object.freeze([]);
      }
      if (relationKind !== "r") {
        throw databaseError(
          "POSTGRES_MIGRATION_HISTORY_READ_FAILED",
          "The Core migration history could not be read.",
        );
      }

      const result = await this.#client.query<HistoryRow>(readHistorySql);
      return Object.freeze(result.rows.map(toAppliedMigrationRecord));
    } catch (error) {
      if (isPostgresMigrationDatabaseError(error)) {
        throw error;
      }
      throw databaseError(
        "POSTGRES_MIGRATION_HISTORY_READ_FAILED",
        "The Core migration history could not be read.",
      );
    }
  }

  async acquireMigrationLock(timeoutMs: number): Promise<boolean> {
    this.#assertOpen("POSTGRES_MIGRATION_LOCK_FAILED");
    if (!Number.isFinite(timeoutMs) || timeoutMs < 0 || timeoutMs > 2_147_483_647) {
      throw databaseError(
        "POSTGRES_MIGRATION_LOCK_FAILED",
        "The migration lock timeout is invalid.",
      );
    }
    if (this.#lockHeld) {
      return true;
    }

    const deadline = performance.now() + timeoutMs;
    try {
      while (true) {
        const result = await this.#client.query<AdvisoryLockRow>(
          "SELECT pg_try_advisory_lock($1::integer, $2::integer) AS acquired",
          [advisoryLockNamespace, advisoryLockKey],
        );
        const acquired = result.rows[0]?.acquired;
        if (typeof acquired !== "boolean") {
          throw databaseError(
            "POSTGRES_MIGRATION_LOCK_FAILED",
            "The migration advisory lock could not be acquired.",
          );
        }
        if (acquired) {
          this.#lockHeld = true;
          return true;
        }

        const remaining = deadline - performance.now();
        if (remaining <= 0) {
          return false;
        }
        await pause(Math.min(advisoryLockPollIntervalMs, remaining));
      }
    } catch (error) {
      if (isPostgresMigrationDatabaseError(error)) {
        throw error;
      }
      throw databaseError(
        "POSTGRES_MIGRATION_LOCK_FAILED",
        "The migration advisory lock could not be acquired.",
      );
    }
  }

  async releaseMigrationLock(): Promise<void> {
    this.#assertOpen("POSTGRES_MIGRATION_LOCK_FAILED");
    if (!this.#lockHeld) {
      return;
    }

    try {
      const result = await this.#client.query<AdvisoryUnlockRow>(
        "SELECT pg_advisory_unlock($1::integer, $2::integer) AS released",
        [advisoryLockNamespace, advisoryLockKey],
      );
      if (result.rows[0]?.released !== true) {
        throw databaseError(
          "POSTGRES_MIGRATION_LOCK_FAILED",
          "The migration advisory lock could not be released.",
        );
      }
      this.#lockHeld = false;
    } catch (error) {
      if (isPostgresMigrationDatabaseError(error)) {
        throw error;
      }
      throw databaseError(
        "POSTGRES_MIGRATION_LOCK_FAILED",
        "The migration advisory lock could not be released.",
      );
    }
  }

  async transaction<T>(operation: (transaction: MigrationTransaction) => Promise<T>): Promise<T> {
    this.#assertOpen("POSTGRES_MIGRATION_TRANSACTION_FAILED");
    if (this.#transactionActive) {
      throw databaseError(
        "POSTGRES_MIGRATION_TRANSACTION_FAILED",
        "Nested migration transactions are not supported.",
      );
    }

    this.#transactionActive = true;
    try {
      await this.#client.query("BEGIN");
    } catch {
      this.#transactionActive = false;
      throw databaseError(
        "POSTGRES_MIGRATION_TRANSACTION_FAILED",
        "The migration transaction could not be started.",
      );
    }

    const transaction = Object.freeze<MigrationTransaction>({
      ensureHistoryTable: async () => this.#ensureHistoryTable(),
      executeMigrationSql: async (sql) => this.#executeMigrationSql(sql),
      insertAppliedMigration: async (migration) => this.#insertAppliedMigration(migration),
    });

    try {
      const result = await operation(transaction);
      await this.#client.query("COMMIT");
      return result;
    } catch (error) {
      try {
        await this.#client.query("ROLLBACK");
      } catch {
        throw databaseError(
          "POSTGRES_MIGRATION_TRANSACTION_FAILED",
          "The migration transaction could not be rolled back.",
        );
      }
      if (isPostgresMigrationDatabaseError(error)) {
        throw error;
      }
      throw databaseError(
        "POSTGRES_MIGRATION_TRANSACTION_FAILED",
        "The migration transaction failed.",
      );
    } finally {
      this.#transactionActive = false;
    }
  }

  async close(): Promise<void> {
    if (this.#closed) {
      return;
    }

    let cleanupFailed = false;
    if (this.#lockHeld) {
      try {
        const result = await this.#client.query<AdvisoryUnlockRow>(
          "SELECT pg_advisory_unlock($1::integer, $2::integer) AS released",
          [advisoryLockNamespace, advisoryLockKey],
        );
        cleanupFailed = result.rows[0]?.released !== true;
      } catch {
        cleanupFailed = true;
      }
    }

    try {
      await this.#client.end();
    } catch {
      cleanupFailed = true;
    } finally {
      this.#closed = true;
      this.#lockHeld = false;
      this.#transactionActive = false;
    }

    if (cleanupFailed) {
      throw databaseError(
        "POSTGRES_MIGRATION_SESSION_CLOSE_FAILED",
        "The PostgreSQL migration session could not be closed cleanly.",
      );
    }
  }

  async #ensureHistoryTable(): Promise<void> {
    this.#assertTransactionActive("POSTGRES_MIGRATION_HISTORY_WRITE_FAILED");
    try {
      await this.#client.query(createHistoryTableSql);
    } catch {
      throw databaseError(
        "POSTGRES_MIGRATION_HISTORY_WRITE_FAILED",
        "The Core migration history table could not be created.",
      );
    }
  }

  /**
   * Sends one migration's complete SQL to PostgreSQL in a single simple-query call — no values, no
   * prepared-statement name, no rewriting — so the server performs all SQL parsing and the packaged
   * bytes are what actually execute.
   */
  async #executeMigrationSql(sql: string): Promise<void> {
    this.#assertTransactionActive("POSTGRES_MIGRATION_EXECUTION_FAILED");
    try {
      await this.#client.query(sql);
    } catch {
      throw databaseError("POSTGRES_MIGRATION_EXECUTION_FAILED", "A module migration failed.");
    }
  }

  async #insertAppliedMigration(migration: MigrationHistoryInsert): Promise<void> {
    this.#assertTransactionActive("POSTGRES_MIGRATION_HISTORY_WRITE_FAILED");
    try {
      await this.#client.query(insertHistorySql, [
        migration.identity,
        migration.moduleId,
        migration.tag,
        migration.index,
        migration.checksum,
      ]);
    } catch {
      throw databaseError(
        "POSTGRES_MIGRATION_HISTORY_WRITE_FAILED",
        "The applied migration could not be recorded.",
      );
    }
  }

  #assertOpen(code: PostgresMigrationDatabaseErrorCode): void {
    if (this.#closed) {
      throw databaseError(code, "The PostgreSQL migration session is closed.");
    }
  }

  #assertTransactionActive(code: PostgresMigrationDatabaseErrorCode): void {
    if (!this.#transactionActive) {
      throw databaseError(code, "A migration transaction is not active.");
    }
  }
}

export function createPostgresMigrationDatabase(
  connectionString: string | undefined,
  options?: PostgresMigrationDatabaseOptions,
): MigrationDatabase {
  const validatedConnectionString = validateConnectionString(connectionString);
  const clientConfig = createClientConfig(validatedConnectionString, options);

  return Object.freeze({
    async openSession(): Promise<MigrationDatabaseSession> {
      const client = new Client(clientConfig);
      client.on("error", () => undefined);
      try {
        await client.connect();
      } catch {
        try {
          await client.end();
        } catch {
          // The public error stays fixed even when driver cleanup also fails.
        }
        throw databaseError(
          "POSTGRES_MIGRATION_CONNECTION_FAILED",
          "The PostgreSQL migration connection could not be established.",
        );
      }
      return new PostgresMigrationDatabaseSession(client);
    },
  });
}