Merge pull request 'feat: add transactional module migration runner' (#3) from phase-3/transactional-migration-runner into main
CI / verify (push) Successful in 19s
CI / infrastructure-smoke (push) Successful in 13s

Reviewed-on: http://git.riverofglorymn.org/River_of_Glory_Church/Ekklesia/pulls/3
This commit was merged in pull request #3.
This commit is contained in:
The_River
2026-07-21 13:55:10 +00:00
22 changed files with 2589 additions and 47 deletions
+14 -3
View File
@@ -11,9 +11,11 @@ capabilities, static module manifests, deterministic automatic discovery, compil
a staged capability registry, dependency-ordered lifecycle with rollback, module-owned Drizzle table
contracts, deterministic schema aggregation, and a cloneable backend-only module template. Migration
artifacts now have module-owned Drizzle contracts, deterministic offline discovery, explicit
destructive-change declarations, and exact checksums. Transactional migration execution, identity,
HTTP routing, jobs, physical storage provisioning, surfaces, and feature modules remain later
vertical slices.
destructive-change declarations, and exact checksums. Core now provides read-only applied-history
planning and a locked, whole-batch transactional PostgreSQL runner with exact external destructive
approval matching. Live PostgreSQL runner integration coverage, production approval-provider
composition, identity, HTTP routing, jobs, physical storage provisioning, surfaces, and feature
modules remain later vertical slices.
## Quick start
@@ -25,9 +27,18 @@ cp .env.example .env
pnpm infrastructure:up
pnpm modules:validate
pnpm migrations:validate
pnpm migrations:plan
pnpm verify
```
`pnpm migrations:plan` uses `DATABASE_URL` from the uncommitted `.env` and is read-only.
`pnpm migrations:apply` uses the same validated inventory but mutates PostgreSQL under Core's
advisory lock and one transaction. Review the plan before applying. Destructive migrations are
blocked by default and require an exact externally supplied approval; there is no force flag.
The runner currently has deterministic unit coverage through a database port. Live PostgreSQL
migration-runner tests are deferred and are not yet claimed by `pnpm verify`.
Run `pnpm infrastructure:smoke` for an isolated end-to-end PostgreSQL and authenticated S3 check.
The development services bind only to loopback:
+7 -1
View File
@@ -14,6 +14,10 @@
"./modules": {
"types": "./dist/modules/index.d.ts",
"import": "./dist/modules/index.js"
},
"./migrations": {
"types": "./dist/migrations/index.d.ts",
"import": "./dist/migrations/index.js"
}
},
"scripts": {
@@ -24,11 +28,13 @@
},
"dependencies": {
"@ekklesia/contracts": "workspace:*",
"drizzle-orm": "0.45.2"
"drizzle-orm": "0.45.2",
"pg": "8.22.0"
},
"devDependencies": {
"@biomejs/biome": "2.5.3",
"@types/node": "24.13.3",
"@types/pg": "8.20.0",
"typescript": "7.0.2",
"vitest": "4.1.10"
},
+19
View File
@@ -0,0 +1,19 @@
import { formatMigrationApplyReport } from "./migration-report.js";
import { discoverRepositoryMigrations } from "./migration-command-support.js";
import { reportMigrationCommandError } from "./migration-command-errors.js";
import { applyModuleMigrations, createPostgresMigrationDatabase } from "../migrations/index.js";
async function main(): Promise<void> {
const aggregate = await discoverRepositoryMigrations();
const database = createPostgresMigrationDatabase(process.env.DATABASE_URL);
const result = await applyModuleMigrations({ aggregate, database });
for (const line of formatMigrationApplyReport(result)) {
console.log(line);
}
}
main().catch((error: unknown) => {
reportMigrationCommandError(error);
process.exitCode = 1;
});
+37
View File
@@ -0,0 +1,37 @@
import { MigrationRunnerError, PostgresMigrationDatabaseError } from "../migrations/index.js";
import { ModuleDiscoveryError, ModuleMigrationDiscoveryError } from "../modules/index.js";
export function reportMigrationCommandError(error: unknown): void {
if (error instanceof ModuleDiscoveryError) {
console.error(error.message);
for (const issue of error.issues) {
const issuePath = issue.path ? ` (${issue.path})` : "";
console.error(`[${issue.code}] ${issue.moduleDirectory}${issuePath}: ${issue.message}`);
}
return;
}
if (error instanceof ModuleMigrationDiscoveryError) {
console.error(error.message);
for (const issue of error.issues) {
const issuePath = issue.path ? ` (${issue.path})` : "";
console.error(`[${issue.code}] ${issue.moduleId}${issuePath}: ${issue.message}`);
}
return;
}
if (error instanceof MigrationRunnerError) {
console.error(`[${error.code}] ${error.message}`);
return;
}
if (
error instanceof PostgresMigrationDatabaseError &&
error.code === "POSTGRES_MIGRATION_CONFIGURATION_INVALID"
) {
console.error("[MIGRATION_CONFIGURATION_INVALID] Migration configuration is invalid.");
return;
}
console.error("Migration command failed unexpectedly.");
}
+30
View File
@@ -0,0 +1,30 @@
import { readFile } from "node:fs/promises";
import path from "node:path";
import { discoverModuleMigrations, type ModuleMigrationAggregate } from "../modules/index.js";
interface PackageVersion {
readonly version?: unknown;
}
async function readPackageVersion(packagePath: string): Promise<string> {
const packageData = JSON.parse(await readFile(packagePath, "utf8")) as PackageVersion;
if (typeof packageData.version !== "string") {
throw new TypeError("A required workspace package does not declare a version.");
}
return packageData.version;
}
export async function discoverRepositoryMigrations(
repositoryRoot = process.cwd(),
): Promise<ModuleMigrationAggregate> {
const [platformVersion, moduleSdkVersion] = await Promise.all([
readPackageVersion(path.join(repositoryRoot, "package.json")),
readPackageVersion(path.join(repositoryRoot, "packages/module-sdk/package.json")),
]);
return discoverModuleMigrations({
moduleSdkVersion,
modulesDirectory: path.join(repositoryRoot, "modules"),
platformVersion,
});
}
+34
View File
@@ -1,4 +1,5 @@
import type { ModuleMigrationAggregate } from "../modules/index.js";
import type { MigrationApplyResult, MigrationPlanResult } from "../migrations/index.js";
function plural(count: number, singular: string): string {
return count === 1 ? singular : `${singular}s`;
@@ -26,3 +27,36 @@ export function formatMigrationValidationReport(
return Object.freeze(lines);
}
export function formatMigrationPlanReport(plan: MigrationPlanResult): readonly string[] {
const readyCount = plan.pendingCount - plan.approvalRequiredCount;
const lines = [
`Migration plan (${plan.appliedCount} applied, ${readyCount} pending, ${plan.approvalRequiredCount} approval-blocked):`,
];
for (const entry of plan.entries) {
lines.push(`- [${entry.status}] ${entry.identity} sha256=${entry.checksum}`);
}
if (plan.entries.length === 0) {
lines.push("- No installed module migrations.");
}
lines.push(
"Informational only: this plan is read-only and may become stale; apply revalidates history under the migration lock.",
);
return Object.freeze(lines);
}
export function formatMigrationApplyReport(result: MigrationApplyResult): readonly string[] {
if (result.noOp) {
return Object.freeze(["Migration apply completed: database is already current (no changes)."]);
}
const lines = [
`Migration apply completed (${result.appliedCount} ${plural(result.appliedCount, "migration")} applied):`,
];
for (const migration of result.applied) {
lines.push(`- ${migration.identity} sha256=${migration.checksum}`);
}
return Object.freeze(lines);
}
+19
View File
@@ -0,0 +1,19 @@
import { formatMigrationPlanReport } from "./migration-report.js";
import { discoverRepositoryMigrations } from "./migration-command-support.js";
import { reportMigrationCommandError } from "./migration-command-errors.js";
import { createPostgresMigrationDatabase, planModuleMigrations } from "../migrations/index.js";
async function main(): Promise<void> {
const aggregate = await discoverRepositoryMigrations();
const database = createPostgresMigrationDatabase(process.env.DATABASE_URL);
const plan = await planModuleMigrations({ aggregate, database });
for (const line of formatMigrationPlanReport(plan)) {
console.log(line);
}
}
main().catch((error: unknown) => {
reportMigrationCommandError(error);
process.exitCode = 1;
});
+3 -29
View File
@@ -1,35 +1,9 @@
import { readFile } from "node:fs/promises";
import path from "node:path";
import { formatMigrationValidationReport } from "./migration-report.js";
import {
discoverModuleMigrations,
ModuleDiscoveryError,
ModuleMigrationDiscoveryError,
} from "../modules/index.js";
interface PackageVersion {
version?: unknown;
}
async function readPackageVersion(packagePath: string): Promise<string> {
const packageData = JSON.parse(await readFile(packagePath, "utf8")) as PackageVersion;
if (typeof packageData.version !== "string") {
throw new TypeError("A required workspace package does not declare a version.");
}
return packageData.version;
}
import { ModuleDiscoveryError, ModuleMigrationDiscoveryError } from "../modules/index.js";
import { discoverRepositoryMigrations } from "./migration-command-support.js";
async function main(): Promise<void> {
const repositoryRoot = process.cwd();
const [platformVersion, moduleSdkVersion] = await Promise.all([
readPackageVersion(path.join(repositoryRoot, "package.json")),
readPackageVersion(path.join(repositoryRoot, "packages/module-sdk/package.json")),
]);
const aggregate = await discoverModuleMigrations({
moduleSdkVersion,
modulesDirectory: path.join(repositoryRoot, "modules"),
platformVersion,
});
const aggregate = await discoverRepositoryMigrations();
for (const line of formatMigrationValidationReport(aggregate)) {
console.log(line);
+1
View File
@@ -1 +1,2 @@
export * from "./modules/index.js";
export * from "./migrations/index.js";
+2
View File
@@ -0,0 +1,2 @@
export * from "./postgres-database.js";
export * from "./runner.js";
+466
View File
@@ -0,0 +1,466 @@
import { Client, type QueryResultRow } from "pg";
import type {
AppliedMigrationRecord,
MigrationDatabase,
MigrationDatabaseSession,
MigrationHistoryInsert,
MigrationTransaction,
} from "./runner.js";
const historyTableSchema = "public";
const historyTableName = "ekklesia_core_migrations";
const advisoryLockNamespace = 1_162_562_380;
const advisoryLockKey = 1_296_648_018;
const advisoryLockPollIntervalMs = 50;
const createHistoryTableSql = `
CREATE TABLE IF NOT EXISTS "public"."ekklesia_core_migrations" (
"applied_order" bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
"identity" text NOT NULL UNIQUE,
"module_id" text NOT NULL,
"migration_tag" text NOT NULL,
"migration_index" integer NOT NULL,
"checksum" text NOT NULL,
"applied_at" timestamptz NOT NULL DEFAULT clock_timestamp(),
UNIQUE ("module_id", "migration_tag"),
UNIQUE ("module_id", "migration_index")
)
`;
const readHistorySql = `
SELECT
"applied_order",
"identity",
"module_id",
"migration_tag",
"migration_index",
"checksum",
"applied_at"
FROM "public"."ekklesia_core_migrations"
ORDER BY "applied_order" ASC
`;
const insertHistorySql = `
INSERT INTO "public"."ekklesia_core_migrations" (
"identity",
"module_id",
"migration_tag",
"migration_index",
"checksum"
)
VALUES ($1, $2, $3, $4, $5)
`;
export type PostgresMigrationDatabaseErrorCode =
| "POSTGRES_MIGRATION_CONFIGURATION_INVALID"
| "POSTGRES_MIGRATION_CONNECTION_FAILED"
| "POSTGRES_MIGRATION_EXECUTION_FAILED"
| "POSTGRES_MIGRATION_HISTORY_READ_FAILED"
| "POSTGRES_MIGRATION_HISTORY_WRITE_FAILED"
| "POSTGRES_MIGRATION_LOCK_FAILED"
| "POSTGRES_MIGRATION_SESSION_CLOSE_FAILED"
| "POSTGRES_MIGRATION_TRANSACTION_FAILED";
export class PostgresMigrationDatabaseError extends Error {
readonly code: PostgresMigrationDatabaseErrorCode;
constructor(code: PostgresMigrationDatabaseErrorCode, message: string) {
super(message);
this.name = "PostgresMigrationDatabaseError";
this.code = code;
this.stack = `${this.name}: ${this.message}`;
}
}
interface HistoryExistenceRow extends QueryResultRow {
readonly relation_kind: string;
}
interface HistoryRow extends QueryResultRow {
readonly applied_at: Date | string;
readonly applied_order: number | string;
readonly checksum: string;
readonly identity: string;
readonly migration_index: number;
readonly migration_tag: string;
readonly module_id: string;
}
interface AdvisoryLockRow extends QueryResultRow {
readonly acquired: boolean;
}
interface AdvisoryUnlockRow extends QueryResultRow {
readonly released: boolean;
}
function databaseError(
code: PostgresMigrationDatabaseErrorCode,
message: string,
): PostgresMigrationDatabaseError {
return new PostgresMigrationDatabaseError(code, message);
}
function validateConnectionString(connectionString: string | undefined): string {
if (typeof connectionString !== "string" || connectionString.trim() === "") {
throw databaseError(
"POSTGRES_MIGRATION_CONFIGURATION_INVALID",
"A PostgreSQL database URL is required.",
);
}
try {
const url = new URL(connectionString);
if (url.protocol !== "postgres:" && url.protocol !== "postgresql:") {
throw new Error("unsupported protocol");
}
} catch {
throw databaseError(
"POSTGRES_MIGRATION_CONFIGURATION_INVALID",
"The PostgreSQL database URL is invalid.",
);
}
return connectionString;
}
function isPostgresMigrationDatabaseError(error: unknown): error is PostgresMigrationDatabaseError {
return error instanceof PostgresMigrationDatabaseError;
}
function parseAppliedOrder(value: number | string): number | undefined {
const parsed = typeof value === "number" ? value : Number(value);
return Number.isSafeInteger(parsed) && parsed > 0 ? parsed : undefined;
}
function parseAppliedAt(value: Date | string): string | undefined {
const parsed = value instanceof Date ? value : new Date(value);
return Number.isNaN(parsed.getTime()) ? undefined : parsed.toISOString();
}
function toAppliedMigrationRecord(row: HistoryRow): AppliedMigrationRecord {
const appliedOrder = parseAppliedOrder(row.applied_order);
const appliedAt = parseAppliedAt(row.applied_at);
if (
appliedOrder === undefined ||
appliedAt === undefined ||
typeof row.identity !== "string" ||
typeof row.module_id !== "string" ||
typeof row.migration_tag !== "string" ||
!Number.isSafeInteger(row.migration_index) ||
typeof row.checksum !== "string"
) {
throw databaseError(
"POSTGRES_MIGRATION_HISTORY_READ_FAILED",
"The Core migration history could not be read.",
);
}
return Object.freeze({
appliedAt,
appliedOrder,
checksum: row.checksum,
identity: row.identity,
index: row.migration_index,
moduleId: row.module_id,
tag: row.migration_tag,
});
}
async function pause(milliseconds: number): Promise<void> {
await new Promise<void>((resolve) => setTimeout(resolve, milliseconds));
}
class PostgresMigrationDatabaseSession implements MigrationDatabaseSession {
readonly #client: Client;
#closed = false;
#lockHeld = false;
#transactionActive = false;
constructor(client: Client) {
this.#client = client;
}
async readAppliedMigrations(): Promise<readonly AppliedMigrationRecord[]> {
this.#assertOpen("POSTGRES_MIGRATION_HISTORY_READ_FAILED");
try {
const existence = await this.#client.query<HistoryExistenceRow>(
`
SELECT relation.relkind AS relation_kind
FROM pg_catalog.pg_class AS relation
INNER JOIN pg_catalog.pg_namespace AS namespace
ON namespace.oid = relation.relnamespace
WHERE namespace.nspname = $1 AND relation.relname = $2
`,
[historyTableSchema, historyTableName],
);
const relationKind = existence.rows[0]?.relation_kind;
if (relationKind === undefined) {
return Object.freeze([]);
}
if (relationKind !== "r") {
throw databaseError(
"POSTGRES_MIGRATION_HISTORY_READ_FAILED",
"The Core migration history could not be read.",
);
}
const result = await this.#client.query<HistoryRow>(readHistorySql);
return Object.freeze(result.rows.map(toAppliedMigrationRecord));
} catch (error) {
if (isPostgresMigrationDatabaseError(error)) {
throw error;
}
throw databaseError(
"POSTGRES_MIGRATION_HISTORY_READ_FAILED",
"The Core migration history could not be read.",
);
}
}
async acquireMigrationLock(timeoutMs: number): Promise<boolean> {
this.#assertOpen("POSTGRES_MIGRATION_LOCK_FAILED");
if (!Number.isFinite(timeoutMs) || timeoutMs < 0 || timeoutMs > 2_147_483_647) {
throw databaseError(
"POSTGRES_MIGRATION_LOCK_FAILED",
"The migration lock timeout is invalid.",
);
}
if (this.#lockHeld) {
return true;
}
const deadline = performance.now() + timeoutMs;
try {
while (true) {
const result = await this.#client.query<AdvisoryLockRow>(
"SELECT pg_try_advisory_lock($1::integer, $2::integer) AS acquired",
[advisoryLockNamespace, advisoryLockKey],
);
const acquired = result.rows[0]?.acquired;
if (typeof acquired !== "boolean") {
throw databaseError(
"POSTGRES_MIGRATION_LOCK_FAILED",
"The migration advisory lock could not be acquired.",
);
}
if (acquired) {
this.#lockHeld = true;
return true;
}
const remaining = deadline - performance.now();
if (remaining <= 0) {
return false;
}
await pause(Math.min(advisoryLockPollIntervalMs, remaining));
}
} catch (error) {
if (isPostgresMigrationDatabaseError(error)) {
throw error;
}
throw databaseError(
"POSTGRES_MIGRATION_LOCK_FAILED",
"The migration advisory lock could not be acquired.",
);
}
}
async releaseMigrationLock(): Promise<void> {
this.#assertOpen("POSTGRES_MIGRATION_LOCK_FAILED");
if (!this.#lockHeld) {
return;
}
try {
const result = await this.#client.query<AdvisoryUnlockRow>(
"SELECT pg_advisory_unlock($1::integer, $2::integer) AS released",
[advisoryLockNamespace, advisoryLockKey],
);
if (result.rows[0]?.released !== true) {
throw databaseError(
"POSTGRES_MIGRATION_LOCK_FAILED",
"The migration advisory lock could not be released.",
);
}
this.#lockHeld = false;
} catch (error) {
if (isPostgresMigrationDatabaseError(error)) {
throw error;
}
throw databaseError(
"POSTGRES_MIGRATION_LOCK_FAILED",
"The migration advisory lock could not be released.",
);
}
}
async transaction<T>(operation: (transaction: MigrationTransaction) => Promise<T>): Promise<T> {
this.#assertOpen("POSTGRES_MIGRATION_TRANSACTION_FAILED");
if (this.#transactionActive) {
throw databaseError(
"POSTGRES_MIGRATION_TRANSACTION_FAILED",
"Nested migration transactions are not supported.",
);
}
this.#transactionActive = true;
try {
await this.#client.query("BEGIN");
} catch {
this.#transactionActive = false;
throw databaseError(
"POSTGRES_MIGRATION_TRANSACTION_FAILED",
"The migration transaction could not be started.",
);
}
const transaction = Object.freeze<MigrationTransaction>({
ensureHistoryTable: async () => this.#ensureHistoryTable(),
executeStatement: async (sql) => this.#executeStatement(sql),
insertAppliedMigration: async (migration) => this.#insertAppliedMigration(migration),
});
try {
const result = await operation(transaction);
await this.#client.query("COMMIT");
return result;
} catch (error) {
try {
await this.#client.query("ROLLBACK");
} catch {
throw databaseError(
"POSTGRES_MIGRATION_TRANSACTION_FAILED",
"The migration transaction could not be rolled back.",
);
}
if (isPostgresMigrationDatabaseError(error)) {
throw error;
}
throw databaseError(
"POSTGRES_MIGRATION_TRANSACTION_FAILED",
"The migration transaction failed.",
);
} finally {
this.#transactionActive = false;
}
}
async close(): Promise<void> {
if (this.#closed) {
return;
}
let cleanupFailed = false;
if (this.#lockHeld) {
try {
const result = await this.#client.query<AdvisoryUnlockRow>(
"SELECT pg_advisory_unlock($1::integer, $2::integer) AS released",
[advisoryLockNamespace, advisoryLockKey],
);
cleanupFailed = result.rows[0]?.released !== true;
} catch {
cleanupFailed = true;
}
}
try {
await this.#client.end();
} catch {
cleanupFailed = true;
} finally {
this.#closed = true;
this.#lockHeld = false;
this.#transactionActive = false;
}
if (cleanupFailed) {
throw databaseError(
"POSTGRES_MIGRATION_SESSION_CLOSE_FAILED",
"The PostgreSQL migration session could not be closed cleanly.",
);
}
}
async #ensureHistoryTable(): Promise<void> {
this.#assertTransactionActive("POSTGRES_MIGRATION_HISTORY_WRITE_FAILED");
try {
await this.#client.query(createHistoryTableSql);
} catch {
throw databaseError(
"POSTGRES_MIGRATION_HISTORY_WRITE_FAILED",
"The Core migration history table could not be created.",
);
}
}
async #executeStatement(sql: string): Promise<void> {
this.#assertTransactionActive("POSTGRES_MIGRATION_EXECUTION_FAILED");
try {
await this.#client.query(sql);
} catch {
throw databaseError(
"POSTGRES_MIGRATION_EXECUTION_FAILED",
"A module migration statement failed.",
);
}
}
async #insertAppliedMigration(migration: MigrationHistoryInsert): Promise<void> {
this.#assertTransactionActive("POSTGRES_MIGRATION_HISTORY_WRITE_FAILED");
try {
await this.#client.query(insertHistorySql, [
migration.identity,
migration.moduleId,
migration.tag,
migration.index,
migration.checksum,
]);
} catch {
throw databaseError(
"POSTGRES_MIGRATION_HISTORY_WRITE_FAILED",
"The applied migration could not be recorded.",
);
}
}
#assertOpen(code: PostgresMigrationDatabaseErrorCode): void {
if (this.#closed) {
throw databaseError(code, "The PostgreSQL migration session is closed.");
}
}
#assertTransactionActive(code: PostgresMigrationDatabaseErrorCode): void {
if (!this.#transactionActive) {
throw databaseError(code, "A migration transaction is not active.");
}
}
}
export function createPostgresMigrationDatabase(
connectionString: string | undefined,
): MigrationDatabase {
const validatedConnectionString = validateConnectionString(connectionString);
return Object.freeze({
async openSession(): Promise<MigrationDatabaseSession> {
const client = new Client({ connectionString: validatedConnectionString });
client.on("error", () => undefined);
try {
await client.connect();
} catch {
try {
await client.end();
} catch {
// The public error stays fixed even when driver cleanup also fails.
}
throw databaseError(
"POSTGRES_MIGRATION_CONNECTION_FAILED",
"The PostgreSQL migration connection could not be established.",
);
}
return new PostgresMigrationDatabaseSession(client);
},
});
}
+509
View File
@@ -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);
}
}
+96
View File
@@ -0,0 +1,96 @@
import { describe, expect, it } from "vitest";
import {
formatMigrationApplyReport,
formatMigrationPlanReport,
} from "../src/cli/migration-report.js";
import type { MigrationApplyResult, MigrationPlanResult } from "../src/migrations/index.js";
describe("migration command reports", () => {
it("reports a deterministic safe plan and its stale-plan warning", () => {
const plan: MigrationPlanResult = Object.freeze({
appliedCount: 1,
approvalRequiredCount: 1,
entries: Object.freeze([
Object.freeze({
approvalRequired: false,
checksum: "a".repeat(64),
identity: "people:0000_initial",
index: 0,
moduleId: "people",
status: "applied",
tag: "0000_initial",
}),
Object.freeze({
approvalRequired: false,
checksum: "b".repeat(64),
identity: "people:0001_add_email",
index: 1,
moduleId: "people",
status: "pending",
tag: "0001_add_email",
}),
Object.freeze({
approvalRequired: true,
checksum: "c".repeat(64),
identity: "calendar:0000_replace_events",
index: 0,
moduleId: "calendar",
status: "approval-blocked",
tag: "0000_replace_events",
}),
]),
pendingCount: 2,
});
expect(formatMigrationPlanReport(plan)).toEqual([
"Migration plan (1 applied, 1 pending, 1 approval-blocked):",
`- [applied] people:0000_initial sha256=${"a".repeat(64)}`,
`- [pending] people:0001_add_email sha256=${"b".repeat(64)}`,
`- [approval-blocked] calendar:0000_replace_events sha256=${"c".repeat(64)}`,
"Informational only: this plan is read-only and may become stale; apply revalidates history under the migration lock.",
]);
});
it("reports applied identities without SQL and handles a genuine no-op", () => {
const applied: MigrationApplyResult = Object.freeze({
applied: Object.freeze([
Object.freeze({
checksum: "d".repeat(64),
identity: "people:0000_initial",
index: 0,
moduleId: "people",
tag: "0000_initial",
}),
]),
appliedCount: 1,
noOp: false,
});
expect(formatMigrationApplyReport(applied)).toEqual([
"Migration apply completed (1 migration applied):",
`- people:0000_initial sha256=${"d".repeat(64)}`,
]);
expect(
formatMigrationApplyReport(
Object.freeze({ applied: Object.freeze([]), appliedCount: 0, noOp: true }),
),
).toEqual(["Migration apply completed: database is already current (no changes)."]);
});
it("renders an empty installed inventory explicitly", () => {
expect(
formatMigrationPlanReport(
Object.freeze({
appliedCount: 0,
approvalRequiredCount: 0,
entries: Object.freeze([]),
pendingCount: 0,
}),
),
).toEqual([
"Migration plan (0 applied, 0 pending, 0 approval-blocked):",
"- No installed module migrations.",
"Informational only: this plan is read-only and may become stale; apply revalidates history under the migration lock.",
]);
});
});
+943
View File
@@ -0,0 +1,943 @@
import { createHash } from "node:crypto";
import { describe, expect, it } from "vitest";
import {
applyModuleMigrations,
type AppliedMigrationRecord,
type MigrationApproval,
type MigrationApprovalProvider,
type MigrationDatabase,
type MigrationDatabaseSession,
type MigrationHistoryInsert,
MigrationRunnerError,
type MigrationRunnerErrorCode,
type MigrationTransaction,
planModuleMigrations,
} from "../src/migrations/runner.js";
import type {
DiscoveredModuleMigration,
ModuleMigrationAggregate,
} from "../src/modules/migration-discovery.js";
const appliedAt = "2026-07-21T12:00:00.000Z";
const hostileDatabaseError = new Error(
'postgresql://admin:[email protected]/ekklesia failed at /home/ekklesia/private.sql while running DROP TABLE "people_records"',
);
interface MigrationFixtureOptions {
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: MigrationFixtureOptions = {},
): DiscoveredModuleMigration {
const tag = options.tag ?? `${index.toString().padStart(4, "0")}_change`;
const sql = options.sql ?? `SELECT ${index + 1};\n`;
const approvalRequired = options.approvalRequired ?? false;
return Object.freeze({
approvalRequired,
breakpoints: true,
checksum: createHash("sha256").update(sql).digest("hex"),
destructive: Object.freeze(
approvalRequired
? {
mode: "approval-required" as const,
reason: "An administrator must approve this destructive fixture migration.",
}
: { mode: "none" as const },
),
file: `${tag}.sql`,
identity: `${moduleId}:${tag}`,
index,
moduleId,
sql,
tag,
timestamp: 1_700_000_000_000 + index,
});
}
function aggregate(...modules: readonly ModuleFixture[]): ModuleMigrationAggregate {
const frozenModules = Object.freeze(
modules.map((module) =>
Object.freeze({
migrations: Object.freeze([...module.migrations]),
moduleId: module.id,
}),
),
);
return Object.freeze({
discoveredModules: Object.freeze([]),
migrations: Object.freeze(frozenModules.flatMap((module) => module.migrations)),
modules: frozenModules,
});
}
function historyRecord(
source: DiscoveredModuleMigration,
appliedOrder: number,
overrides: Partial<AppliedMigrationRecord> = {},
): AppliedMigrationRecord {
return Object.freeze({
appliedAt,
appliedOrder,
checksum: source.checksum,
identity: source.identity,
index: source.index,
moduleId: source.moduleId,
tag: source.tag,
...overrides,
});
}
type FailurePoint =
| "close"
| "commit"
| "ensure-history"
| "execute"
| "history-insert"
| "lock"
| "open"
| "read-history"
| "release-lock";
interface FakeDatabaseOptions {
readonly failures?: readonly FailurePoint[];
readonly history?: readonly AppliedMigrationRecord[];
readonly historyAfterLock?: readonly AppliedMigrationRecord[];
readonly historyTableExists?: boolean;
readonly lockResult?: boolean;
readonly statementFailureIndex?: number;
}
class FakeMigrationDatabase implements MigrationDatabase {
readonly events: string[] = [];
readonly executedStatements: string[] = [];
readonly failures: ReadonlySet<FailurePoint>;
readonly historyAfterLock: readonly AppliedMigrationRecord[] | undefined;
readonly insertedMigrations: MigrationHistoryInsert[] = [];
readonly lockResult: boolean;
readonly statementFailureIndex: number | undefined;
committedHistory: AppliedMigrationRecord[];
historyTableExists: boolean;
constructor(options: FakeDatabaseOptions = {}) {
this.committedHistory = [...(options.history ?? [])];
this.failures = new Set(options.failures ?? []);
this.historyAfterLock = options.historyAfterLock;
this.historyTableExists = options.historyTableExists ?? true;
this.lockResult = options.lockResult ?? true;
this.statementFailureIndex = options.statementFailureIndex;
}
async openSession(): Promise<MigrationDatabaseSession> {
this.events.push("session:open");
if (this.failures.has("open")) {
throw hostileDatabaseError;
}
return new FakeMigrationDatabaseSession(this);
}
}
class FakeMigrationDatabaseSession implements MigrationDatabaseSession {
readonly #database: FakeMigrationDatabase;
#statementCount = 0;
constructor(database: FakeMigrationDatabase) {
this.#database = database;
}
async acquireMigrationLock(timeoutMs: number): Promise<boolean> {
this.#database.events.push(`lock:acquire:${timeoutMs}`);
if (this.#database.failures.has("lock")) {
throw hostileDatabaseError;
}
if (this.#database.lockResult && this.#database.historyAfterLock) {
this.#database.committedHistory = [...this.#database.historyAfterLock];
}
return this.#database.lockResult;
}
async close(): Promise<void> {
this.#database.events.push("session:close");
if (this.#database.failures.has("close")) {
throw hostileDatabaseError;
}
}
async readAppliedMigrations(): Promise<readonly AppliedMigrationRecord[]> {
this.#database.events.push("history:read");
if (this.#database.failures.has("read-history")) {
throw hostileDatabaseError;
}
return Object.freeze([...this.#database.committedHistory]);
}
async releaseMigrationLock(): Promise<void> {
this.#database.events.push("lock:release");
if (this.#database.failures.has("release-lock")) {
throw hostileDatabaseError;
}
}
async transaction<T>(operation: (transaction: MigrationTransaction) => Promise<T>): Promise<T> {
this.#database.events.push("transaction:begin");
const stagedHistory = [...this.#database.committedHistory];
const stagedInserts: MigrationHistoryInsert[] = [];
let stagedTableExists = this.#database.historyTableExists;
const transaction = Object.freeze<MigrationTransaction>({
ensureHistoryTable: async () => {
this.#database.events.push("history:ensure");
if (this.#database.failures.has("ensure-history")) {
throw hostileDatabaseError;
}
stagedTableExists = true;
},
executeStatement: async (sql) => {
this.#database.events.push("sql:execute");
this.#database.executedStatements.push(sql);
const statementIndex = this.#statementCount;
this.#statementCount += 1;
if (
this.#database.failures.has("execute") ||
this.#database.statementFailureIndex === statementIndex
) {
throw hostileDatabaseError;
}
},
insertAppliedMigration: async (insert) => {
this.#database.events.push(`history:insert:${insert.identity}`);
if (this.#database.failures.has("history-insert")) {
throw hostileDatabaseError;
}
const frozenInsert = Object.freeze({ ...insert });
stagedInserts.push(frozenInsert);
stagedHistory.push(
Object.freeze({
...frozenInsert,
appliedAt,
appliedOrder: stagedHistory.length + 1,
}),
);
},
});
try {
const result = await operation(transaction);
if (this.#database.failures.has("commit")) {
throw hostileDatabaseError;
}
this.#database.committedHistory = stagedHistory;
this.#database.historyTableExists = stagedTableExists;
this.#database.insertedMigrations.push(...stagedInserts);
this.#database.events.push("transaction:commit");
return result;
} catch (error) {
this.#database.events.push("transaction:rollback");
throw error;
}
}
}
class CapturingApprovalProvider implements MigrationApprovalProvider {
readonly requests: (readonly MigrationApproval[])[] = [];
readonly #response: readonly MigrationApproval[];
readonly #throwError: boolean;
constructor(response: readonly MigrationApproval[], throwError = false) {
this.#response = response;
this.#throwError = throwError;
}
async getApprovals(
required: readonly MigrationApproval[],
): Promise<readonly MigrationApproval[]> {
this.requests.push(required);
if (this.#throwError) {
throw hostileDatabaseError;
}
return this.#response;
}
}
function approvalFor(source: DiscoveredModuleMigration): MigrationApproval {
return Object.freeze({
checksum: source.checksum,
identity: source.identity,
moduleId: source.moduleId,
});
}
async function captureRunnerError(
operation: Promise<unknown>,
code: MigrationRunnerErrorCode,
): Promise<MigrationRunnerError> {
try {
await operation;
} catch (error) {
expect(error).toBeInstanceOf(MigrationRunnerError);
const runnerError = error as MigrationRunnerError;
expect(runnerError.code).toBe(code);
expect(Object.isFrozen(runnerError)).toBe(true);
return runnerError;
}
throw new Error(`Expected ${code}.`);
}
function expectNoPersistentMutation(database: FakeMigrationDatabase): void {
expect(database.events).not.toContain("transaction:begin");
expect(database.events).not.toContain("history:ensure");
expect(database.events).not.toContain("sql:execute");
expect(database.insertedMigrations).toEqual([]);
}
describe("module migration planning", () => {
it("treats a fresh database with a missing history table as empty without writing", async () => {
const initial = migration("people", 0);
const database = new FakeMigrationDatabase({ historyTableExists: false });
const result = await planModuleMigrations({
aggregate: aggregate({ id: "people", migrations: [initial] }),
database,
});
expect(result).toEqual({
appliedCount: 0,
approvalRequiredCount: 0,
entries: [
{
approvalRequired: false,
checksum: initial.checksum,
identity: initial.identity,
index: 0,
moduleId: "people",
status: "pending",
tag: initial.tag,
},
],
pendingCount: 1,
});
expect(database.events).toEqual(["session:open", "history:read", "session:close"]);
expect(database.historyTableExists).toBe(false);
expectNoPersistentMutation(database);
});
it("returns an empty frozen plan when no installed module owns migrations", async () => {
const database = new FakeMigrationDatabase({ historyTableExists: false });
const inventory = aggregate({ id: "announcements", migrations: [] });
const result = await planModuleMigrations({
aggregate: inventory,
database,
});
const applyResult = await applyModuleMigrations({ aggregate: inventory, database });
expect(result).toEqual({
appliedCount: 0,
approvalRequiredCount: 0,
entries: [],
pendingCount: 0,
});
expect(Object.isFrozen(result)).toBe(true);
expect(Object.isFrozen(result.entries)).toBe(true);
expect(applyResult).toEqual({ applied: [], appliedCount: 0, noOp: true });
expect(database.historyTableExists).toBe(false);
expect(database.events).not.toContain("transaction:begin");
});
it("reports applied, pending, and approval-blocked entries in aggregate order", async () => {
const peopleInitial = migration("people", 0, { tag: "0000_initial" });
const peopleArchive = migration("people", 1, {
approvalRequired: true,
tag: "0001_archive",
});
const calendarInitial = migration("calendar", 0, { tag: "0000_initial" });
const inventory = aggregate(
{ id: "people", migrations: [peopleInitial, peopleArchive] },
{ id: "calendar", migrations: [calendarInitial] },
);
const result = await planModuleMigrations({
aggregate: inventory,
database: new FakeMigrationDatabase({ history: [historyRecord(peopleInitial, 1)] }),
});
expect(result.entries.map((entry) => [entry.identity, entry.status])).toEqual([
[peopleInitial.identity, "applied"],
[peopleArchive.identity, "approval-blocked"],
[calendarInitial.identity, "pending"],
]);
expect(result).toMatchObject({
appliedCount: 1,
approvalRequiredCount: 1,
pendingCount: 2,
});
expect(Object.isFrozen(result)).toBe(true);
expect(Object.isFrozen(result.entries)).toBe(true);
expect(result.entries.every(Object.isFrozen)).toBe(true);
});
it("accepts historical interleaving while enforcing each module's journal prefix", async () => {
const peopleInitial = migration("people", 0);
const peopleLater = migration("people", 1);
const calendarInitial = migration("calendar", 0);
const inventory = aggregate(
{ id: "people", migrations: [peopleInitial, peopleLater] },
{ id: "calendar", migrations: [calendarInitial] },
);
const database = new FakeMigrationDatabase({
history: [
historyRecord(peopleInitial, 1),
historyRecord(calendarInitial, 2),
historyRecord(peopleLater, 3),
],
});
const result = await planModuleMigrations({ aggregate: inventory, database });
expect(result.entries.every((entry) => entry.status === "applied")).toBe(true);
expect(result.pendingCount).toBe(0);
});
it.each([
[
"a journal gap",
(_first: DiscoveredModuleMigration, second: DiscoveredModuleMigration) => [
historyRecord(second, 1),
],
],
[
"out-of-order module history",
(first: DiscoveredModuleMigration, second: DiscoveredModuleMigration) => [
historyRecord(second, 1),
historyRecord(first, 2),
],
],
])("rejects per-module history with %s", async (_label, makeHistory) => {
const first = migration("people", 0);
const second = migration("people", 1);
const database = new FakeMigrationDatabase({ history: makeHistory(first, second) });
await captureRunnerError(
planModuleMigrations({
aggregate: aggregate({ id: "people", migrations: [first, second] }),
database,
}),
"MIGRATION_HISTORY_INVALID",
);
expect(database.events.at(-1)).toBe("session:close");
});
it("allows gaps in database-assigned applied order", async () => {
const peopleInitial = migration("people", 0);
const peopleLater = migration("people", 1);
const calendarInitial = migration("calendar", 0);
const inventory = aggregate(
{ id: "people", migrations: [peopleInitial, peopleLater] },
{ id: "calendar", migrations: [calendarInitial] },
);
const result = await planModuleMigrations({
aggregate: inventory,
database: new FakeMigrationDatabase({
history: [
historyRecord(peopleInitial, 7),
historyRecord(calendarInitial, 40),
historyRecord(peopleLater, 99),
],
}),
});
expect(result.appliedCount).toBe(3);
expect(result.pendingCount).toBe(0);
});
it.each([
[
"duplicate identity",
(source: DiscoveredModuleMigration) => [historyRecord(source, 1), historyRecord(source, 2)],
],
[
"inconsistent known metadata",
(source: DiscoveredModuleMigration) => [historyRecord(source, 1, { moduleId: "calendar" })],
],
])("rejects %s records without repairing history", async (_label, makeHistory) => {
const initial = migration("people", 0);
const history = makeHistory(initial);
const database = new FakeMigrationDatabase({ history });
await captureRunnerError(
planModuleMigrations({
aggregate: aggregate({ id: "people", migrations: [initial] }),
database,
}),
"MIGRATION_HISTORY_INVALID",
);
expect(database.committedHistory).toEqual(history);
});
it("distinguishes exact checksum mismatches from unknown applied identities", async () => {
const initial = migration("people", 0);
const inventory = aggregate({ id: "people", migrations: [initial] });
const mismatch = new FakeMigrationDatabase({
history: [historyRecord(initial, 1, { checksum: "0".repeat(64) })],
});
const unknown = new FakeMigrationDatabase({
history: [
historyRecord(initial, 1, {
identity: "hostile:postgresql-admin-secret-private-path",
}),
],
});
const mismatchError = await captureRunnerError(
planModuleMigrations({ aggregate: inventory, database: mismatch }),
"MIGRATION_CHECKSUM_MISMATCH",
);
const unknownError = await captureRunnerError(
planModuleMigrations({ aggregate: inventory, database: unknown }),
"MIGRATION_UNKNOWN_APPLIED",
);
expect(mismatchError.migrationIdentity).toBe(initial.identity);
expect(unknownError.message).not.toContain("hostile");
expect(unknownError.message).not.toContain("private-path");
});
});
describe("module migration apply", () => {
it("acquires the lock before reading history and applies missing-history work atomically", async () => {
const initial = migration("people", 0);
const database = new FakeMigrationDatabase({ historyTableExists: false });
const result = await applyModuleMigrations({
aggregate: aggregate({ id: "people", migrations: [initial] }),
database,
});
expect(database.events).toEqual([
"session:open",
"lock:acquire:10000",
"history:read",
"transaction:begin",
"history:ensure",
"sql:execute",
`history:insert:${initial.identity}`,
"transaction:commit",
"lock:release",
"session:close",
]);
expect(database.historyTableExists).toBe(true);
expect(database.committedHistory.map((record) => record.identity)).toEqual([initial.identity]);
expect(result).toEqual({
applied: [
{
checksum: initial.checksum,
identity: initial.identity,
index: initial.index,
moduleId: initial.moduleId,
tag: initial.tag,
},
],
appliedCount: 1,
noOp: false,
});
});
it("re-reads history under the lock before deciding which migrations remain", async () => {
const initial = migration("people", 0);
const later = migration("people", 1);
const database = new FakeMigrationDatabase({
history: [],
historyAfterLock: [historyRecord(initial, 1)],
});
const result = await applyModuleMigrations({
aggregate: aggregate({ id: "people", migrations: [initial, later] }),
database,
});
expect(result.applied.map((entry) => entry.identity)).toEqual([later.identity]);
expect(database.executedStatements).toEqual([later.sql]);
expect(database.events.indexOf("lock:acquire:10000")).toBeLessThan(
database.events.indexOf("history:read"),
);
});
it("returns a genuine frozen no-op when history is already current", async () => {
const initial = migration("people", 0);
const database = new FakeMigrationDatabase({ history: [historyRecord(initial, 1)] });
const result = await applyModuleMigrations({
aggregate: aggregate({ id: "people", migrations: [initial] }),
database,
});
expect(result).toEqual({ applied: [], appliedCount: 0, noOp: true });
expect(Object.isFrozen(result)).toBe(true);
expect(Object.isFrozen(result.applied)).toBe(true);
expectNoPersistentMutation(database);
expect(database.events).toEqual([
"session:open",
"lock:acquire:10000",
"history:read",
"lock:release",
"session:close",
]);
});
it("returns a locked no-op without creating history for an inventory with no migrations", async () => {
const database = new FakeMigrationDatabase({ historyTableExists: false });
const result = await applyModuleMigrations({
aggregate: aggregate({ id: "announcements", migrations: [] }),
database,
});
expect(result).toEqual({ applied: [], appliedCount: 0, noOp: true });
expect(database.historyTableExists).toBe(false);
expect(database.events).toEqual([
"session:open",
"lock:acquire:10000",
"history:read",
"lock:release",
"session:close",
]);
});
it("applies a later migration after a valid existing prefix", async () => {
const initial = migration("people", 0);
const later = migration("people", 1);
const database = new FakeMigrationDatabase({ history: [historyRecord(initial, 1)] });
const result = await applyModuleMigrations({
aggregate: aggregate({ id: "people", migrations: [initial, later] }),
database,
});
expect(result.applied.map((entry) => entry.identity)).toEqual([later.identity]);
expect(database.committedHistory.map((record) => record.identity)).toEqual([
initial.identity,
later.identity,
]);
});
it("executes dependency-first aggregate order and native journal order in one batch", async () => {
const peopleInitial = migration("people", 0);
const peopleLater = migration("people", 1);
const calendarInitial = migration("calendar", 0);
const database = new FakeMigrationDatabase();
const result = await applyModuleMigrations({
aggregate: aggregate(
{ id: "people", migrations: [peopleInitial, peopleLater] },
{ id: "calendar", migrations: [calendarInitial] },
),
database,
});
expect(result.applied.map((entry) => entry.identity)).toEqual([
peopleInitial.identity,
peopleLater.identity,
calendarInitial.identity,
]);
expect(database.events.filter((event) => event === "transaction:begin")).toHaveLength(1);
expect(database.events.filter((event) => event === "transaction:commit")).toHaveLength(1);
expect(database.insertedMigrations.map((entry) => entry.identity)).toEqual(
result.applied.map((entry) => entry.identity),
);
});
it("preserves exact Drizzle breakpoints without naively splitting semicolons", async () => {
const sql =
'CREATE TABLE "people_records" ("id" integer);\n--> statement-breakpoint\nALTER TABLE "people_records" ADD COLUMN "name" text; SELECT 2;';
const initial = migration("people", 0, { sql });
const database = new FakeMigrationDatabase();
await applyModuleMigrations({
aggregate: aggregate({ id: "people", migrations: [initial] }),
database,
});
expect(database.executedStatements).toEqual([
'CREATE TABLE "people_records" ("id" integer);\n',
'\nALTER TABLE "people_records" ADD COLUMN "name" text; SELECT 2;',
]);
expect(database.events.indexOf(`history:insert:${initial.identity}`)).toBeGreaterThan(
database.events.lastIndexOf("sql:execute"),
);
});
it("rolls back the whole pending batch on a later SQL failure and preserves prior history", async () => {
const prior = migration("people", 0);
const pendingFirst = migration("people", 1);
const pendingSecond = migration("people", 2);
const database = new FakeMigrationDatabase({
history: [historyRecord(prior, 1)],
statementFailureIndex: 1,
});
const error = await captureRunnerError(
applyModuleMigrations({
aggregate: aggregate({ id: "people", migrations: [prior, pendingFirst, pendingSecond] }),
database,
}),
"MIGRATION_EXECUTION_FAILED",
);
expect(error.migrationIdentity).toBe(pendingSecond.identity);
expect(database.committedHistory).toEqual([historyRecord(prior, 1)]);
expect(database.insertedMigrations).toEqual([]);
expect(database.events).toContain(`history:insert:${pendingFirst.identity}`);
expect(database.events).toContain("transaction:rollback");
expect(database.events).not.toContain("transaction:commit");
expect(database.events.slice(-2)).toEqual(["lock:release", "session:close"]);
});
it("rolls back SQL when the matching history insertion fails", async () => {
const initial = migration("people", 0);
const database = new FakeMigrationDatabase({ failures: ["history-insert"] });
await captureRunnerError(
applyModuleMigrations({
aggregate: aggregate({ id: "people", migrations: [initial] }),
database,
}),
"MIGRATION_EXECUTION_FAILED",
);
expect(database.executedStatements).toEqual([initial.sql]);
expect(database.committedHistory).toEqual([]);
expect(database.events).toContain("transaction:rollback");
});
it("publishes no success when commit fails and still releases the session", async () => {
const initial = migration("people", 0);
const database = new FakeMigrationDatabase({ failures: ["commit"] });
await captureRunnerError(
applyModuleMigrations({
aggregate: aggregate({ id: "people", migrations: [initial] }),
database,
}),
"MIGRATION_EXECUTION_FAILED",
);
expect(database.committedHistory).toEqual([]);
expect(database.insertedMigrations).toEqual([]);
expect(database.events).toContain("transaction:rollback");
expect(database.events.slice(-2)).toEqual(["lock:release", "session:close"]);
});
it("times out without reading history or trying to release an unheld lock", async () => {
const initial = migration("people", 0);
const database = new FakeMigrationDatabase({ lockResult: false });
await captureRunnerError(
applyModuleMigrations({
aggregate: aggregate({ id: "people", migrations: [initial] }),
database,
lockTimeoutMs: 250,
}),
"MIGRATION_LOCK_TIMEOUT",
);
expect(database.events).toEqual(["session:open", "lock:acquire:250", "session:close"]);
expectNoPersistentMutation(database);
});
it("does not ask for approval for ordinary migrations", async () => {
const initial = migration("people", 0);
const approvals = new CapturingApprovalProvider([]);
const result = await applyModuleMigrations({
aggregate: aggregate({ id: "people", migrations: [initial] }),
approvals,
database: new FakeMigrationDatabase(),
});
expect(result.appliedCount).toBe(1);
expect(approvals.requests).toEqual([]);
});
it("accepts an exact destructive approval bound to module, identity, and checksum", async () => {
const destructive = migration("people", 0, { approvalRequired: true });
const approvals = new CapturingApprovalProvider([approvalFor(destructive)]);
const database = new FakeMigrationDatabase();
const result = await applyModuleMigrations({
aggregate: aggregate({ id: "people", migrations: [destructive] }),
approvals,
database,
});
expect(result.applied.map((entry) => entry.identity)).toEqual([destructive.identity]);
expect(approvals.requests).toHaveLength(1);
expect(approvals.requests[0]).toEqual([approvalFor(destructive)]);
expect(Object.isFrozen(approvals.requests[0])).toBe(true);
expect(approvals.requests[0]?.every(Object.isFrozen)).toBe(true);
});
it.each([
["missing", []],
["wildcard-like", [{ moduleId: "*", identity: "*", checksum: "*" }]],
["wrong identity", [{ moduleId: "people", identity: "people:*", checksum: "CHECKSUM" }]],
["wrong module", [{ moduleId: "calendar", identity: "IDENTITY", checksum: "CHECKSUM" }]],
["wrong checksum", [{ moduleId: "people", identity: "IDENTITY", checksum: "0".repeat(64) }]],
])("blocks a destructive migration for a %s approval before mutation", async (_label, raw) => {
const destructive = migration("people", 0, { approvalRequired: true });
const supplied = raw.map((entry) =>
Object.freeze({
checksum: entry.checksum === "CHECKSUM" ? destructive.checksum : entry.checksum,
identity: entry.identity === "IDENTITY" ? destructive.identity : entry.identity,
moduleId: entry.moduleId,
}),
);
const database = new FakeMigrationDatabase();
const error = await captureRunnerError(
applyModuleMigrations({
aggregate: aggregate({ id: "people", migrations: [destructive] }),
approvals: new CapturingApprovalProvider(supplied),
database,
}),
"MIGRATION_APPROVAL_REQUIRED",
);
expect(error.migrationIdentity).toBe(destructive.identity);
expectNoPersistentMutation(database);
expect(database.events.slice(-2)).toEqual(["lock:release", "session:close"]);
});
it("preflights every destructive approval before starting any persistent mutation", async () => {
const first = migration("people", 0, { approvalRequired: true });
const second = migration("people", 1, { approvalRequired: true });
const approvals = new CapturingApprovalProvider([approvalFor(first)]);
const database = new FakeMigrationDatabase();
await captureRunnerError(
applyModuleMigrations({
aggregate: aggregate({ id: "people", migrations: [first, second] }),
approvals,
database,
}),
"MIGRATION_APPROVAL_REQUIRED",
);
expect(approvals.requests[0]).toEqual([approvalFor(first), approvalFor(second)]);
expectNoPersistentMutation(database);
});
it("returns frozen deterministic applied results", async () => {
const first = migration("people", 0);
const second = migration("people", 1);
const inventory = aggregate({ id: "people", migrations: [first, second] });
const [left, right] = await Promise.all([
applyModuleMigrations({ aggregate: inventory, database: new FakeMigrationDatabase() }),
applyModuleMigrations({ aggregate: inventory, database: new FakeMigrationDatabase() }),
]);
expect(left).toEqual(right);
expect(Object.isFrozen(left)).toBe(true);
expect(Object.isFrozen(left.applied)).toBe(true);
expect(left.applied.every(Object.isFrozen)).toBe(true);
});
});
describe("module migration failures and side effects", () => {
it("sanitizes hostile connection, history, approval, and execution failures", async () => {
const ordinary = migration("people", 0);
const destructive = migration("people", 0, { approvalRequired: true });
const inventory = aggregate({ id: "people", migrations: [ordinary] });
const cases: readonly [Promise<unknown>, MigrationRunnerErrorCode][] = [
[
planModuleMigrations({
aggregate: inventory,
database: new FakeMigrationDatabase({ failures: ["open"] }),
}),
"MIGRATION_CONNECTION_FAILED",
],
[
planModuleMigrations({
aggregate: inventory,
database: new FakeMigrationDatabase({ failures: ["read-history"] }),
}),
"MIGRATION_HISTORY_READ_FAILED",
],
[
applyModuleMigrations({
aggregate: aggregate({ id: "people", migrations: [destructive] }),
approvals: new CapturingApprovalProvider([], true),
database: new FakeMigrationDatabase(),
}),
"MIGRATION_APPROVAL_LOOKUP_FAILED",
],
[
applyModuleMigrations({
aggregate: inventory,
database: new FakeMigrationDatabase({ failures: ["execute"] }),
}),
"MIGRATION_EXECUTION_FAILED",
],
];
for (const [operation, code] of cases) {
const error = await captureRunnerError(operation, code);
const publicText = `${error.message}\n${error.stack ?? ""}`;
expect(publicText).not.toContain("super-secret");
expect(publicText).not.toContain("private-db.internal");
expect(publicText).not.toContain("/home/ekklesia");
expect(publicText).not.toContain("DROP TABLE");
expect("cause" in error).toBe(false);
}
});
it("always closes sessions and preserves a primary error when cleanup also fails", async () => {
const initial = migration("people", 0);
const database = new FakeMigrationDatabase({
failures: ["read-history", "release-lock", "close"],
});
const error = await captureRunnerError(
applyModuleMigrations({
aggregate: aggregate({ id: "people", migrations: [initial] }),
database,
}),
"MIGRATION_HISTORY_READ_FAILED",
);
expect(error.code).toBe("MIGRATION_HISTORY_READ_FAILED");
expect(database.events.slice(-2)).toEqual(["lock:release", "session:close"]);
});
it("reports cleanup failure when successful work cannot release its resources", async () => {
const initial = migration("people", 0);
const database = new FakeMigrationDatabase({ failures: ["release-lock"] });
await captureRunnerError(
applyModuleMigrations({
aggregate: aggregate({ id: "people", migrations: [initial] }),
database,
}),
"MIGRATION_CLEANUP_FAILED",
);
expect(database.events.at(-1)).toBe("session:close");
});
it("does not connect or execute SQL merely by importing the runner", async () => {
const database = new FakeMigrationDatabase();
await import("../src/migrations/runner.js");
expect(database.events).toEqual([]);
expect(database.executedStatements).toEqual([]);
expect(database.committedHistory).toEqual([]);
});
});
@@ -0,0 +1,52 @@
import { describe, expect, it, vi } from "vitest";
const clientConstructor = vi.hoisted(() => vi.fn());
vi.mock("pg", () => ({ Client: clientConstructor }));
import {
createPostgresMigrationDatabase,
PostgresMigrationDatabaseError,
} from "../src/migrations/postgres-database.js";
describe("PostgreSQL migration database composition", () => {
it("does not construct a client or execute SQL merely because the adapter is imported", () => {
expect(clientConstructor).not.toHaveBeenCalled();
});
it.each([
undefined,
"",
"not-a-database-url",
"https://admin:[email protected]/database",
])("rejects missing or invalid configuration without exposing it (%s)", (databaseUrl) => {
let captured: unknown;
try {
createPostgresMigrationDatabase(databaseUrl);
} catch (error) {
captured = error;
}
expect(captured).toBeInstanceOf(PostgresMigrationDatabaseError);
expect(captured).toMatchObject({ code: "POSTGRES_MIGRATION_CONFIGURATION_INVALID" });
const publicText = `${String(captured)}\n${
captured instanceof Error ? (captured.stack ?? "") : ""
}\n${JSON.stringify(captured)}`;
if (databaseUrl) {
expect(publicText).not.toContain(databaseUrl);
}
expect(clientConstructor).not.toHaveBeenCalled();
});
it("creates a lazy database composition without opening a connection", () => {
const database = createPostgresMigrationDatabase(
"postgresql://operator:[email protected]/ekklesia",
);
expect(Object.isFrozen(database)).toBe(true);
expect(clientConstructor).not.toHaveBeenCalled();
expect(JSON.stringify(database)).not.toContain("operator");
expect(JSON.stringify(database)).not.toContain("do-not-log");
expect(JSON.stringify(database)).not.toContain("example.test");
});
});
@@ -0,0 +1,167 @@
# 0009: Transactional module migration execution
- Status: Accepted
- Date: 2026-07-21
## Context
ADR 0008 defines deterministic, module-owned Drizzle migration artifacts and offline validation.
It deliberately stops before connecting to PostgreSQL. Ekklesia now needs one Core-controlled path
that compares that immutable inventory with applied database history and either reports a plan or
applies every pending migration safely.
Migration execution is a platform-wide coordination problem. Independently installed modules can
share one database, their dependencies determine current execution order, and a concurrent deploy
must not race another runner. Destructive declarations also need approval from outside the module
artifact so editing SQL invalidates an earlier approval. These invariants cannot be delegated to
module entrypoints or Drizzle Kit configuration.
## Decision
### Core owns connectivity and execution
Only Core opens the PostgreSQL connection, reads applied migration history, acquires the migration
lock, executes SQL, and records success. Modules contribute only the validated artifacts defined by
ADR 0008. Migration execution is not a module lifecycle hook and does not occur automatically when
the runtime starts.
Core uses the pinned `pg` 8.22.0 driver. A run uses one explicitly managed client session because
PostgreSQL transactions and session-level advisory locks must remain on the same connection. The
client is released in every success and failure path.
### Fixed applied-history contract
Core owns the fixed table `public.ekklesia_core_migrations`. Modules may neither migrate nor query
it as private module state. Its columns are:
| Column | PostgreSQL type | Contract |
| --- | --- | --- |
| `identity` | `text` | Unique exact `<module-id>:<tag>` identity |
| `module_id` | `text` | Owning module ID |
| `migration_tag` | `text` | Drizzle journal tag |
| `migration_index` | `integer` | Module-local contiguous journal index |
| `checksum` | `text` | Lowercase SHA-256 of the exact packaged SQL bytes |
| `applied_order` | `bigint` identity | Primary key and database-assigned insertion order |
| `applied_at` | `timestamp with time zone` | Database application time, defaulting to `clock_timestamp()` |
The table also has unique constraints for `(module_id, migration_index)` and
`(module_id, migration_tag)`. A missing table means no migrations have been applied. Planning does
not create it. Applying creates it inside the migration transaction only when at least one
migration is pending, so a no-op apply does not mutate the database.
Packaged SQL remains the source of migration identity and exact contents; the database history is
the authority for what was successfully applied. Every history row must identify a migration in
the current validated inventory and match its module, tag, index, and checksum. Unknown identities,
duplicate or inconsistent metadata, and checksum mismatches stop planning and applying.
Within each module, applied history must be an exact prefix of its current Drizzle journal. A gap
or a later migration without all predecessors is invalid. Historical rows are not required to be
one prefix of today's global aggregate: dependencies and installed module sets can change, so
valid historical migrations from different modules may interleave. `applied_order` records that
historical interleaving for audit rather than redefining module-local order. Its values must be
unique and increasing but need not be contiguous because PostgreSQL identity sequences can retain
gaps after a rollback. An applied migration
which is absent from the current inventory is still rejected; removing an installed history
requires a future explicit platform policy.
Migrations remain forward-only. There are no downgrade or uninstall operations.
### Read-only planning and locked applying
`migrations:plan` completes static module and migration discovery, reads history, validates it,
and reports each migration as applied, pending, or awaiting destructive approval. It does not
acquire the apply lock or write to PostgreSQL. Its result is informational and can become stale as
soon as another process applies migrations.
`migrations:apply` completes the same atomic discovery before opening a database session. It then
acquires a session advisory lock and rereads history while holding that lock before it calculates
the authoritative pending set. The lock uses PostgreSQL's two-key advisory-lock namespace with the
stable signed 32-bit keys `1162562380` (`EKKL`) and `1296648018` (`MIGR`). Core polls
`pg_try_advisory_lock` for at most ten seconds rather than waiting indefinitely. A timeout fails
without executing SQL. Core explicitly releases the lock in cleanup and closes the session as a
second release guarantee, including after transaction, approval, or execution failures.
### One transaction for the whole pending batch
After history and approval preflight succeed, Core applies the complete dependency-first pending
aggregate in one PostgreSQL transaction. This matches Drizzle ORM 0.45.2's PostgreSQL migration
transaction boundary. The history table is created if needed, each migration is executed in
aggregate order, and its history row is inserted only after that migration succeeds. A later
failure rolls back every schema change and history insert from the current invocation. Migrations
committed by earlier invocations remain applied.
Core matches Drizzle's generated statement-breakpoint behavior: it splits only on the exact
literal `--> statement-breakpoint` marker. It never splits SQL on semicolons or tries to parse,
classify, reorder, or repair arbitrary SQL. Transaction-incompatible SQL fails the transaction;
there is no non-transactional fallback and no regular-expression safety claim. A module author
must express a compatible forward migration or wait for a future explicitly designed execution
mode.
### Destructive approval contract
Before beginning the transaction, Core resolves approvals from an externally supplied approval
provider and preflights the entire pending batch. Every migration declared
`approval-required` needs an exact approval triple:
```text
module ID + migration identity + exact SHA-256 checksum
```
All three values must match. Wildcards, module-wide grants, a command-line force flag, an
approve-all mode, and an `approved` value in the module sidecar are not accepted. The default
provider supplies no approvals, so destructive migrations are blocked until a deployment composes
a trusted external provider. Because all approvals are checked before the transaction begins, a
missing or stale approval causes no migration mutation.
### Deterministic and sanitized reporting
The runner consumes only the fully validated, frozen migration aggregate and publishes frozen,
deterministically ordered plans or results. Errors use stable categories and safe inventory
identifiers. They do not expose SQL contents, connection URLs, credentials, driver exception
messages, machine-specific paths, causes, stacks, or partial execution results. Cleanup failures
are reported without hiding the primary failure and without copying underlying exception text.
The developer commands read `DATABASE_URL` from the environment. A local uncommitted `.env` may
supply it, but commands never print it. This slice verifies database behavior through a narrow
in-memory database port and unit tests. A real PostgreSQL integration suite for lock contention,
DDL rollback, history creation, and driver behavior is deliberately deferred rather than claimed
by the current verification gate.
The transaction and single-client choices follow node-postgres's
[transaction guidance](https://node-postgres.com/features/transactions). The non-blocking lock and
explicit release behavior follow PostgreSQL's
[advisory-lock functions](https://www.postgresql.org/docs/current/functions-admin.html#FUNCTIONS-ADVISORY-LOCKS).
The batch transaction and exact breakpoint delimiter match Drizzle ORM 0.45.2's
[PostgreSQL dialect](https://github.com/drizzle-team/drizzle-orm/blob/0.45.2/drizzle-orm/src/pg-core/dialect.ts)
and [migration reader](https://github.com/drizzle-team/drizzle-orm/blob/0.45.2/drizzle-orm/src/migrator.ts).
## Alternatives
- Running migrations during module startup was rejected because database mutation must remain an
explicit deployment operation and complete discovery must precede all execution.
- A transaction per module or migration was rejected because a failure would publish a partially
applied invocation and diverge from the pinned Drizzle PostgreSQL runner.
- Transaction-level advisory locks were rejected because the pending set must be protected while
history is reread and approvals are checked before the transaction begins.
- An indefinitely blocking advisory lock was rejected because deploy automation needs a bounded,
deterministic failure mode.
- Trusting only a timestamp or the newest history row was rejected because installed modules can
interleave and exact identity and checksum comparison is required.
- A source-controlled approval flag or force option was rejected because the migration author
could grant their own approval and an edited artifact could reuse it.
- Parsing SQL with regular expressions was rejected for the reasons in ADR 0008 and because it
would imply safety guarantees the runner does not provide.
- Automatically falling back to non-transactional execution was rejected because it would weaken
rollback guarantees exactly when a migration is least compatible with them.
## Consequences
Operators receive an explicit read-only plan and a serialized transactional apply command. A
successful invocation records exact immutable history, while a failed invocation leaves none of
that invocation's schema or history changes committed. Destructive migrations are unusable until
a trusted approval provider is composed, by design.
The all-or-nothing batch can hold locks for longer than per-migration transactions and excludes
operations PostgreSQL forbids in a transaction. The current unit-tested database port gives fast,
deterministic coverage but cannot prove real server behavior. The next infrastructure-focused
slice must add live PostgreSQL tests before production operation is considered complete.
+1
View File
@@ -17,3 +17,4 @@ ADRs capture durable choices that affect more than one package or phase.
- [0006: Versioned capabilities and compiled module runtime](0006-versioned-capabilities-and-compiled-module-runtime.md) — Accepted
- [0007: Module-owned Drizzle schemas and table isolation](0007-module-owned-drizzle-schemas.md) — Accepted
- [0008: Module-owned migration contracts and offline validation](0008-module-owned-migration-contracts.md) — Accepted
- [0009: Transactional module migration execution](0009-transactional-module-migrations.md) — Accepted
+6 -3
View File
@@ -65,9 +65,12 @@ registration, dependency-ordered startup and reverse shutdown, rollback, and a b
template. Optional compiled schema entrypoints now contribute validated, namespaced module-owned
Drizzle tables to a deterministic aggregate without connecting to a database. Optional module-owned
Drizzle migration directories now produce an atomic dependency-ordered offline inventory with
validated metadata, explicit destructive-change declarations, and exact SHA-256 checksums.
validated metadata, explicit destructive-change declarations, and exact SHA-256 checksums. Core can
now compare that inventory with its fixed PostgreSQL history table, produce a read-only plan, and
apply the complete pending batch under a session advisory lock and one transaction. Destructive
migrations require an exact externally supplied approval bound to module, identity, and checksum.
Target architecture, not yet implemented: transactional migration execution and applied history,
Better Auth,
Target architecture, not yet implemented: production approval-provider composition, live
PostgreSQL migration-runner integration coverage, Better Auth,
permission enforcement, physical storage-bucket provisioning, Surface SDK, richer public resource
and action contracts, API server, dashboard, Builder, and reference feature modules.
+27 -1
View File
@@ -20,10 +20,12 @@ pnpm lint # Biome recommended rules
pnpm boundaries # architectural directions, UI ownership, and cycles
pnpm modules:validate # installed module manifests, compatibility, and dependency graph
pnpm migrations:validate # offline module migration inventory and exact checksums
pnpm migrations:plan # read-only comparison with PostgreSQL applied history
pnpm migrations:apply # locked, transactional application of the pending batch
pnpm typecheck # strict TypeScript
pnpm test # Vitest
pnpm build # Turborepo package build graph
pnpm verify # every code gate above
pnpm verify # every code-only gate; never plan or apply
```
The lint, typecheck, and test commands run both repository-root checks and every matching
@@ -41,6 +43,30 @@ checks every declared stable Drizzle migration directory. It reports dependency-
inventories and exact SHA-256 checksums without importing module code, connecting to PostgreSQL, or
executing SQL. Modules with no migration history are valid.
Migration planning and application require `DATABASE_URL`. Keep it in the uncommitted `.env`
created from `.env.example`:
```bash
cp .env.example .env
pnpm migrations:plan
pnpm migrations:apply
```
Planning completes discovery and validation, reads Core's applied history, and prints a
deterministic applied/pending/approval-blocked inventory. It is read-only and can become stale if
another process applies migrations afterward. Applying is an explicit mutating operation: it
acquires the bounded Core advisory lock, rereads history, checks every destructive approval, and
executes the complete pending aggregate in one transaction. A later failure rolls back all schema
and history changes from that invocation; migrations committed by earlier invocations remain.
The default command composition supplies no destructive approvals. There is no `--force`, wildcard,
or source-controlled approval escape hatch; deployments must later compose a trusted external
provider that supplies an exact module ID, migration identity, and checksum triple.
The migration runner currently has comprehensive unit coverage through its narrow database port.
Live PostgreSQL tests for real DDL rollback, advisory-lock contention, history-table behavior, and
driver cleanup are deliberately deferred and are not part of `pnpm verify` or CI yet.
CI runs `pnpm ci:verify`, which adds Compose configuration validation to `pnpm verify`. A separate
CI job runs the live infrastructure smoke test so code-only local verification does not require
Docker. The explicit suffix avoids pnpm's built-in `ci` install command.
+24 -5
View File
@@ -218,18 +218,20 @@ Run the module's `migration:generate` script with a lowercase snake-case name, r
SQL and snapshot, and add the matching entry to `drizzle/meta/_ekklesia.json`. Every entry must
explicitly declare either `{"mode":"none"}` or
`{"mode":"approval-required","reason":"..."}`. Do not add an `approved` flag; approval will be
an external Core-owned record bound to the migration checksum.
supplied by an external Core-owned provider and bound to module ID, identity, and checksum.
Migration identities are `<module-id>:<generated-tag>`. Core preserves Drizzle journal order inside
each module and dependency-first discovery order between modules. It calculates SHA-256 from the
exact SQL bytes, so changing whitespace or line endings changes the checksum. Never edit an
already-applied migration; the future database history will detect checksum changes. Add a new
forward migration instead. Downgrade and uninstall migrations are not supported.
already-applied migration: Core compares the current exact checksum and metadata with its fixed
PostgreSQL history and refuses changed, missing, unknown, or out-of-prefix entries. Add a new forward
migration instead. Downgrade and uninstall migrations are not supported.
A module may migrate only its own objects. SQL is intentionally treated as opaque: offline
validation checks the artifact and metadata contract but cannot prove SQL ownership or classify
destructive behavior. Review those properties explicitly. Core remains the only future migration
executor.
destructive behavior. Review those properties explicitly. Core is the only migration executor. It
follows Drizzle's exact `--> statement-breakpoint` delimiter and never naively splits semicolons or
falls back to non-transactional execution.
## Discovery and lifecycle
@@ -239,6 +241,7 @@ From the repository root, run:
pnpm install
pnpm modules:validate
pnpm migrations:validate
pnpm migrations:plan
pnpm verify
```
@@ -260,6 +263,22 @@ Drizzle journals and snapshot chains, Ekklesia policy metadata, ordering, invent
checksums atomically. Its concise report lists every installed module and migration in deterministic
order.
With `DATABASE_URL` set through the uncommitted root `.env`, `pnpm migrations:plan` performs a
read-only applied-history comparison. The plan can become stale. `pnpm migrations:apply` is a
separate, explicit database mutation: Core acquires its bounded session advisory lock, rereads and
validates history, preflights the whole destructive-approval set, and applies all pending modules in
dependency-first order in one transaction. If a later migration fails, every schema and history
change from that invocation rolls back; migrations committed by an earlier invocation remain.
An `approval-required` migration needs an approval from an external provider matching its module
ID, exact identity, and exact checksum. Module metadata cannot grant approval, and there is no
wildcard, force, or approve-all option. The default developer command supplies no approvals and
therefore refuses destructive pending migrations.
Runner behavior is unit-tested through a narrow database port. Live PostgreSQL coverage of DDL
rollback, advisory-lock contention, and driver behavior is a deferred infrastructure slice; the
current verification suite does not claim that coverage.
Core completes discovery before importing any server entrypoint. It then loads and verifies every
runtime definition before running hooks, registers all modules dependency-first into an unpublished
staged capability registry, and starts them in the same order. Shutdown is reverse-order. If a later
+2
View File
@@ -22,6 +22,8 @@
"infrastructure:up": "docker compose --env-file .env -f infrastructure/compose.yaml up --detach --wait",
"lint": "turbo run lint lint:root",
"lint:root": "biome lint .",
"migrations:apply": "turbo run build && node --env-file-if-exists=.env core/dist/cli/apply-migrations.js",
"migrations:plan": "turbo run build && node --env-file-if-exists=.env core/dist/cli/plan-migrations.js",
"migrations:validate": "turbo run build && node core/dist/cli/validate-migrations.js",
"modules:validate": "turbo run build && node core/dist/cli/validate-modules.js",
"test": "turbo run test test:root",
+130 -5
View File
@@ -34,7 +34,10 @@ importers:
version: link:../packages/contracts
drizzle-orm:
specifier: 0.45.2
version: 0.45.2
version: 0.45.2(@types/[email protected])([email protected])
pg:
specifier: 8.22.0
version: 8.22.0
devDependencies:
'@biomejs/biome':
specifier: 2.5.3
@@ -42,6 +45,9 @@ importers:
'@types/node':
specifier: 24.13.3
version: 24.13.3
'@types/pg':
specifier: 8.20.0
version: 8.20.0
typescript:
specifier: 7.0.2
version: 7.0.2
@@ -53,7 +59,7 @@ importers:
dependencies:
drizzle-orm:
specifier: 0.45.2
version: 0.45.2
version: 0.45.2(@types/[email protected])([email protected])
semver:
specifier: 7.8.5
version: 7.8.5
@@ -91,7 +97,7 @@ importers:
version: 24.13.3
drizzle-orm:
specifier: 0.45.2
version: 0.45.2
version: 0.45.2(@types/[email protected])([email protected])
typescript:
specifier: 7.0.2
version: 7.0.2
@@ -108,7 +114,7 @@ importers:
version: link:../../packages/module-sdk
drizzle-orm:
specifier: 0.45.2
version: 0.45.2
version: 0.45.2(@types/[email protected])([email protected])
devDependencies:
'@biomejs/biome':
specifier: 2.5.3
@@ -810,6 +816,9 @@ packages:
'@types/[email protected]':
resolution: {integrity: sha512-Dh8vAsV36ig5wa9OX4pXvMc9D3Veibfw2wix0CUwYODLD8nkj9UsLjASr49nPg+2eKzxhBV+v7L8pXvT4e639Q==}
'@types/[email protected]':
resolution: {integrity: sha512-bEPFOaMAHTEP1EzpvHTbmwR8UsFyHSKsRisLIHVMXnpNefSbGA1bD6CVy+qKjGSqmZqNqBDV2azOBo8TgkcVow==}
'@types/[email protected]':
resolution: {integrity: sha512-FmgJfu+MOcQ370SD0ev7EI8TlCAfKYU+B4m5T3yXc1CiRN94g/SZPtsCkk506aUDtlMnFZvasDwHHUcZUEaYuA==}
@@ -1207,6 +1216,40 @@ packages:
[email protected]:
resolution: {integrity: sha512-WUjGcAqP1gQacoQe+OBJsFA7Ld4DyXuUIjZ5cc75cLHvJ7dtNsTugphxIADwspS+AraAUePCKrSVtPLFj/F88w==}
[email protected]:
resolution: {integrity: sha512-Vo7z/6rrQYxpNRylp4Tlob2elzbh+N/MOQbxFVWCxS7oEx6jF53GTJFxK2WWpKuBRkmiin4Mt+xofFDjx09R0A==}
[email protected]:
resolution: {integrity: sha512-XwWDGcLRGCXAR8F/AM5bG7Q+A3Wm2s6QeEjlOKZLlH3UYcguiqCWKyWXVag5TLTIjR7oOJUY8kcADaZgWPyLeg==}
[email protected]:
resolution: {integrity: sha512-WCtabS6t3c8SkpDBUlb1kjOs7l66xsGdKpIPZsg4wR+B3+u9UAum2odSsF9tnvxg80h4ZxLWMy4pRjOsFIqQpw==}
engines: {node: '>=4.0.0'}
[email protected]:
resolution: {integrity: sha512-gKtPkFdQPU3DksooVLi9LsjZxrsBUZIpa+7aVx+LV5pNh0KzP4Zleud2po+ConrxbuXGBJ6Hfer6hdgpIBpBaw==}
peerDependencies:
pg: '>=8.0'
[email protected]:
resolution: {integrity: sha512-cq9sECI5s0+uPUXjbz8ioyPJni6RzsRib0US67i5IoTZKw8fNeYlVE7u8F4dG7vEJJtc5wdD1K189lCCUwqWTQ==}
[email protected]:
resolution: {integrity: sha512-qTAAlrEsl8s4OiEQY69wDvcMIdQN6wdz5ojQiOy6YRMuynxenON0O5oCpJI6lshc6scgAY8qvJ2On/p+CXY0GA==}
engines: {node: '>=4'}
[email protected]:
resolution: {integrity: sha512-8wih1vVIBMxoUM2oB4soJsD9tDnDpLv4OXBJ+EJzFsvycD+lfyIreC2gGHq78f8jbLLt+bvlPTFdFZfJkOuzAA==}
engines: {node: '>= 16.0.0'}
peerDependencies:
pg-native: '>=3.0.1'
peerDependenciesMeta:
pg-native:
optional: true
[email protected]:
resolution: {integrity: sha512-FdW9r/jQZhSeohs1Z3sI1yxFQNFvMcnmfuj4WBMUTxOrAyLMaTcE1aAMBiTlbMNaXvBCQuVi0R7hd8udDSP7ug==}
[email protected]:
resolution: {integrity: sha512-xceH2snhtb5M9liqDsmEw56le376mTZkEX/jEb/RxNFyegNul7eNslCXP9FDj/Lcu0X8KEyMceP2ntpaHrDEVA==}
@@ -1218,6 +1261,22 @@ packages:
resolution: {integrity: sha512-J7EF+8X+CzRPaJPOv9Ck2wNWJvGnnl3PcNPAdGg6GTLjyVpyQ0yATMSXRFRV01BviT/9Gwuc3rjEyJbDJG9a4w==}
engines: {node: ^10 || ^12 || >=14}
[email protected]:
resolution: {integrity: sha512-VpZrUqU5A69eQyW2c5CA1jtLecCsN2U/bD6VilrFDWq5+5UIEVO7nazS3TEcHf1zuPYO/sqGvUvW62g86RXZuA==}
engines: {node: '>=4'}
[email protected]:
resolution: {integrity: sha512-5+5HqXnsZPE65IJZSMkZtURARZelel2oXUEO8rH83VS/hxH5vv1uHquPg5wZs8yMAfdv971IU+kcPUczi7NVBQ==}
engines: {node: '>=0.10.0'}
[email protected]:
resolution: {integrity: sha512-suDmjLVQg78nMK2UZ454hAG+OAW+HQPZ6n++TNDUX+L0+uUlLywnoxJKDou51Zm+zTCjrCl0Nq6J9C5hP9vK/Q==}
engines: {node: '>=0.10.0'}
[email protected]:
resolution: {integrity: sha512-9ZhXKM/rw350N1ovuWHbGxnGh/SNJ4cnxHiM0rxE4VN41wsg8P8zWn9hv/buK00RP4WvlOyr/RBDiptyxVbkZQ==}
engines: {node: '>=0.10.0'}
[email protected]:
resolution: {integrity: sha512-seS2Tj26TBVOC2NIc2rOe2y2ZO7efxITtLZcGSOnHHNOQ7CkiUBfw0Iw2ck6xkIhPwLhKNLS8BO+hEpngQlqzw==}
@@ -1245,6 +1304,10 @@ packages:
resolution: {integrity: sha512-UjgapumWlbMhkBgzT7Ykc5YXUT46F0iKu8SGXq0bcwP5dz/h0Plj6enJqjz1Zbq2l5WaqYnrVbwWOWMyF3F47g==}
engines: {node: '>=0.10.0'}
[email protected]:
resolution: {integrity: sha512-UcjcJOWknrNkF6PLX83qcHM6KHgVKNkV62Y8a5uYDVv9ydGQVwAHMKqHdJje1VTWpljG0WYpCDhrCdAOYH4TWg==}
engines: {node: '>= 10.x'}
[email protected]:
resolution: {integrity: sha512-1XMJE5fQo1jGH6Y/7ebnwPOBEkIEnT4QF32d5R1+VXdXveM0IBMJt8zfaxX1P3QhVwrYe+576+jkANtSS2mBbw==}
@@ -1375,6 +1438,10 @@ packages:
engines: {node: '>=8'}
hasBin: true
[email protected]:
resolution: {integrity: sha512-LKYU1iAXJXUgAXn9URjiu+MWhyUXHsvfp7mcuYm9dSUKK0/CjtrUwFAxD82/mCWbtLsGjFIad0wIsod4zrTAEQ==}
engines: {node: '>=0.4'}
[email protected]:
resolution: {integrity: sha512-2AvhNX3mb8zd6Zy7INTtSpl1F15HW6Wnqj0srWlkKLcpYl/gMIMJiyuGq2KeI2YFxUPjdlB+3Lc10seMLtL4cA==}
engines: {node: '>= 14.6'}
@@ -1770,6 +1837,12 @@ snapshots:
dependencies:
undici-types: 7.18.2
'@types/[email protected]':
dependencies:
'@types/node': 24.13.3
pg-protocol: 1.15.0
pg-types: 2.2.0
'@types/[email protected]': {}
'@typescript/[email protected]':
@@ -1890,7 +1963,10 @@ snapshots:
esbuild: 0.25.12
tsx: 4.23.1
[email protected]: {}
[email protected](@types/[email protected])([email protected]):
optionalDependencies:
'@types/pg': 8.20.0
pg: 8.22.0
[email protected]: {}
@@ -2053,6 +2129,41 @@ snapshots:
[email protected]: {}
[email protected]:
optional: true
[email protected]: {}
[email protected]: {}
[email protected]([email protected]):
dependencies:
pg: 8.22.0
[email protected]: {}
[email protected]:
dependencies:
pg-int8: 1.0.1
postgres-array: 2.0.0
postgres-bytea: 1.0.1
postgres-date: 1.0.7
postgres-interval: 1.2.0
[email protected]:
dependencies:
pg-connection-string: 2.14.0
pg-pool: 3.14.0([email protected])
pg-protocol: 1.15.0
pg-types: 2.2.0
pgpass: 1.0.5
optionalDependencies:
pg-cloudflare: 1.4.0
[email protected]:
dependencies:
split2: 4.2.0
[email protected]: {}
[email protected]: {}
@@ -2063,6 +2174,16 @@ snapshots:
picocolors: 1.1.1
source-map-js: 1.2.1
[email protected]: {}
[email protected]: {}
[email protected]: {}
[email protected]:
dependencies:
xtend: 4.0.2
[email protected]: {}
[email protected]:
@@ -2099,6 +2220,8 @@ snapshots:
[email protected]: {}
[email protected]: {}
[email protected]: {}
[email protected]: {}
@@ -2203,6 +2326,8 @@ snapshots:
siginfo: 2.0.0
stackback: 0.0.2
[email protected]: {}
[email protected]: {}
[email protected]: {}