Ekklesia/core/src/migrations/runner.ts

import type {
  DiscoveredModuleMigration,
  ModuleMigrationAggregate,
} from "../modules/migration-discovery.js";
import { CORE_MIGRATION_SOURCE_ID } from "./core-source.js";

const DEFAULT_MIGRATION_LOCK_TIMEOUT_MS = 10_000;

export type MigrationRunnerErrorCode =
  | "MIGRATION_APPROVAL_LOOKUP_FAILED"
  | "MIGRATION_APPROVAL_REQUIRED"
  | "MIGRATION_CHECKSUM_MISMATCH"
  | "MIGRATION_CLEANUP_FAILED"
  | "MIGRATION_CONFIGURATION_INVALID"
  | "MIGRATION_CONNECTION_FAILED"
  | "MIGRATION_EXECUTION_FAILED"
  | "MIGRATION_HISTORY_INVALID"
  | "MIGRATION_HISTORY_READ_FAILED"
  | "MIGRATION_LOCK_FAILED"
  | "MIGRATION_LOCK_TIMEOUT"
  | "MIGRATION_UNKNOWN_APPLIED";

export interface AppliedMigrationRecord {
  readonly appliedAt: string;
  readonly appliedOrder: number;
  readonly checksum: string;
  readonly identity: string;
  readonly index: number;
  readonly moduleId: string;
  readonly tag: string;
}

export interface MigrationHistoryInsert {
  readonly checksum: string;
  readonly identity: string;
  readonly index: number;
  readonly moduleId: string;
  readonly tag: string;
}

export interface MigrationTransaction {
  ensureHistoryTable(): Promise<void>;
  /**
   * Executes one migration's **complete** SQL exactly as packaged, in a single call, inside the
   * caller's transaction. Never a fragment: the runner does not split migration SQL, so the
   * implementation must pass the string through to PostgreSQL unmodified.
   */
  executeMigrationSql(sql: string): Promise<void>;
  insertAppliedMigration(migration: MigrationHistoryInsert): Promise<void>;
}

export interface MigrationDatabaseSession {
  acquireMigrationLock(timeoutMs: number): Promise<boolean>;
  close(): Promise<void>;
  readAppliedMigrations(): Promise<readonly AppliedMigrationRecord[]>;
  releaseMigrationLock(): Promise<void>;
  transaction<T>(operation: (transaction: MigrationTransaction) => Promise<T>): Promise<T>;
}

export interface MigrationDatabase {
  openSession(): Promise<MigrationDatabaseSession>;
}

export interface MigrationApproval {
  readonly checksum: string;
  readonly identity: string;
  readonly moduleId: string;
}

export interface MigrationApprovalProvider {
  getApprovals(required: readonly MigrationApproval[]): Promise<readonly MigrationApproval[]>;
}

export const denyAllMigrationApprovals: MigrationApprovalProvider = Object.freeze({
  getApprovals: async (): Promise<readonly MigrationApproval[]> => Object.freeze([]),
});

export type MigrationPlanStatus = "applied" | "approval-blocked" | "pending";

export interface MigrationPlanEntry {
  readonly approvalRequired: boolean;
  readonly checksum: string;
  readonly identity: string;
  readonly index: number;
  readonly moduleId: string;
  readonly status: MigrationPlanStatus;
  readonly tag: string;
}

/**
 * Applied history kept for a module that is not installed (ADR 0048): the module id and how many of
 * its migrations were applied. It is reported, never matched against an inventory, and never a
 * reason to refuse planning or applying anything else.
 */
export interface RetainedMigrationHistory {
  readonly appliedCount: number;
  readonly moduleId: string;
}

export interface MigrationPlanResult {
  readonly appliedCount: number;
  readonly approvalRequiredCount: number;
  readonly entries: readonly MigrationPlanEntry[];
  readonly pendingCount: number;
  /** Absent modules' applied history, ordered by module id; empty when every applied migration
   * belongs to Core or an installed module. */
  readonly retained: readonly RetainedMigrationHistory[];
}

export interface AppliedMigrationResult {
  readonly checksum: string;
  readonly identity: string;
  readonly index: number;
  readonly moduleId: string;
  readonly tag: string;
}

export interface MigrationApplyResult {
  readonly applied: readonly AppliedMigrationResult[];
  readonly appliedCount: number;
  readonly noOp: boolean;
  /** The same report as `MigrationPlanResult.retained`, read under the migration lock. */
  readonly retained: readonly RetainedMigrationHistory[];
}

export interface PlanModuleMigrationsOptions {
  readonly aggregate: ModuleMigrationAggregate;
  readonly database: MigrationDatabase;
}

export interface ApplyModuleMigrationsOptions extends PlanModuleMigrationsOptions {
  readonly approvals?: MigrationApprovalProvider;
  readonly lockTimeoutMs?: number;
}

const errorMessages: Readonly<Record<MigrationRunnerErrorCode, string>> = Object.freeze({
  MIGRATION_APPROVAL_LOOKUP_FAILED: "Migration approval lookup failed.",
  MIGRATION_APPROVAL_REQUIRED: "A pending migration requires an exact destructive-change approval.",
  MIGRATION_CHECKSUM_MISMATCH: "An applied migration checksum does not match packaged inventory.",
  MIGRATION_CLEANUP_FAILED: "Migration database cleanup failed.",
  MIGRATION_CONFIGURATION_INVALID: "Migration configuration is invalid.",
  MIGRATION_CONNECTION_FAILED: "The migration database connection failed.",
  MIGRATION_EXECUTION_FAILED: "Migration execution failed.",
  MIGRATION_HISTORY_INVALID: "Applied migration history is invalid.",
  MIGRATION_HISTORY_READ_FAILED: "Applied migration history could not be read.",
  MIGRATION_LOCK_FAILED: "The Ekklesia migration lock operation failed.",
  MIGRATION_LOCK_TIMEOUT: "Timed out while waiting for the Ekklesia migration lock.",
  MIGRATION_UNKNOWN_APPLIED:
    "Applied migration history contains an identity absent from packaged inventory.",
});

export class MigrationRunnerError extends Error {
  readonly code: MigrationRunnerErrorCode;
  readonly migrationIdentity?: string;

  constructor(code: MigrationRunnerErrorCode, migrationIdentity?: string) {
    const suffix = migrationIdentity === undefined ? "" : ` Migration: ${migrationIdentity}.`;
    super(`${errorMessages[code]}${suffix}`);
    this.name = "MigrationRunnerError";
    this.code = code;
    if (migrationIdentity !== undefined) {
      this.migrationIdentity = migrationIdentity;
    }
    this.stack = `${this.name}: ${this.message}`;
    Object.freeze(this);
  }
}

function runnerError(
  code: MigrationRunnerErrorCode,
  migrationIdentity?: string,
): MigrationRunnerError {
  return new MigrationRunnerError(code, migrationIdentity);
}

function freezeHistoryInsert(migration: DiscoveredModuleMigration): MigrationHistoryInsert {
  return Object.freeze({
    checksum: migration.checksum,
    identity: migration.identity,
    index: migration.index,
    moduleId: migration.moduleId,
    tag: migration.tag,
  });
}

function freezeResultMigration(migration: DiscoveredModuleMigration): AppliedMigrationResult {
  return Object.freeze({
    checksum: migration.checksum,
    identity: migration.identity,
    index: migration.index,
    moduleId: migration.moduleId,
    tag: migration.tag,
  });
}

function knownMigration(
  inventory: ReadonlyMap<string, DiscoveredModuleMigration>,
  identity: unknown,
): DiscoveredModuleMigration | undefined {
  return typeof identity === "string" ? inventory.get(identity) : undefined;
}

function validateHistoryRecordShape(record: AppliedMigrationRecord): void {
  if (
    typeof record !== "object" ||
    record === null ||
    typeof record.appliedAt !== "string" ||
    record.appliedAt.length === 0 ||
    Number.isNaN(Date.parse(record.appliedAt)) ||
    !Number.isSafeInteger(record.appliedOrder) ||
    record.appliedOrder < 1 ||
    typeof record.checksum !== "string" ||
    typeof record.identity !== "string" ||
    !Number.isSafeInteger(record.index) ||
    record.index < 0 ||
    typeof record.moduleId !== "string" ||
    typeof record.tag !== "string"
  ) {
    throw runnerError("MIGRATION_HISTORY_INVALID");
  }
}

interface ValidatedAppliedHistory {
  /** Identities of applied migrations that belong to Core or an installed module. */
  readonly applied: ReadonlySet<string>;
  readonly retained: readonly RetainedMigrationHistory[];
}

/**
 * Every module id this inventory treats as installed: each discovered module, whether or not it
 * owns migrations, every module that contributed migrations, and always Core. A module counts as
 * absent only when it is none of these, so a module that is installed but no longer ships its old
 * migrations stays strict, and Core's history can never be classified as an absent module's.
 */
function installedMigrationSources(aggregate: ModuleMigrationAggregate): ReadonlySet<string> {
  return new Set<string>([
    CORE_MIGRATION_SOURCE_ID,
    ...aggregate.discoveredModules.map((module) => module.manifest.id),
    ...aggregate.modules.map((module) => module.moduleId),
  ]);
}

function freezeRetained(counts: ReadonlyMap<string, number>): readonly RetainedMigrationHistory[] {
  return Object.freeze(
    [...counts.entries()]
      .sort(([left], [right]) => (left < right ? -1 : left > right ? 1 : 0))
      .map(([moduleId, appliedCount]) => Object.freeze({ appliedCount, moduleId })),
  );
}

function validateAppliedHistoryUnchecked(
  aggregate: ModuleMigrationAggregate,
  history: readonly AppliedMigrationRecord[],
): ValidatedAppliedHistory {
  if (!Array.isArray(history)) {
    throw runnerError("MIGRATION_HISTORY_INVALID");
  }

  const inventory = new Map(
    aggregate.migrations.map((migration) => [migration.identity, migration]),
  );
  const installed = installedMigrationSources(aggregate);
  const applied = new Set<string>();
  const retained = new Map<string, number>();
  const seenOrders = new Set<number>();
  const seenIdentities = new Set<string>();
  const seenModuleIndexes = new Set<string>();
  const seenModuleTags = new Set<string>();
  let previousOrder = 0;

  for (const record of history) {
    validateHistoryRecordShape(record);
    if (
      record.appliedOrder <= previousOrder ||
      seenOrders.has(record.appliedOrder) ||
      seenIdentities.has(record.identity) ||
      seenModuleIndexes.has(`${record.moduleId}\u0000${record.index}`) ||
      seenModuleTags.has(`${record.moduleId}\u0000${record.tag}`)
    ) {
      throw runnerError("MIGRATION_HISTORY_INVALID");
    }
    previousOrder = record.appliedOrder;
    seenOrders.add(record.appliedOrder);
    seenIdentities.add(record.identity);
    seenModuleIndexes.add(`${record.moduleId}\u0000${record.index}`);
    seenModuleTags.add(`${record.moduleId}\u0000${record.tag}`);

    const migration = knownMigration(inventory, record.identity);
    if (!migration) {
      if (installed.has(record.moduleId)) {
        // Core, or a module that is installed now: its history is strict, exactly as ADR 0009
        // established. A migration it no longer ships is a downgrade or a rewritten package.
        throw runnerError("MIGRATION_UNKNOWN_APPLIED");
      }
      // An absent module's history (ADR 0048): retained and reported, never repaired, deleted, or
      // matched against anything. It must still name its own module and tag, so a corrupt row is
      // never mistaken for one.
      if (record.identity !== `${record.moduleId}:${record.tag}`) {
        throw runnerError("MIGRATION_HISTORY_INVALID");
      }
      retained.set(record.moduleId, (retained.get(record.moduleId) ?? 0) + 1);
      continue;
    }
    if (
      record.moduleId !== migration.moduleId ||
      record.tag !== migration.tag ||
      record.index !== migration.index
    ) {
      throw runnerError("MIGRATION_HISTORY_INVALID", migration.identity);
    }
    if (record.checksum !== migration.checksum) {
      throw runnerError("MIGRATION_CHECKSUM_MISMATCH", migration.identity);
    }
    applied.add(record.identity);
  }

  for (const module of aggregate.modules) {
    const moduleHistory = history.filter((record) => record.moduleId === module.moduleId);
    for (const [position, record] of moduleHistory.entries()) {
      if (record.identity !== module.migrations[position]?.identity) {
        const migration = knownMigration(inventory, record.identity);
        throw runnerError("MIGRATION_HISTORY_INVALID", migration?.identity);
      }
    }
  }

  return Object.freeze({ applied, retained: freezeRetained(retained) });
}

function validateAppliedHistory(
  aggregate: ModuleMigrationAggregate,
  history: readonly AppliedMigrationRecord[],
): ValidatedAppliedHistory {
  try {
    return validateAppliedHistoryUnchecked(aggregate, history);
  } catch (error) {
    if (error instanceof MigrationRunnerError) {
      throw error;
    }
    throw runnerError("MIGRATION_HISTORY_INVALID");
  }
}

function planFromHistory(
  aggregate: ModuleMigrationAggregate,
  history: readonly AppliedMigrationRecord[],
): MigrationPlanResult {
  const { applied, retained } = validateAppliedHistory(aggregate, history);
  const entries = Object.freeze(
    aggregate.migrations.map((migration): MigrationPlanEntry => {
      const isApplied = applied.has(migration.identity);
      const status: MigrationPlanStatus = isApplied
        ? "applied"
        : migration.approvalRequired
          ? "approval-blocked"
          : "pending";
      return Object.freeze({
        approvalRequired: migration.approvalRequired,
        checksum: migration.checksum,
        identity: migration.identity,
        index: migration.index,
        moduleId: migration.moduleId,
        status,
        tag: migration.tag,
      });
    }),
  );
  const appliedCount = entries.filter((entry) => entry.status === "applied").length;
  const approvalRequiredCount = entries.filter(
    (entry) => entry.status === "approval-blocked",
  ).length;
  return Object.freeze({
    appliedCount,
    approvalRequiredCount,
    entries,
    pendingCount: entries.length - appliedCount,
    retained,
  });
}

async function closeSession(
  session: MigrationDatabaseSession,
  primaryError: unknown,
): Promise<void> {
  try {
    await session.close();
  } catch {
    if (primaryError === undefined) {
      throw runnerError("MIGRATION_CLEANUP_FAILED");
    }
  }
}

export async function planModuleMigrations(
  options: PlanModuleMigrationsOptions,
): Promise<MigrationPlanResult> {
  let session: MigrationDatabaseSession;
  try {
    session = await options.database.openSession();
  } catch {
    throw runnerError("MIGRATION_CONNECTION_FAILED");
  }

  let primaryError: unknown;
  try {
    let history: readonly AppliedMigrationRecord[];
    try {
      history = await session.readAppliedMigrations();
    } catch {
      throw runnerError("MIGRATION_HISTORY_READ_FAILED");
    }
    return planFromHistory(options.aggregate, history);
  } catch (error) {
    primaryError = error;
    throw error;
  } finally {
    await closeSession(session, primaryError);
  }
}

function validateLockTimeout(lockTimeoutMs: number): void {
  if (!Number.isSafeInteger(lockTimeoutMs) || lockTimeoutMs <= 0) {
    throw runnerError("MIGRATION_CONFIGURATION_INVALID");
  }
}

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

async function preflightApprovals(
  pending: readonly DiscoveredModuleMigration[],
  provider: MigrationApprovalProvider,
): Promise<void> {
  const required = Object.freeze(
    pending.filter((migration) => migration.approvalRequired).map(approvalRequest),
  );
  if (required.length === 0) {
    return;
  }

  let supplied: readonly MigrationApproval[];
  try {
    supplied = await provider.getApprovals(required);
  } catch {
    throw runnerError("MIGRATION_APPROVAL_LOOKUP_FAILED");
  }
  const approvalKeys = new Set<string>();
  try {
    if (!Array.isArray(supplied)) {
      throw new TypeError("Invalid migration approval response.");
    }
    for (const approval of supplied) {
      if (
        typeof approval === "object" &&
        approval !== null &&
        typeof approval.moduleId === "string" &&
        typeof approval.identity === "string" &&
        typeof approval.checksum === "string"
      ) {
        approvalKeys.add(
          `${approval.moduleId}\u0000${approval.identity}\u0000${approval.checksum}`,
        );
      }
    }
  } catch {
    throw runnerError("MIGRATION_APPROVAL_LOOKUP_FAILED");
  }

  for (const request of required) {
    const key = `${request.moduleId}\u0000${request.identity}\u0000${request.checksum}`;
    if (!approvalKeys.has(key)) {
      throw runnerError("MIGRATION_APPROVAL_REQUIRED", request.identity);
    }
  }
}

async function cleanupApplySession(
  session: MigrationDatabaseSession,
  lockAcquired: boolean,
  primaryError: unknown,
): Promise<void> {
  let cleanupFailed = false;
  if (lockAcquired) {
    try {
      await session.releaseMigrationLock();
    } catch {
      cleanupFailed = true;
    }
  }
  try {
    await session.close();
  } catch {
    cleanupFailed = true;
  }
  if (cleanupFailed && primaryError === undefined) {
    throw runnerError("MIGRATION_CLEANUP_FAILED");
  }
}

export async function applyModuleMigrations(
  options: ApplyModuleMigrationsOptions,
): Promise<MigrationApplyResult> {
  const lockTimeoutMs = options.lockTimeoutMs ?? DEFAULT_MIGRATION_LOCK_TIMEOUT_MS;
  validateLockTimeout(lockTimeoutMs);

  let session: MigrationDatabaseSession;
  try {
    session = await options.database.openSession();
  } catch {
    throw runnerError("MIGRATION_CONNECTION_FAILED");
  }

  let lockAcquired = false;
  let primaryError: unknown;
  try {
    try {
      lockAcquired = await session.acquireMigrationLock(lockTimeoutMs);
    } catch {
      throw runnerError("MIGRATION_LOCK_FAILED");
    }
    if (!lockAcquired) {
      throw runnerError("MIGRATION_LOCK_TIMEOUT");
    }

    let history: readonly AppliedMigrationRecord[];
    try {
      history = await session.readAppliedMigrations();
    } catch {
      throw runnerError("MIGRATION_HISTORY_READ_FAILED");
    }
    const { applied, retained } = validateAppliedHistory(options.aggregate, history);
    const pending = Object.freeze(
      options.aggregate.migrations.filter((migration) => !applied.has(migration.identity)),
    );
    await preflightApprovals(pending, options.approvals ?? denyAllMigrationApprovals);

    if (pending.length === 0) {
      return Object.freeze({ applied: Object.freeze([]), appliedCount: 0, noOp: true, retained });
    }

    let executingIdentity: string | undefined;
    try {
      await session.transaction(async (transaction) => {
        await transaction.ensureHistoryTable();
        for (const migration of pending) {
          executingIdentity = migration.identity;
          // PostgreSQL parses the batch, not this runner. Drizzle's `--> statement-breakpoint`
          // lines are ordinary `--` comments to the server, so sending the migration whole keeps
          // the marker's bytes meaningful wherever they legitimately appear — inside a string
          // literal, a dollar-quoted function body, a quoted identifier, or a comment — while a
          // literal `split` on those bytes tore such a migration into fragments that were no
          // longer valid SQL. Deliberately not a homegrown SQL lexer and not a new parser
          // dependency: the database already has the only correct parser.
          await transaction.executeMigrationSql(migration.sql);
          await transaction.insertAppliedMigration(freezeHistoryInsert(migration));
        }
      });
    } catch {
      throw runnerError("MIGRATION_EXECUTION_FAILED", executingIdentity);
    }

    const resultMigrations = Object.freeze(pending.map(freezeResultMigration));
    return Object.freeze({
      applied: resultMigrations,
      appliedCount: resultMigrations.length,
      noOp: false,
      retained,
    });
  } catch (error) {
    primaryError = error;
    throw error;
  } finally {
    await cleanupApplySession(session, lockAcquired, primaryError);
  }
}