Ekklesia/core/tests/postgres-migration.integration.ts

import { createHash, randomBytes } from "node:crypto";
import net from "node:net";
import path from "node:path";
import { fileURLToPath } from "node:url";
import { Client, type QueryResult, type QueryResultRow } from "pg";
import { afterAll, describe, expect, it, vi } from "vitest";
import { reportMigrationCommandError } from "../src/cli/migration-command-errors.js";
import {
  applyModuleMigrations,
  createPostgresMigrationDatabase,
  discoverCoreMigrations,
  type MigrationApproval,
  type MigrationApprovalProvider,
  MigrationRunnerError,
  type MigrationRunnerErrorCode,
  planModuleMigrations,
  PostgresMigrationDatabaseError,
} from "../src/migrations/index.js";
import type {
  DiscoveredModuleMigration,
  ModuleMigrationAggregate,
} from "../src/modules/migration-discovery.js";

const coreMigrationsDirectory = path.join(
  path.dirname(fileURLToPath(import.meta.url)),
  "..",
  "drizzle",
);

const integrationDatabaseUrlEnvironment = "EKKLESIA_MIGRATION_TEST_DATABASE_URL";
const historyTable = "public.ekklesia_core_migrations";
const migrationLockNamespace = 1_162_562_380;
const migrationLockKey = 1_296_648_018;
const concurrencyBarrierNamespace = 1_162_562_380;
const concurrencyBarrierKey = 1_414_743_380;
const trackedDatabases = new Set<string>();
let databaseCounter = 0;

class IntegrationHarnessError extends Error {
  constructor(message: string) {
    super(message);
    this.name = "IntegrationHarnessError";
    this.stack = `${this.name}: ${this.message}`;
  }
}

function harnessError(message: string): IntegrationHarnessError {
  return new IntegrationHarnessError(message);
}

function baseDatabaseUrl(): string {
  const value = process.env[integrationDatabaseUrlEnvironment];
  if (typeof value !== "string" || value.trim() === "") {
    throw harnessError("The migration integration database is not configured.");
  }

  try {
    const parsed = new URL(value);
    if (parsed.protocol !== "postgres:" && parsed.protocol !== "postgresql:") {
      throw new TypeError("Unsupported protocol.");
    }
  } catch {
    throw harnessError("The migration integration database configuration is invalid.");
  }
  return value;
}

function quoteDatabaseName(databaseName: string): string {
  if (!/^[a-z][a-z0-9_]{0,62}$/u.test(databaseName)) {
    throw harnessError("The generated integration database name is invalid.");
  }
  return `"${databaseName}"`;
}

function makeDatabaseName(label: string): string {
  const safeLabel = label
    .toLowerCase()
    .replaceAll(/[^a-z0-9]+/gu, "_")
    .replaceAll(/^_+|_+$/gu, "")
    .slice(0, 18);
  databaseCounter += 1;
  return `ekk_mig_${process.pid}_${databaseCounter}_${safeLabel || "scenario"}_${randomBytes(4).toString("hex")}`;
}

function databaseUrl(databaseName: string, applicationName?: string): string {
  const parsed = new URL(baseDatabaseUrl());
  parsed.pathname = `/${databaseName}`;
  if (applicationName === undefined) {
    parsed.searchParams.delete("application_name");
  } else {
    parsed.searchParams.set("application_name", applicationName);
  }
  return parsed.toString();
}

async function openAdministrativeClient(
  connectionString: string,
  applicationName: string,
): Promise<Client> {
  const client = new Client({ application_name: applicationName, connectionString });
  client.on("error", () => undefined);
  try {
    await client.connect();
    return client;
  } catch {
    try {
      await client.end();
    } catch {
      // The public harness failure remains static when connection cleanup also fails.
    }
    throw harnessError("The migration integration database connection failed.");
  }
}

async function safeQuery<Row extends QueryResultRow>(
  client: Client,
  text: string,
  values: unknown[] = [],
  failureMessage = "An integration database administration operation failed.",
): Promise<QueryResult<Row>> {
  try {
    return await client.query<Row>(text, values);
  } catch {
    throw harnessError(failureMessage);
  }
}

async function safeEnd(client: Client): Promise<void> {
  try {
    await client.end();
  } catch {
    throw harnessError("An integration database connection could not be closed.");
  }
}

async function createDatabase(databaseName: string): Promise<void> {
  const admin = await openAdministrativeClient(baseDatabaseUrl(), "ekklesia_migration_test_setup");
  let operationError: unknown;
  try {
    await safeQuery(admin, `CREATE DATABASE ${quoteDatabaseName(databaseName)} TEMPLATE template0`);
  } catch (error) {
    operationError = error;
  }
  try {
    await safeEnd(admin);
  } catch (error) {
    if (operationError === undefined) {
      operationError = error;
    }
  }
  if (operationError !== undefined) {
    throw operationError;
  }
}

async function dropDatabase(databaseName: string): Promise<void> {
  const admin = await openAdministrativeClient(
    baseDatabaseUrl(),
    "ekklesia_migration_test_cleanup",
  );
  let operationError: unknown;
  try {
    await safeQuery(
      admin,
      "SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname = $1 AND pid <> pg_backend_pid()",
      [databaseName],
    );
    await safeQuery(
      admin,
      `DROP DATABASE IF EXISTS ${quoteDatabaseName(databaseName)} WITH (FORCE)`,
    );
    trackedDatabases.delete(databaseName);
  } catch (error) {
    operationError = error;
  }
  try {
    await safeEnd(admin);
  } catch (error) {
    if (operationError === undefined) {
      operationError = error;
    }
  }
  if (operationError !== undefined) {
    throw operationError;
  }
}

interface ScenarioDatabase {
  readonly client: Client;
  readonly name: string;
  readonly url: string;
}

async function withIsolatedDatabase<T>(
  label: string,
  operation: (database: ScenarioDatabase) => Promise<T>,
): Promise<T> {
  const name = makeDatabaseName(label);
  await createDatabase(name);
  trackedDatabases.add(name);

  let client: Client | undefined;
  let outcome:
    | { readonly error: unknown; readonly success: false }
    | { readonly success: true; readonly value: T };
  try {
    const url = databaseUrl(name);
    client = await openAdministrativeClient(url, "ekklesia_migration_test_inspection");
    outcome = { success: true, value: await operation(Object.freeze({ client, name, url })) };
  } catch (error) {
    outcome = { error, success: false };
  }

  let cleanupFailed = false;
  if (client !== undefined) {
    try {
      await safeEnd(client);
    } catch {
      cleanupFailed = true;
    }
  }
  try {
    await dropDatabase(name);
  } catch {
    cleanupFailed = true;
  }

  if (!outcome.success) {
    throw outcome.error;
  }
  if (cleanupFailed) {
    throw harnessError("An isolated migration integration database could not be removed.");
  }
  return outcome.value;
}

afterAll(async () => {
  let cleanupFailed = false;
  for (const databaseName of [...trackedDatabases]) {
    try {
      await dropDatabase(databaseName);
    } catch {
      cleanupFailed = true;
    }
  }
  if (cleanupFailed) {
    throw harnessError("Migration integration database cleanup was incomplete.");
  }
});

interface MigrationOptions {
  readonly approvalRequired?: boolean;
  readonly sql: string;
  readonly tag: string;
}

interface ModuleFixture {
  readonly id: string;
  readonly migrations: readonly DiscoveredModuleMigration[];
}

function migration(
  moduleId: string,
  index: number,
  options: MigrationOptions,
): DiscoveredModuleMigration {
  const approvalRequired = options.approvalRequired ?? false;
  return Object.freeze({
    approvalRequired,
    breakpoints: true,
    checksum: createHash("sha256").update(options.sql).digest("hex"),
    destructive: Object.freeze(
      approvalRequired
        ? {
            mode: "approval-required" as const,
            reason: "This integration fixture requires an exact external approval.",
          }
        : { mode: "none" as const },
    ),
    file: `${options.tag}.sql`,
    identity: `${moduleId}:${options.tag}`,
    index,
    moduleId,
    sql: options.sql,
    tag: options.tag,
    timestamp: 1_800_000_000_000 + index,
  });
}

function aggregate(...moduleFixtures: readonly ModuleFixture[]): ModuleMigrationAggregate {
  const modules = Object.freeze(
    moduleFixtures.map((moduleFixture) =>
      Object.freeze({
        migrations: Object.freeze([...moduleFixture.migrations]),
        moduleId: moduleFixture.id,
      }),
    ),
  );
  return Object.freeze({
    discoveredModules: Object.freeze([]),
    migrations: Object.freeze(modules.flatMap((moduleFixture) => moduleFixture.migrations)),
    modules,
  });
}

function moduleFixture(
  id: string,
  migrations: readonly DiscoveredModuleMigration[],
): ModuleFixture {
  return Object.freeze({ id, migrations: Object.freeze([...migrations]) });
}

const peopleInitialSql = `CREATE TABLE "people_records" (
  "id" bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
  "display_name" text NOT NULL
);\n`;
const calendarInitialSql = `CREATE TABLE "calendar_events" (
  "id" bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
  "owner_id" bigint NOT NULL,
  "title" text NOT NULL
);\n`;
const peopleLaterSql = `ALTER TABLE "people_records" ADD COLUMN "email" text;\n`;

function peopleInitial(sql = peopleInitialSql): DiscoveredModuleMigration {
  return migration("people", 0, { sql, tag: "0000_initial" });
}

function calendarInitial(): DiscoveredModuleMigration {
  return migration("calendar", 0, { sql: calendarInitialSql, tag: "0000_initial" });
}

function peopleLater(sql = peopleLaterSql): DiscoveredModuleMigration {
  return migration("people", 1, { sql, tag: "0001_add_email" });
}

interface RelationExistenceRow extends QueryResultRow {
  readonly relation_name: string | null;
}

interface UserObjectRow extends QueryResultRow {
  readonly object_kind: string;
  readonly object_name: string;
  readonly schema_name: string;
}

interface ColumnRow extends QueryResultRow {
  readonly column_default: string | null;
  readonly column_name: string;
  readonly data_type: string;
  readonly is_nullable: string;
  readonly ordinal_position: number;
  readonly table_name: string;
  readonly table_schema: string;
}

interface ConstraintRow extends QueryResultRow {
  readonly constraint_definition: string;
  readonly constraint_name: string;
  readonly table_name: string;
  readonly table_schema: string;
}

interface SequenceRow extends QueryResultRow {
  readonly cache_size: string;
  readonly cycle: boolean;
  readonly increment_by: string;
  readonly last_value: string | null;
  readonly sequence_name: string;
  readonly sequence_schema: string;
}

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

interface BooleanRow extends QueryResultRow {
  readonly value: boolean;
}

interface CountRow extends QueryResultRow {
  readonly value: number;
}

interface DatabaseState {
  readonly columns: readonly ColumnRow[];
  readonly constraints: readonly ConstraintRow[];
  readonly history: readonly HistoryRow[];
  readonly objects: readonly UserObjectRow[];
  readonly sequences: readonly SequenceRow[];
  readonly supplementalObjects: readonly UserObjectRow[];
}

async function relationExists(client: Client, qualifiedName: string): Promise<boolean> {
  const result = await safeQuery<RelationExistenceRow>(
    client,
    "SELECT to_regclass($1) AS relation_name",
    [qualifiedName],
  );
  return result.rows[0]?.relation_name !== null && result.rows[0]?.relation_name !== undefined;
}

async function readHistory(client: Client): Promise<readonly HistoryRow[]> {
  if (!(await relationExists(client, historyTable))) {
    return Object.freeze([]);
  }
  const result = await safeQuery<HistoryRow>(
    client,
    `
      SELECT
        applied_order::integer AS applied_order,
        identity,
        module_id,
        migration_tag,
        migration_index,
        checksum,
        applied_at
      FROM public.ekklesia_core_migrations
      ORDER BY applied_order ASC
    `,
  );
  return Object.freeze(result.rows.map((row) => Object.freeze({ ...row })));
}

async function databaseState(client: Client): Promise<DatabaseState> {
  const objects = await safeQuery<UserObjectRow>(
    client,
    `
        SELECT
          namespace.nspname AS schema_name,
          relation.relname AS object_name,
          relation.relkind::text AS object_kind
        FROM pg_catalog.pg_class AS relation
        INNER JOIN pg_catalog.pg_namespace AS namespace
          ON namespace.oid = relation.relnamespace
        WHERE namespace.nspname NOT LIKE 'pg\\_%' ESCAPE '\\'
          AND namespace.nspname <> 'information_schema'
        ORDER BY namespace.nspname, relation.relname, relation.relkind
      `,
    [],
    "Integration user-object inspection failed.",
  );
  const columns = await safeQuery<ColumnRow>(
    client,
    `
        SELECT
          table_schema,
          table_name,
          column_name,
          ordinal_position,
          data_type,
          is_nullable,
          column_default
        FROM information_schema.columns
        WHERE table_schema NOT LIKE 'pg\\_%' ESCAPE '\\'
          AND table_schema <> 'information_schema'
        ORDER BY table_schema, table_name, ordinal_position
      `,
    [],
    "Integration column inspection failed.",
  );
  const constraints = await safeQuery<ConstraintRow>(
    client,
    `
        SELECT
          namespace.nspname AS table_schema,
          relation.relname AS table_name,
          catalog_constraint.conname AS constraint_name,
          pg_catalog.pg_get_constraintdef(catalog_constraint.oid, true) AS constraint_definition
        FROM pg_catalog.pg_constraint AS catalog_constraint
        INNER JOIN pg_catalog.pg_class AS relation
          ON relation.oid = catalog_constraint.conrelid
        INNER JOIN pg_catalog.pg_namespace AS namespace ON namespace.oid = relation.relnamespace
        WHERE namespace.nspname NOT LIKE 'pg\\_%' ESCAPE '\\'
          AND namespace.nspname <> 'information_schema'
        ORDER BY namespace.nspname, relation.relname, catalog_constraint.conname
      `,
    [],
    "Integration constraint inspection failed.",
  );
  const sequences = await safeQuery<SequenceRow>(
    client,
    `
        SELECT
          schemaname AS sequence_schema,
          sequencename AS sequence_name,
          increment_by::text,
          cycle,
          cache_size::text,
          last_value::text
        FROM pg_catalog.pg_sequences
        WHERE schemaname NOT LIKE 'pg\\_%' ESCAPE '\\'
          AND schemaname <> 'information_schema'
        ORDER BY schemaname, sequencename
      `,
    [],
    "Integration sequence inspection failed.",
  );
  const supplementalObjects = await safeQuery<UserObjectRow>(
    client,
    `
      SELECT
        'schema' AS object_kind,
        namespace.nspname AS object_name,
        '' AS schema_name
      FROM pg_catalog.pg_namespace AS namespace
      WHERE namespace.nspname NOT LIKE 'pg\\_%' ESCAPE '\\'
        AND namespace.nspname <> 'information_schema'
      UNION ALL
      SELECT
        'extension' AS object_kind,
        catalog_extension.extname AS object_name,
        namespace.nspname AS schema_name
      FROM pg_catalog.pg_extension AS catalog_extension
      INNER JOIN pg_catalog.pg_namespace AS namespace
        ON namespace.oid = catalog_extension.extnamespace
      UNION ALL
      SELECT
        'routine' AS object_kind,
        catalog_routine.oid::regprocedure::text AS object_name,
        namespace.nspname AS schema_name
      FROM pg_catalog.pg_proc AS catalog_routine
      INNER JOIN pg_catalog.pg_namespace AS namespace
        ON namespace.oid = catalog_routine.pronamespace
      WHERE namespace.nspname NOT LIKE 'pg\\_%' ESCAPE '\\'
        AND namespace.nspname <> 'information_schema'
      UNION ALL
      SELECT
        CASE user_type.typtype
          WHEN 'd' THEN 'domain'
          WHEN 'e' THEN 'enum'
          WHEN 'm' THEN 'multirange'
          WHEN 'r' THEN 'range'
        END AS object_kind,
        user_type.typname AS object_name,
        namespace.nspname AS schema_name
      FROM pg_catalog.pg_type AS user_type
      INNER JOIN pg_catalog.pg_namespace AS namespace
        ON namespace.oid = user_type.typnamespace
      WHERE namespace.nspname NOT LIKE 'pg\\_%' ESCAPE '\\'
        AND namespace.nspname <> 'information_schema'
        AND user_type.typtype IN ('d', 'e', 'm', 'r')
      ORDER BY object_kind, schema_name, object_name
    `,
    [],
    "Integration supplemental-object inspection failed.",
  );
  const history = await readHistory(client);

  return Object.freeze({
    columns: Object.freeze(columns.rows.map((row) => Object.freeze({ ...row }))),
    constraints: Object.freeze(constraints.rows.map((row) => Object.freeze({ ...row }))),
    history,
    objects: Object.freeze(objects.rows.map((row) => Object.freeze({ ...row }))),
    sequences: Object.freeze(sequences.rows.map((row) => Object.freeze({ ...row }))),
    supplementalObjects: Object.freeze(
      supplementalObjects.rows.map((row) => Object.freeze({ ...row })),
    ),
  });
}

async function columnExists(
  client: Client,
  tableName: string,
  columnName: string,
): Promise<boolean> {
  const result = await safeQuery<BooleanRow>(
    client,
    `
      SELECT EXISTS (
        SELECT 1
        FROM information_schema.columns
        WHERE table_schema = 'public' AND table_name = $1 AND column_name = $2
      ) AS value
    `,
    [tableName, columnName],
  );
  return result.rows[0]?.value === true;
}

async function captureRunnerError(
  operation: Promise<unknown>,
  code: MigrationRunnerErrorCode,
): Promise<MigrationRunnerError> {
  try {
    await operation;
  } catch (error) {
    if (!(error instanceof MigrationRunnerError)) {
      throw harnessError("A migration operation returned an unsafe error type.");
    }
    const migrationError = error as MigrationRunnerError;
    expect(migrationError.code).toBe(code);
    return migrationError;
  }
  throw harnessError("A migration operation unexpectedly succeeded.");
}

function approvalFor(migrationFixture: DiscoveredModuleMigration): MigrationApproval {
  return Object.freeze({
    checksum: migrationFixture.checksum,
    identity: migrationFixture.identity,
    moduleId: migrationFixture.moduleId,
  });
}

function approvalProvider(approvals: readonly MigrationApproval[]): MigrationApprovalProvider {
  const frozenApprovals = Object.freeze(
    approvals.map((approval) => Object.freeze({ ...approval })),
  );
  return Object.freeze({
    getApprovals: () => Promise.resolve(frozenApprovals),
  });
}

async function acquireAdministrativeLock(
  client: Client,
  namespace: number,
  key: number,
): Promise<void> {
  await safeQuery(client, "SELECT pg_advisory_lock($1::integer, $2::integer)", [namespace, key]);
}

async function releaseAdministrativeLock(
  client: Client,
  namespace: number,
  key: number,
): Promise<void> {
  const result = await safeQuery<BooleanRow>(
    client,
    "SELECT pg_advisory_unlock($1::integer, $2::integer) AS value",
    [namespace, key],
  );
  if (result.rows[0]?.value !== true) {
    throw harnessError("An integration advisory lock was not released.");
  }
}

async function expectMigrationLockAvailable(client: Client): Promise<void> {
  const result = await safeQuery<BooleanRow>(
    client,
    "SELECT pg_try_advisory_lock($1::integer, $2::integer) AS value",
    [migrationLockNamespace, migrationLockKey],
  );
  expect(result.rows[0]?.value).toBe(true);
  if (result.rows[0]?.value === true) {
    await releaseAdministrativeLock(client, migrationLockNamespace, migrationLockKey);
  }
}

async function waitForDatabaseCondition(
  check: () => Promise<boolean>,
  timeoutMs = 5_000,
): Promise<void> {
  const deadline = performance.now() + timeoutMs;
  while (performance.now() < deadline) {
    if (await check()) {
      return;
    }
    await new Promise<void>((resolve) => setTimeout(resolve, 25));
  }
  throw harnessError("A bounded integration database condition was not observed.");
}

describe.sequential("live PostgreSQL module migration execution", () => {
  it("keeps fresh planning read-only", async () => {
    await withIsolatedDatabase("fresh_plan", async (database) => {
      const people = peopleInitial();
      const calendar = calendarInitial();
      const inventory = aggregate(
        moduleFixture("people", [people]),
        moduleFixture("calendar", [calendar]),
      );
      const before = await databaseState(database.client);

      const plan = await planModuleMigrations({
        aggregate: inventory,
        database: createPostgresMigrationDatabase(database.url),
      });

      expect(plan.entries.map((entry) => [entry.identity, entry.status])).toEqual([
        [people.identity, "pending"],
        [calendar.identity, "pending"],
      ]);
      expect(plan).toMatchObject({ appliedCount: 0, approvalRequiredCount: 0, pendingCount: 2 });
      expect(await databaseState(database.client)).toEqual(before);
      expect(await relationExists(database.client, historyTable)).toBe(false);
      expect(await relationExists(database.client, "public.people_records")).toBe(false);
      expect(await relationExists(database.client, "public.calendar_events")).toBe(false);
    });
  });

  it("applies a fresh dependency-ordered inventory and records exact history", async () => {
    await withIsolatedDatabase("fresh_apply", async (database) => {
      const people = peopleInitial();
      const calendar = calendarInitial();
      const inventory = aggregate(
        moduleFixture("people", [people]),
        moduleFixture("calendar", [calendar]),
      );

      const result = await applyModuleMigrations({
        aggregate: inventory,
        database: createPostgresMigrationDatabase(database.url),
      });

      expect(result.applied.map((entry) => entry.identity)).toEqual([
        people.identity,
        calendar.identity,
      ]);
      expect(await relationExists(database.client, "public.people_records")).toBe(true);
      expect(await relationExists(database.client, "public.calendar_events")).toBe(true);
      const history = await readHistory(database.client);
      expect(history.map(({ applied_at: _appliedAt, ...row }) => row)).toEqual([
        {
          applied_order: 1,
          checksum: people.checksum,
          identity: people.identity,
          migration_index: 0,
          migration_tag: people.tag,
          module_id: people.moduleId,
        },
        {
          applied_order: 2,
          checksum: calendar.checksum,
          identity: calendar.identity,
          migration_index: 0,
          migration_tag: calendar.tag,
          module_id: calendar.moduleId,
        },
      ]);
      expect(history).toHaveLength(2);
      expect(
        history.every(
          (row) => row.applied_at instanceof Date && !Number.isNaN(row.applied_at.getTime()),
        ),
      ).toBe(true);
    });
  });

  it("performs a genuine no-op without changing schema, data, sequences, or history", async () => {
    await withIsolatedDatabase("no_op", async (database) => {
      const initial = peopleInitial();
      const inventory = aggregate(moduleFixture("people", [initial]));
      const migrationDatabase = createPostgresMigrationDatabase(database.url);
      await applyModuleMigrations({ aggregate: inventory, database: migrationDatabase });
      await safeQuery(
        database.client,
        "INSERT INTO public.people_records (display_name) VALUES ($1)",
        ["No-op sentinel"],
      );
      const before = await databaseState(database.client);
      const beforeData = await safeQuery<QueryResultRow>(
        database.client,
        "SELECT id::integer, display_name FROM public.people_records ORDER BY id",
      );

      const result = await applyModuleMigrations({
        aggregate: inventory,
        database: migrationDatabase,
      });

      expect(result).toEqual({ applied: [], appliedCount: 0, noOp: true });
      expect(await databaseState(database.client)).toEqual(before);
      const afterData = await safeQuery<QueryResultRow>(
        database.client,
        "SELECT id::integer, display_name FROM public.people_records ORDER BY id",
      );
      expect(afterData.rows).toEqual(beforeData.rows);
    });
  });

  it("executes only a newly presented later migration", async () => {
    await withIsolatedDatabase("later", async (database) => {
      const initial = peopleInitial();
      const later = peopleLater();
      const initialInventory = aggregate(moduleFixture("people", [initial]));
      const laterInventory = aggregate(moduleFixture("people", [initial, later]));
      const migrationDatabase = createPostgresMigrationDatabase(database.url);
      await applyModuleMigrations({ aggregate: initialInventory, database: migrationDatabase });
      const originalHistory = await readHistory(database.client);

      const result = await applyModuleMigrations({
        aggregate: laterInventory,
        database: migrationDatabase,
      });

      expect(result.applied.map((entry) => entry.identity)).toEqual([later.identity]);
      expect(await columnExists(database.client, "people_records", "email")).toBe(true);
      const history = await readHistory(database.client);
      expect(history).toHaveLength(2);
      expect(history[0]).toEqual(originalHistory[0]);
      expect(history[1]).toMatchObject({
        applied_order: 2,
        checksum: later.checksum,
        identity: later.identity,
        migration_index: 1,
        migration_tag: later.tag,
        module_id: later.moduleId,
      });
    });
  });

  it("rejects edited applied contents during both planning and applying", async () => {
    await withIsolatedDatabase("edited", async (database) => {
      const initial = peopleInitial();
      const edited = peopleInitial(
        `CREATE TABLE "people_records" ("id" bigint PRIMARY KEY, "edited_marker" text);\n`,
      );
      const originalInventory = aggregate(moduleFixture("people", [initial]));
      const editedInventory = aggregate(moduleFixture("people", [edited]));
      const migrationDatabase = createPostgresMigrationDatabase(database.url);
      await applyModuleMigrations({ aggregate: originalInventory, database: migrationDatabase });
      const before = await databaseState(database.client);

      const planError = await captureRunnerError(
        planModuleMigrations({ aggregate: editedInventory, database: migrationDatabase }),
        "MIGRATION_CHECKSUM_MISMATCH",
      );
      expect(planError.migrationIdentity).toBe(initial.identity);
      expect(await databaseState(database.client)).toEqual(before);

      const applyError = await captureRunnerError(
        applyModuleMigrations({ aggregate: editedInventory, database: migrationDatabase }),
        "MIGRATION_CHECKSUM_MISMATCH",
      );
      expect(applyError.migrationIdentity).toBe(initial.identity);
      expect(await databaseState(database.client)).toEqual(before);
      expect(await columnExists(database.client, "people_records", "edited_marker")).toBe(false);
    });
  });

  it("rejects an unknown applied identity without repair or further execution", async () => {
    await withIsolatedDatabase("unknown_history", async (database) => {
      const initial = peopleInitial();
      const later = peopleLater();
      const initialInventory = aggregate(moduleFixture("people", [initial]));
      const completeInventory = aggregate(moduleFixture("people", [initial, later]));
      const migrationDatabase = createPostgresMigrationDatabase(database.url);
      await applyModuleMigrations({ aggregate: initialInventory, database: migrationDatabase });
      await safeQuery(
        database.client,
        `
          INSERT INTO public.ekklesia_core_migrations
            (identity, module_id, migration_tag, migration_index, checksum)
          VALUES ($1, $2, $3, $4, $5)
        `,
        ["ghost:0000_unknown", "ghost", "0000_unknown", 0, "f".repeat(64)],
      );
      const corrupted = await databaseState(database.client);

      await captureRunnerError(
        planModuleMigrations({ aggregate: completeInventory, database: migrationDatabase }),
        "MIGRATION_UNKNOWN_APPLIED",
      );
      await captureRunnerError(
        applyModuleMigrations({ aggregate: completeInventory, database: migrationDatabase }),
        "MIGRATION_UNKNOWN_APPLIED",
      );

      expect(await databaseState(database.client)).toEqual(corrupted);
      expect(await columnExists(database.client, "people_records", "email")).toBe(false);
      expect((await readHistory(database.client)).map((row) => row.identity)).toEqual([
        initial.identity,
        "ghost:0000_unknown",
      ]);
    });
  });

  it("rejects an invalid per-module history prefix without silent repair", async () => {
    await withIsolatedDatabase("history_gap", async (database) => {
      const initial = peopleInitial();
      const later = peopleLater();
      const unreachable = migration("people", 2, {
        sql: `CREATE TABLE "people_unreachable" ("id" bigint PRIMARY KEY);\n`,
        tag: "0002_unreachable",
      });
      const appliedInventory = aggregate(moduleFixture("people", [initial, later]));
      const completeInventory = aggregate(moduleFixture("people", [initial, later, unreachable]));
      const migrationDatabase = createPostgresMigrationDatabase(database.url);
      await applyModuleMigrations({ aggregate: appliedInventory, database: migrationDatabase });
      await safeQuery(
        database.client,
        "DELETE FROM public.ekklesia_core_migrations WHERE identity = $1",
        [initial.identity],
      );
      const corrupted = await databaseState(database.client);

      await captureRunnerError(
        planModuleMigrations({ aggregate: completeInventory, database: migrationDatabase }),
        "MIGRATION_HISTORY_INVALID",
      );
      await captureRunnerError(
        applyModuleMigrations({ aggregate: completeInventory, database: migrationDatabase }),
        "MIGRATION_HISTORY_INVALID",
      );

      expect(await databaseState(database.client)).toEqual(corrupted);
      expect(await relationExists(database.client, "public.people_unreachable")).toBe(false);
      expect((await readHistory(database.client)).map((row) => row.identity)).toEqual([
        later.identity,
      ]);
    });
  });

  it("rolls back the entire current pending batch after a real PostgreSQL failure", async () => {
    await withIsolatedDatabase("rollback", async (database) => {
      const committed = migration("rollback", 0, {
        sql: `CREATE TABLE "rollback_committed" ("id" bigint PRIMARY KEY);\n`,
        tag: "0000_committed",
      });
      const pendingFirst = migration("rollback", 1, {
        sql: `CREATE TABLE "rollback_pending_one" ("id" bigint PRIMARY KEY);\n`,
        tag: "0001_pending_one",
      });
      const pendingFailure = migration("rollback", 2, {
        sql: `CREATE TABLE "rollback_pending_two" ("id" bigint PRIMARY KEY);\n--> statement-breakpoint\nSELECT 1 / 0;\n`,
        tag: "0002_pending_failure",
      });
      const committedInventory = aggregate(moduleFixture("rollback", [committed]));
      const completeInventory = aggregate(
        moduleFixture("rollback", [committed, pendingFirst, pendingFailure]),
      );
      const migrationDatabase = createPostgresMigrationDatabase(database.url);
      await applyModuleMigrations({ aggregate: committedInventory, database: migrationDatabase });

      const error = await captureRunnerError(
        applyModuleMigrations({ aggregate: completeInventory, database: migrationDatabase }),
        "MIGRATION_EXECUTION_FAILED",
      );

      expect(error.migrationIdentity).toBe(pendingFailure.identity);
      expect(await relationExists(database.client, "public.rollback_committed")).toBe(true);
      expect(await relationExists(database.client, "public.rollback_pending_one")).toBe(false);
      expect(await relationExists(database.client, "public.rollback_pending_two")).toBe(false);
      expect((await readHistory(database.client)).map((row) => row.identity)).toEqual([
        committed.identity,
      ]);
      await expectMigrationLockAvailable(database.client);
    });
  });

  it("honors Drizzle statement breakpoints without splitting procedural semicolons", async () => {
    await withIsolatedDatabase("breakpoints", async (database) => {
      const sql = `CREATE TABLE "breakpoints_steps" (
  "id" integer PRIMARY KEY,
  "note" text NOT NULL
);
--> statement-breakpoint
DO $body$
BEGIN
  INSERT INTO "breakpoints_steps" ("id", "note") VALUES (1, 'first;still-one-statement');
END
$body$;
--> statement-breakpoint
ALTER TABLE "breakpoints_steps" ADD COLUMN "sequence_value" integer;
UPDATE "breakpoints_steps" SET "sequence_value" = 1 WHERE "id" = 1;
INSERT INTO "breakpoints_steps" ("id", "note", "sequence_value")
SELECT 2, 'second', "sequence_value" + 1 FROM "breakpoints_steps" WHERE "id" = 1;
`;
      const initial = migration("breakpoints", 0, { sql, tag: "0000_ordered_steps" });

      await applyModuleMigrations({
        aggregate: aggregate(moduleFixture("breakpoints", [initial])),
        database: createPostgresMigrationDatabase(database.url),
      });

      const rows = await safeQuery<QueryResultRow>(
        database.client,
        `SELECT id, note, sequence_value FROM public.breakpoints_steps ORDER BY id`,
      );
      expect(rows.rows).toEqual([
        { id: 1, note: "first;still-one-statement", sequence_value: 1 },
        { id: 2, note: "second", sequence_value: 2 },
      ]);
      expect((await readHistory(database.client)).map((row) => row.identity)).toEqual([
        initial.identity,
      ]);
    });
  });

  it("serializes two real migrators with the session advisory lock", async () => {
    await withIsolatedDatabase("concurrency", async (database) => {
      await acquireAdministrativeLock(
        database.client,
        concurrencyBarrierNamespace,
        concurrencyBarrierKey,
      );
      let barrierHeld = true;
      const sql = `SELECT pg_advisory_xact_lock(${concurrencyBarrierNamespace}::integer, ${concurrencyBarrierKey}::integer);\n--> statement-breakpoint\nCREATE TABLE "concurrency_records" ("id" bigint PRIMARY KEY);\n`;
      const initial = migration("concurrency", 0, { sql, tag: "0000_serialized" });
      const inventory = aggregate(moduleFixture("concurrency", [initial]));
      const applicationName = "ekklesia_migration_test_concurrency";
      const productionUrl = databaseUrl(database.name, applicationName);
      const firstApply = applyModuleMigrations({
        aggregate: inventory,
        database: createPostgresMigrationDatabase(productionUrl),
        lockTimeoutMs: 5_000,
      });

      try {
        await waitForDatabaseCondition(async () => {
          const result = await safeQuery<CountRow>(
            database.client,
            `
              SELECT count(*)::integer AS value
              FROM pg_catalog.pg_locks AS lock
              INNER JOIN pg_catalog.pg_stat_activity AS activity ON activity.pid = lock.pid
              WHERE lock.locktype = 'advisory'
                AND lock.classid = $1::oid
                AND lock.objid = $2::oid
                AND lock.objsubid = 2
                AND lock.granted
                AND activity.datname = current_database()
                AND activity.application_name = $3
            `,
            [migrationLockNamespace, migrationLockKey, applicationName],
          );
          return result.rows[0]?.value === 1;
        });

        const secondError = await captureRunnerError(
          applyModuleMigrations({
            aggregate: inventory,
            database: createPostgresMigrationDatabase(productionUrl),
            lockTimeoutMs: 250,
          }),
          "MIGRATION_LOCK_TIMEOUT",
        );
        expect(secondError.migrationIdentity).toBeUndefined();
        expect(await relationExists(database.client, historyTable)).toBe(false);
        expect(await relationExists(database.client, "public.concurrency_records")).toBe(false);

        await releaseAdministrativeLock(
          database.client,
          concurrencyBarrierNamespace,
          concurrencyBarrierKey,
        );
        barrierHeld = false;
        const firstResult = await firstApply;
        expect(firstResult.applied.map((entry) => entry.identity)).toEqual([initial.identity]);
      } finally {
        if (barrierHeld) {
          await releaseAdministrativeLock(
            database.client,
            concurrencyBarrierNamespace,
            concurrencyBarrierKey,
          );
        }
        await Promise.allSettled([firstApply]);
      }

      expect(await relationExists(database.client, "public.concurrency_records")).toBe(true);
      expect((await readHistory(database.client)).map((row) => row.identity)).toEqual([
        initial.identity,
      ]);
    });
  });

  it("times out on the real lock, releases it, and permits a later apply", async () => {
    await withIsolatedDatabase("lock_timeout", async (database) => {
      const initial = migration("locks", 0, {
        sql: `CREATE TABLE "locks_records" ("id" bigint PRIMARY KEY);\n`,
        tag: "0000_initial",
      });
      const inventory = aggregate(moduleFixture("locks", [initial]));
      const migrationDatabase = createPostgresMigrationDatabase(database.url);
      await acquireAdministrativeLock(database.client, migrationLockNamespace, migrationLockKey);
      let lockHeld = true;
      const before = await databaseState(database.client);
      const startedAt = performance.now();
      try {
        await captureRunnerError(
          applyModuleMigrations({
            aggregate: inventory,
            database: migrationDatabase,
            lockTimeoutMs: 250,
          }),
          "MIGRATION_LOCK_TIMEOUT",
        );
      } finally {
        await releaseAdministrativeLock(database.client, migrationLockNamespace, migrationLockKey);
        lockHeld = false;
      }
      const elapsedMs = performance.now() - startedAt;
      expect(elapsedMs).toBeGreaterThanOrEqual(150);
      expect(elapsedMs).toBeLessThan(2_000);
      expect(await databaseState(database.client)).toEqual(before);
      expect(lockHeld).toBe(false);

      const result = await applyModuleMigrations({
        aggregate: inventory,
        database: migrationDatabase,
      });
      expect(result.applied.map((entry) => entry.identity)).toEqual([initial.identity]);
      expect(await relationExists(database.client, "public.locks_records")).toBe(true);
    });
  });

  it("does not strand the migration lock after execution failure", async () => {
    await withIsolatedDatabase("failed_lock_release", async (database) => {
      const failing = migration("lock_failure", 0, {
        sql: `CREATE TABLE "lock_failure_records" ("id" bigint PRIMARY KEY);\n--> statement-breakpoint\nSELECT 1 / 0;\n`,
        tag: "0000_initial",
      });
      const corrected = migration("lock_failure", 0, {
        sql: `CREATE TABLE "lock_failure_records" ("id" bigint PRIMARY KEY);\n`,
        tag: "0000_initial",
      });
      const migrationDatabase = createPostgresMigrationDatabase(database.url);

      await captureRunnerError(
        applyModuleMigrations({
          aggregate: aggregate(moduleFixture("lock_failure", [failing])),
          database: migrationDatabase,
        }),
        "MIGRATION_EXECUTION_FAILED",
      );
      await expectMigrationLockAvailable(database.client);
      expect(await relationExists(database.client, historyTable)).toBe(false);
      expect(await relationExists(database.client, "public.lock_failure_records")).toBe(false);

      const result = await applyModuleMigrations({
        aggregate: aggregate(moduleFixture("lock_failure", [corrected])),
        database: migrationDatabase,
      });
      expect(result.applied.map((entry) => entry.identity)).toEqual([corrected.identity]);
    });
  });

  it("requires an exact destructive approval before any persistent mutation", async () => {
    await withIsolatedDatabase("approvals", async (database) => {
      const destructive = migration("destructive", 0, {
        approvalRequired: true,
        sql: `CREATE TABLE "destructive_records" ("id" bigint PRIMARY KEY);\n`,
        tag: "0000_replace_records",
      });
      const inventory = aggregate(moduleFixture("destructive", [destructive]));
      const migrationDatabase = createPostgresMigrationDatabase(database.url);
      const exact = approvalFor(destructive);
      const rejectedApprovals: readonly (readonly MigrationApproval[] | undefined)[] = [
        undefined,
        [Object.freeze({ ...exact, moduleId: "wrong-module" })],
        [Object.freeze({ ...exact, identity: "destructive:0000_wrong" })],
        [Object.freeze({ ...exact, checksum: "0".repeat(64) })],
      ];

      for (const supplied of rejectedApprovals) {
        const before = await databaseState(database.client);
        const operation =
          supplied === undefined
            ? applyModuleMigrations({ aggregate: inventory, database: migrationDatabase })
            : applyModuleMigrations({
                aggregate: inventory,
                approvals: approvalProvider(supplied),
                database: migrationDatabase,
              });
        const error = await captureRunnerError(operation, "MIGRATION_APPROVAL_REQUIRED");
        expect(error.migrationIdentity).toBe(destructive.identity);
        expect(await databaseState(database.client)).toEqual(before);
        expect(await relationExists(database.client, historyTable)).toBe(false);
        expect(await relationExists(database.client, "public.destructive_records")).toBe(false);
      }

      const result = await applyModuleMigrations({
        aggregate: inventory,
        approvals: approvalProvider([exact]),
        database: migrationDatabase,
      });
      expect(result.applied.map((entry) => entry.identity)).toEqual([destructive.identity]);
      expect(await relationExists(database.client, "public.destructive_records")).toBe(true);
      expect((await readHistory(database.client)).map((row) => row.identity)).toEqual([
        destructive.identity,
      ]);
    });
  });

  it("sanitizes a real PostgreSQL failure and normal command reporting", async () => {
    await withIsolatedDatabase("sanitization", async (database) => {
      const hostilePassword = "fake-password-7f7e620d";
      const hostileUrl = `postgresql://hostile:${hostilePassword}@private-db.invalid/secret`;
      const hostileHostname = "private-db.invalid";
      const hostilePath = "/srv/ekklesia/private/migration.sql";
      const hostileSqlMarker = "HOSTILE_RAW_SQL_MARKER_45f28a";
      const sql = `DO $body$
BEGIN
  RAISE EXCEPTION '${hostilePassword} ${hostileUrl} ${hostileHostname} ${hostilePath} ${hostileSqlMarker}';
END
$body$;\n`;
      const failing = migration("sanitization", 0, {
        sql,
        tag: "0000_hostile_failure",
      });
      const capturedOutput: string[] = [];
      const errorSpy = vi
        .spyOn(console, "error")
        .mockImplementation((...values: unknown[]) =>
          capturedOutput.push(values.map(String).join(" ")),
        );
      const logSpy = vi
        .spyOn(console, "log")
        .mockImplementation((...values: unknown[]) =>
          capturedOutput.push(values.map(String).join(" ")),
        );
      let publicError: MigrationRunnerError;
      try {
        publicError = await captureRunnerError(
          applyModuleMigrations({
            aggregate: aggregate(moduleFixture("sanitization", [failing])),
            database: createPostgresMigrationDatabase(database.url),
          }),
          "MIGRATION_EXECUTION_FAILED",
        );
        reportMigrationCommandError(publicError);
      } finally {
        errorSpy.mockRestore();
        logSpy.mockRestore();
      }

      const publicText = [
        String(publicError),
        publicError.message,
        publicError.stack ?? "",
        JSON.stringify(publicError),
        ...capturedOutput,
      ].join("\n");
      expect(publicError.migrationIdentity).toBe(failing.identity);
      expect(publicText.includes("MIGRATION_EXECUTION_FAILED")).toBe(true);
      expect(publicText.includes(failing.identity)).toBe(true);
      const hostileValues = [
        hostilePassword,
        hostileUrl,
        hostileHostname,
        hostilePath,
        hostileSqlMarker,
        "RAISE EXCEPTION",
        database.url,
        new URL(database.url).password,
        new URL(database.url).hostname,
      ];
      expect(hostileValues.some((hostileValue) => publicText.includes(hostileValue))).toBe(false);
      expect(await relationExists(database.client, historyTable)).toBe(false);
    });
  });
});

const coreIdentityTables = Object.freeze([
  "ekklesia_core_user",
  "ekklesia_core_session",
  "ekklesia_core_account",
  "ekklesia_core_verification",
]);

async function loadRealCoreMigrations(): Promise<readonly DiscoveredModuleMigration[]> {
  const migrations = await discoverCoreMigrations({ coreMigrationsDirectory });
  if (migrations.length === 0) {
    throw harnessError("The committed Core identity migration is missing.");
  }
  return migrations;
}

interface IndexRow extends QueryResultRow {
  readonly indexname: string;
  readonly tablename: string;
}

async function coreIdentityIndexNames(client: Client): Promise<readonly string[]> {
  const result = await safeQuery<IndexRow>(
    client,
    `SELECT tablename, indexname FROM pg_catalog.pg_indexes
     WHERE schemaname = 'public' AND tablename = ANY($1::text[])
     ORDER BY tablename, indexname`,
    [coreIdentityTables],
  );
  return Object.freeze(result.rows.map((row) => row.indexname));
}

describe.sequential("live PostgreSQL Core identity migration execution", () => {
  it("plans the committed Core identity migration on a fresh database without mutation", async () => {
    await withIsolatedDatabase("core_fresh_plan", async (database) => {
      const coreMigrations = await loadRealCoreMigrations();
      const inventory = aggregate(moduleFixture("core", coreMigrations));
      const before = await databaseState(database.client);

      const plan = await planModuleMigrations({
        aggregate: inventory,
        database: createPostgresMigrationDatabase(database.url),
      });

      expect(plan.entries.map((entry) => entry.status)).toEqual(
        coreMigrations.map(() => "pending"),
      );
      expect(await databaseState(database.client)).toEqual(before);
      expect(await relationExists(database.client, historyTable)).toBe(false);
      for (const table of coreIdentityTables) {
        expect(await relationExists(database.client, `public.${table}`)).toBe(false);
      }
    });
  });

  it("applies the committed Core identity migration and creates exactly the expected tables and indexes", async () => {
    await withIsolatedDatabase("core_apply", async (database) => {
      const coreMigrations = await loadRealCoreMigrations();
      const inventory = aggregate(moduleFixture("core", coreMigrations));

      const result = await applyModuleMigrations({
        aggregate: inventory,
        database: createPostgresMigrationDatabase(database.url),
      });

      expect(result.applied.map((entry) => entry.identity)).toEqual(
        coreMigrations.map((entry) => entry.identity),
      );
      for (const table of coreIdentityTables) {
        expect(await relationExists(database.client, `public.${table}`)).toBe(true);
      }
      expect(await coreIdentityIndexNames(database.client)).toEqual([
        "ekklesia_core_account_pkey",
        "ekklesia_core_account_user_id_idx",
        "ekklesia_core_session_pkey",
        "ekklesia_core_session_token_unique",
        "ekklesia_core_session_user_id_idx",
        "ekklesia_core_user_email_unique",
        "ekklesia_core_user_pkey",
        "ekklesia_core_verification_identifier_idx",
        "ekklesia_core_verification_pkey",
      ]);

      const history = await readHistory(database.client);
      expect(history).toHaveLength(coreMigrations.length);
      expect(history.map(({ applied_at: _appliedAt, ...row }) => row)).toEqual(
        coreMigrations.map((entry, index) => ({
          applied_order: index + 1,
          checksum: entry.checksum,
          identity: entry.identity,
          migration_index: entry.index,
          migration_tag: entry.tag,
          module_id: "core",
        })),
      );
    });
  });

  it("performs a genuine no-op on a second Core identity apply", async () => {
    await withIsolatedDatabase("core_no_op", async (database) => {
      const coreMigrations = await loadRealCoreMigrations();
      const inventory = aggregate(moduleFixture("core", coreMigrations));
      const migrationDatabase = createPostgresMigrationDatabase(database.url);
      await applyModuleMigrations({ aggregate: inventory, database: migrationDatabase });
      const before = await databaseState(database.client);

      const result = await applyModuleMigrations({
        aggregate: inventory,
        database: migrationDatabase,
      });

      expect(result).toEqual({ applied: [], appliedCount: 0, noOp: true });
      expect(await databaseState(database.client)).toEqual(before);
    });
  });

  it("coexists with a fixture module migration, executing Core first as documented", async () => {
    await withIsolatedDatabase("core_and_module", async (database) => {
      const coreMigrations = await loadRealCoreMigrations();
      const people = peopleInitial();
      const inventory = aggregate(
        moduleFixture("core", coreMigrations),
        moduleFixture("people", [people]),
      );

      const result = await applyModuleMigrations({
        aggregate: inventory,
        database: createPostgresMigrationDatabase(database.url),
      });

      expect(result.applied.map((entry) => entry.identity)).toEqual([
        ...coreMigrations.map((entry) => entry.identity),
        people.identity,
      ]);
      for (const table of coreIdentityTables) {
        expect(await relationExists(database.client, `public.${table}`)).toBe(true);
      }
      expect(await relationExists(database.client, "public.people_records")).toBe(true);
      const history = await readHistory(database.client);
      expect(history.map((row) => row.module_id)).toEqual([
        ...coreMigrations.map(() => "core"),
        "people",
      ]);
    });
  });

  it("does not corrupt previously valid module-only history when Core is introduced later", async () => {
    await withIsolatedDatabase("core_added_later", async (database) => {
      const people = peopleInitial();
      const migrationDatabase = createPostgresMigrationDatabase(database.url);
      await applyModuleMigrations({
        aggregate: aggregate(moduleFixture("people", [people])),
        database: migrationDatabase,
      });
      const historyBeforeCore = await readHistory(database.client);

      const coreMigrations = await loadRealCoreMigrations();
      const result = await applyModuleMigrations({
        aggregate: aggregate(
          moduleFixture("core", coreMigrations),
          moduleFixture("people", [people]),
        ),
        database: migrationDatabase,
      });

      expect(result.applied.map((entry) => entry.identity)).toEqual(
        coreMigrations.map((entry) => entry.identity),
      );
      const historyAfterCore = await readHistory(database.client);
      const peopleRowBefore = historyBeforeCore.find((row) => row.module_id === "people");
      const peopleRowAfter = historyAfterCore.find((row) => row.module_id === "people");
      expect(peopleRowAfter).toEqual(peopleRowBefore);
      expect(historyAfterCore).toHaveLength(historyBeforeCore.length + coreMigrations.length);
    });
  });

  it("does not damage the committed Core identity migration when a later module migration fails", async () => {
    await withIsolatedDatabase("core_rollback_safety", async (database) => {
      const coreMigrations = await loadRealCoreMigrations();
      const migrationDatabase = createPostgresMigrationDatabase(database.url);
      await applyModuleMigrations({
        aggregate: aggregate(moduleFixture("core", coreMigrations)),
        database: migrationDatabase,
      });
      const historyAfterCore = await readHistory(database.client);

      const failing = migration("broken", 0, {
        sql: `CREATE TABLE "broken_records" ("id" bigint PRIMARY KEY);\n--> statement-breakpoint\nSELECT 1 / 0;\n`,
        tag: "0000_broken",
      });

      const error = await captureRunnerError(
        applyModuleMigrations({
          aggregate: aggregate(
            moduleFixture("core", coreMigrations),
            moduleFixture("broken", [failing]),
          ),
          database: migrationDatabase,
        }),
        "MIGRATION_EXECUTION_FAILED",
      );

      expect(error.migrationIdentity).toBe(failing.identity);
      for (const table of coreIdentityTables) {
        expect(await relationExists(database.client, `public.${table}`)).toBe(true);
      }
      expect(await relationExists(database.client, "public.broken_records")).toBe(false);
      expect(await readHistory(database.client)).toEqual(historyAfterCore);
      await expectMigrationLockAvailable(database.client);
    });
  });
});

/**
 * Pass 4: the migration commands' finite timeout defaults, and whole-batch SQL execution, proven
 * against real PostgreSQL and a real TCP endpoint rather than a fake.
 *
 * Every listener, client, and disposable database here is closed in `finally`/`afterAll`, including
 * after a failed assertion.
 */
describe.sequential("live PostgreSQL migration command hardening", () => {
  it("bounds a connection to an endpoint that accepts TCP but never completes PostgreSQL startup", async () => {
    // A real blackhole listener bound only to loopback: it accepts the socket and then says
    // nothing, which is exactly the case `connectionTimeoutMillis` exists for.
    const sockets: net.Socket[] = [];
    const listener = net.createServer((socket) => {
      sockets.push(socket);
    });
    try {
      await new Promise<void>((resolve, reject) => {
        listener.once("error", reject);
        listener.listen(0, "127.0.0.1", () => resolve());
      });
      const address = listener.address();
      if (!address || typeof address !== "object") {
        throw harnessError("The blackhole listener has no bound address.");
      }

      // A short, test-only override; production keeps MIGRATION_CONNECTION_TIMEOUT_MS unchanged.
      const blackholeDatabase = createPostgresMigrationDatabase(
        `postgresql://ekklesia:[email protected]:${address.port}/ekklesia`,
        { connectionTimeoutMs: 300, operationTimeoutMs: 300 },
      );

      const startedAt = Date.now();
      let captured: unknown;
      try {
        await blackholeDatabase.openSession();
      } catch (error) {
        captured = error;
      }

      expect(captured).toBeInstanceOf(PostgresMigrationDatabaseError);
      expect(captured).toMatchObject({ code: "POSTGRES_MIGRATION_CONNECTION_FAILED" });
      // Finite, and nowhere near an unbounded wait.
      expect(Date.now() - startedAt).toBeLessThan(10_000);
      const publicText = `${String(captured)}\n${(captured as Error).stack ?? ""}`;
      expect(publicText).not.toContain("127.0.0.1");
      expect(publicText).not.toContain("ekklesia:ekklesia");
      expect(publicText).not.toContain(String(address.port));
    } finally {
      for (const socket of sockets) {
        socket.destroy();
      }
      await new Promise<void>((resolve) => {
        listener.close(() => resolve());
      });
    }
  });

  it("aborts a migration that exceeds the operation deadline and commits no history row", async () => {
    await withIsolatedDatabase("operation_timeout", async (database) => {
      const slow = migration("people", 0, {
        sql: "SELECT pg_sleep(5);",
        tag: "0000_slow",
      });
      // The operation deadline is the behavior under test, so a bounded server-side delay is the
      // only way to exercise it.
      const migrationDatabase = createPostgresMigrationDatabase(database.url, {
        connectionTimeoutMs: 5_000,
        operationTimeoutMs: 250,
      });

      const error = await captureRunnerError(
        applyModuleMigrations({
          aggregate: aggregate({ id: "people", migrations: [slow] }),
          database: migrationDatabase,
        }),
        "MIGRATION_EXECUTION_FAILED",
      );

      expect(error.migrationIdentity).toBe(slow.identity);
      const publicText = `${String(error)}\n${error.stack ?? ""}`;
      expect(publicText).not.toContain("pg_sleep");
      expect(publicText).not.toContain(database.name);

      // Nothing was committed: the transaction rolled back with its history insert.
      const history = await safeQuery<{ identity: string }>(
        database.client,
        `select identity from public.ekklesia_core_migrations`,
      ).catch(() => undefined);
      expect(history?.rows ?? []).toHaveLength(0);
    });
  });

  it("executes marker bytes inside literals, dollar-quoted bodies, and comments intact", async () => {
    await withIsolatedDatabase("intact_sql", async (database) => {
      const marker = "--> statement-breakpoint";
      const sql = [
        `CREATE TABLE "people_notes" ("id" integer primary key, "body" text not null);`,
        marker,
        `INSERT INTO "people_notes" ("id", "body") VALUES (1, 'keep this: ${marker} intact');`,
        marker,
        `CREATE FUNCTION people_touch() RETURNS integer AS $$`,
        `BEGIN`,
        `  -- ${marker}`,
        `  PERFORM 1;`,
        `  RETURN 42;`,
        `END;`,
        `$$ LANGUAGE plpgsql;`,
        marker,
        `/* block comment with ${marker} */`,
        `CREATE TABLE "odd${marker}name" ("id" integer);`,
      ].join("\n");
      const intact = migration("people", 0, { sql, tag: "0000_intact" });

      const result = await applyModuleMigrations({
        aggregate: aggregate({ id: "people", migrations: [intact] }),
        database: createPostgresMigrationDatabase(database.url),
      });
      expect(result.appliedCount).toBe(1);

      // The literal survived byte-for-byte, so the marker was never treated as a split point.
      const stored = await safeQuery<{ body: string }>(
        database.client,
        `select body from "people_notes" where id = 1`,
      );
      expect(stored.rows[0]?.body).toBe(`keep this: ${marker} intact`);

      // The dollar-quoted body executed as one function, not as fragments.
      const called = await safeQuery<{ result: number }>(
        database.client,
        `select people_touch() as result`,
      );
      expect(Number(called.rows[0]?.result)).toBe(42);

      // The quoted identifier kept the marker in its name.
      const relations = await safeQuery<{ table_name: string }>(
        database.client,
        `select table_name from information_schema.tables
         where table_schema = 'public' and table_name like 'odd%'`,
      );
      expect(relations.rows.map((row) => row.table_name)).toEqual([`odd${marker}name`]);
    });
  });

  it("rolls back the whole multi-statement migration when a later statement fails", async () => {
    await withIsolatedDatabase("intact_rollback", async (database) => {
      const failing = migration("people", 0, {
        sql: [
          `CREATE TABLE "people_partial" ("id" integer primary key);`,
          `--> statement-breakpoint`,
          `INSERT INTO "people_partial" ("id") VALUES (1);`,
          `--> statement-breakpoint`,
          `INSERT INTO "people_partial" ("id") VALUES (1);`,
        ].join("\n"),
        tag: "0000_rollback",
      });

      const error = await captureRunnerError(
        applyModuleMigrations({
          aggregate: aggregate({ id: "people", migrations: [failing] }),
          database: createPostgresMigrationDatabase(database.url),
        }),
        "MIGRATION_EXECUTION_FAILED",
      );
      expect(error.migrationIdentity).toBe(failing.identity);

      // The earlier statements in the same batch rolled back with it.
      const relations = await safeQuery(
        database.client,
        `select table_name from information_schema.tables
         where table_schema = 'public' and table_name = 'people_partial'`,
      );
      expect(relations.rows).toHaveLength(0);
    });
  });
});