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; /** * 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; insertAppliedMigration(migration: MigrationHistoryInsert): Promise; } export interface MigrationDatabaseSession { acquireMigrationLock(timeoutMs: number): Promise; close(): Promise; readAppliedMigrations(): Promise; releaseMigrationLock(): Promise; transaction(operation: (transaction: MigrationTransaction) => Promise): Promise; } export interface MigrationDatabase { openSession(): Promise; } export interface MigrationApproval { readonly checksum: string; readonly identity: string; readonly moduleId: string; } export interface MigrationApprovalProvider { getApprovals(required: readonly MigrationApproval[]): Promise; } export const denyAllMigrationApprovals: MigrationApprovalProvider = Object.freeze({ getApprovals: async (): Promise => 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> = 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, 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; 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 { return new Set([ CORE_MIGRATION_SOURCE_ID, ...aggregate.discoveredModules.map((module) => module.manifest.id), ...aggregate.modules.map((module) => module.moduleId), ]); } function freezeRetained(counts: ReadonlyMap): 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(); const retained = new Map(); const seenOrders = new Set(); const seenIdentities = new Set(); const seenModuleIndexes = new Set(); const seenModuleTags = new Set(); 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 { try { await session.close(); } catch { if (primaryError === undefined) { throw runnerError("MIGRATION_CLEANUP_FAILED"); } } } export async function planModuleMigrations( options: PlanModuleMigrationsOptions, ): Promise { 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 { 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(); 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 { 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 { 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); } }