feat: add transactional module migration runner

core/src/migrations/runner.ts

@@ -0,0 +1,509 @@
+import type {
+  DiscoveredModuleMigration,
+  ModuleMigrationAggregate,
+} from "../modules/migration-discovery.js";
+
+const DEFAULT_MIGRATION_LOCK_TIMEOUT_MS = 10_000;
+const DRIZZLE_STATEMENT_BREAKPOINT = "--> statement-breakpoint";
+
+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>;
+  executeStatement(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;
+}
+
+export interface MigrationPlanResult {
+  readonly appliedCount: number;
+  readonly approvalRequiredCount: number;
+  readonly entries: readonly MigrationPlanEntry[];
+  readonly pendingCount: number;
+}
+
+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;
+}
+
+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");
+  }
+}
+
+function validateAppliedHistoryUnchecked(
+  aggregate: ModuleMigrationAggregate,
+  history: readonly AppliedMigrationRecord[],
+): ReadonlySet<string> {
+  if (!Array.isArray(history)) {
+    throw runnerError("MIGRATION_HISTORY_INVALID");
+  }
+
+  const inventory = new Map(
+    aggregate.migrations.map((migration) => [migration.identity, migration]),
+  );
+  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) {
+      throw runnerError("MIGRATION_UNKNOWN_APPLIED");
+    }
+    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);
+    }
+  }
+
+  for (const module of aggregate.modules) {
+    const applied = history.filter((record) => record.moduleId === module.moduleId);
+    for (const [position, record] of applied.entries()) {
+      if (record.identity !== module.migrations[position]?.identity) {
+        const migration = knownMigration(inventory, record.identity);
+        throw runnerError("MIGRATION_HISTORY_INVALID", migration?.identity);
+      }
+    }
+  }
+
+  return seenIdentities;
+}
+
+function validateAppliedHistory(
+  aggregate: ModuleMigrationAggregate,
+  history: readonly AppliedMigrationRecord[],
+): ReadonlySet<string> {
+  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 = 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,
+  });
+}
+
+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 = 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 });
+    }
+
+    let executingIdentity: string | undefined;
+    try {
+      await session.transaction(async (transaction) => {
+        await transaction.ensureHistoryTable();
+        for (const migration of pending) {
+          executingIdentity = migration.identity;
+          for (const statement of migration.sql.split(DRIZZLE_STATEMENT_BREAKPOINT)) {
+            await transaction.executeStatement(statement);
+          }
+          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,
+    });
+  } catch (error) {
+    primaryError = error;
+    throw error;
+  } finally {
+    await cleanupApplySession(session, lockAcquired, primaryError);
+  }
+}