ekklesiaandClaude Sonnet 5
679a5d225c
Adds Core-owned identity contracts, a Core-owned Better Auth database schema (ekklesia_core_user/session/account/verification), a first-class Core-owned migration source that coexists with module-owned migrations under the exact existing migration system, and an explicit Better Auth construction boundary. No authentication endpoint or sign-in UI is exposed yet (ADR 0010). Co-Authored-By: Claude Sonnet 5 <[email protected]>
1460 lines
52 KiB
TypeScript
1460 lines
52 KiB
TypeScript
import { createHash, randomBytes } from "node:crypto";
|
|
import path from "node:path";
|
|
import { fileURLToPath } from "node:url";
|
|
import { Client, type QueryResult, type QueryResultRow } from "pg";
|
|
import { afterAll, describe, expect, it, vi } from "vitest";
|
|
import { reportMigrationCommandError } from "../src/cli/migration-command-errors.js";
|
|
import {
|
|
applyModuleMigrations,
|
|
createPostgresMigrationDatabase,
|
|
discoverCoreMigrations,
|
|
type MigrationApproval,
|
|
type MigrationApprovalProvider,
|
|
MigrationRunnerError,
|
|
type MigrationRunnerErrorCode,
|
|
planModuleMigrations,
|
|
} from "../src/migrations/index.js";
|
|
import type {
|
|
DiscoveredModuleMigration,
|
|
ModuleMigrationAggregate,
|
|
} from "../src/modules/migration-discovery.js";
|
|
|
|
const coreMigrationsDirectory = path.join(
|
|
path.dirname(fileURLToPath(import.meta.url)),
|
|
"..",
|
|
"drizzle",
|
|
);
|
|
|
|
const integrationDatabaseUrlEnvironment = "EKKLESIA_MIGRATION_TEST_DATABASE_URL";
|
|
const historyTable = "public.ekklesia_core_migrations";
|
|
const migrationLockNamespace = 1_162_562_380;
|
|
const migrationLockKey = 1_296_648_018;
|
|
const concurrencyBarrierNamespace = 1_162_562_380;
|
|
const concurrencyBarrierKey = 1_414_743_380;
|
|
const trackedDatabases = new Set<string>();
|
|
let databaseCounter = 0;
|
|
|
|
class IntegrationHarnessError extends Error {
|
|
constructor(message: string) {
|
|
super(message);
|
|
this.name = "IntegrationHarnessError";
|
|
this.stack = `${this.name}: ${this.message}`;
|
|
}
|
|
}
|
|
|
|
function harnessError(message: string): IntegrationHarnessError {
|
|
return new IntegrationHarnessError(message);
|
|
}
|
|
|
|
function baseDatabaseUrl(): string {
|
|
const value = process.env[integrationDatabaseUrlEnvironment];
|
|
if (typeof value !== "string" || value.trim() === "") {
|
|
throw harnessError("The migration integration database is not configured.");
|
|
}
|
|
|
|
try {
|
|
const parsed = new URL(value);
|
|
if (parsed.protocol !== "postgres:" && parsed.protocol !== "postgresql:") {
|
|
throw new TypeError("Unsupported protocol.");
|
|
}
|
|
} catch {
|
|
throw harnessError("The migration integration database configuration is invalid.");
|
|
}
|
|
return value;
|
|
}
|
|
|
|
function quoteDatabaseName(databaseName: string): string {
|
|
if (!/^[a-z][a-z0-9_]{0,62}$/u.test(databaseName)) {
|
|
throw harnessError("The generated integration database name is invalid.");
|
|
}
|
|
return `"${databaseName}"`;
|
|
}
|
|
|
|
function makeDatabaseName(label: string): string {
|
|
const safeLabel = label
|
|
.toLowerCase()
|
|
.replaceAll(/[^a-z0-9]+/gu, "_")
|
|
.replaceAll(/^_+|_+$/gu, "")
|
|
.slice(0, 18);
|
|
databaseCounter += 1;
|
|
return `ekk_mig_${process.pid}_${databaseCounter}_${safeLabel || "scenario"}_${randomBytes(4).toString("hex")}`;
|
|
}
|
|
|
|
function databaseUrl(databaseName: string, applicationName?: string): string {
|
|
const parsed = new URL(baseDatabaseUrl());
|
|
parsed.pathname = `/${databaseName}`;
|
|
if (applicationName === undefined) {
|
|
parsed.searchParams.delete("application_name");
|
|
} else {
|
|
parsed.searchParams.set("application_name", applicationName);
|
|
}
|
|
return parsed.toString();
|
|
}
|
|
|
|
async function openAdministrativeClient(
|
|
connectionString: string,
|
|
applicationName: string,
|
|
): Promise<Client> {
|
|
const client = new Client({ application_name: applicationName, connectionString });
|
|
client.on("error", () => undefined);
|
|
try {
|
|
await client.connect();
|
|
return client;
|
|
} catch {
|
|
try {
|
|
await client.end();
|
|
} catch {
|
|
// The public harness failure remains static when connection cleanup also fails.
|
|
}
|
|
throw harnessError("The migration integration database connection failed.");
|
|
}
|
|
}
|
|
|
|
async function safeQuery<Row extends QueryResultRow>(
|
|
client: Client,
|
|
text: string,
|
|
values: unknown[] = [],
|
|
failureMessage = "An integration database administration operation failed.",
|
|
): Promise<QueryResult<Row>> {
|
|
try {
|
|
return await client.query<Row>(text, values);
|
|
} catch {
|
|
throw harnessError(failureMessage);
|
|
}
|
|
}
|
|
|
|
async function safeEnd(client: Client): Promise<void> {
|
|
try {
|
|
await client.end();
|
|
} catch {
|
|
throw harnessError("An integration database connection could not be closed.");
|
|
}
|
|
}
|
|
|
|
async function createDatabase(databaseName: string): Promise<void> {
|
|
const admin = await openAdministrativeClient(baseDatabaseUrl(), "ekklesia_migration_test_setup");
|
|
let operationError: unknown;
|
|
try {
|
|
await safeQuery(admin, `CREATE DATABASE ${quoteDatabaseName(databaseName)} TEMPLATE template0`);
|
|
} catch (error) {
|
|
operationError = error;
|
|
}
|
|
try {
|
|
await safeEnd(admin);
|
|
} catch (error) {
|
|
if (operationError === undefined) {
|
|
operationError = error;
|
|
}
|
|
}
|
|
if (operationError !== undefined) {
|
|
throw operationError;
|
|
}
|
|
}
|
|
|
|
async function dropDatabase(databaseName: string): Promise<void> {
|
|
const admin = await openAdministrativeClient(
|
|
baseDatabaseUrl(),
|
|
"ekklesia_migration_test_cleanup",
|
|
);
|
|
let operationError: unknown;
|
|
try {
|
|
await safeQuery(
|
|
admin,
|
|
"SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname = $1 AND pid <> pg_backend_pid()",
|
|
[databaseName],
|
|
);
|
|
await safeQuery(
|
|
admin,
|
|
`DROP DATABASE IF EXISTS ${quoteDatabaseName(databaseName)} WITH (FORCE)`,
|
|
);
|
|
trackedDatabases.delete(databaseName);
|
|
} catch (error) {
|
|
operationError = error;
|
|
}
|
|
try {
|
|
await safeEnd(admin);
|
|
} catch (error) {
|
|
if (operationError === undefined) {
|
|
operationError = error;
|
|
}
|
|
}
|
|
if (operationError !== undefined) {
|
|
throw operationError;
|
|
}
|
|
}
|
|
|
|
interface ScenarioDatabase {
|
|
readonly client: Client;
|
|
readonly name: string;
|
|
readonly url: string;
|
|
}
|
|
|
|
async function withIsolatedDatabase<T>(
|
|
label: string,
|
|
operation: (database: ScenarioDatabase) => Promise<T>,
|
|
): Promise<T> {
|
|
const name = makeDatabaseName(label);
|
|
await createDatabase(name);
|
|
trackedDatabases.add(name);
|
|
|
|
let client: Client | undefined;
|
|
let outcome:
|
|
| { readonly error: unknown; readonly success: false }
|
|
| { readonly success: true; readonly value: T };
|
|
try {
|
|
const url = databaseUrl(name);
|
|
client = await openAdministrativeClient(url, "ekklesia_migration_test_inspection");
|
|
outcome = { success: true, value: await operation(Object.freeze({ client, name, url })) };
|
|
} catch (error) {
|
|
outcome = { error, success: false };
|
|
}
|
|
|
|
let cleanupFailed = false;
|
|
if (client !== undefined) {
|
|
try {
|
|
await safeEnd(client);
|
|
} catch {
|
|
cleanupFailed = true;
|
|
}
|
|
}
|
|
try {
|
|
await dropDatabase(name);
|
|
} catch {
|
|
cleanupFailed = true;
|
|
}
|
|
|
|
if (!outcome.success) {
|
|
throw outcome.error;
|
|
}
|
|
if (cleanupFailed) {
|
|
throw harnessError("An isolated migration integration database could not be removed.");
|
|
}
|
|
return outcome.value;
|
|
}
|
|
|
|
afterAll(async () => {
|
|
let cleanupFailed = false;
|
|
for (const databaseName of [...trackedDatabases]) {
|
|
try {
|
|
await dropDatabase(databaseName);
|
|
} catch {
|
|
cleanupFailed = true;
|
|
}
|
|
}
|
|
if (cleanupFailed) {
|
|
throw harnessError("Migration integration database cleanup was incomplete.");
|
|
}
|
|
});
|
|
|
|
interface MigrationOptions {
|
|
readonly approvalRequired?: boolean;
|
|
readonly sql: string;
|
|
readonly tag: string;
|
|
}
|
|
|
|
interface ModuleFixture {
|
|
readonly id: string;
|
|
readonly migrations: readonly DiscoveredModuleMigration[];
|
|
}
|
|
|
|
function migration(
|
|
moduleId: string,
|
|
index: number,
|
|
options: MigrationOptions,
|
|
): DiscoveredModuleMigration {
|
|
const approvalRequired = options.approvalRequired ?? false;
|
|
return Object.freeze({
|
|
approvalRequired,
|
|
breakpoints: true,
|
|
checksum: createHash("sha256").update(options.sql).digest("hex"),
|
|
destructive: Object.freeze(
|
|
approvalRequired
|
|
? {
|
|
mode: "approval-required" as const,
|
|
reason: "This integration fixture requires an exact external approval.",
|
|
}
|
|
: { mode: "none" as const },
|
|
),
|
|
file: `${options.tag}.sql`,
|
|
identity: `${moduleId}:${options.tag}`,
|
|
index,
|
|
moduleId,
|
|
sql: options.sql,
|
|
tag: options.tag,
|
|
timestamp: 1_800_000_000_000 + index,
|
|
});
|
|
}
|
|
|
|
function aggregate(...moduleFixtures: readonly ModuleFixture[]): ModuleMigrationAggregate {
|
|
const modules = Object.freeze(
|
|
moduleFixtures.map((moduleFixture) =>
|
|
Object.freeze({
|
|
migrations: Object.freeze([...moduleFixture.migrations]),
|
|
moduleId: moduleFixture.id,
|
|
}),
|
|
),
|
|
);
|
|
return Object.freeze({
|
|
discoveredModules: Object.freeze([]),
|
|
migrations: Object.freeze(modules.flatMap((moduleFixture) => moduleFixture.migrations)),
|
|
modules,
|
|
});
|
|
}
|
|
|
|
function moduleFixture(
|
|
id: string,
|
|
migrations: readonly DiscoveredModuleMigration[],
|
|
): ModuleFixture {
|
|
return Object.freeze({ id, migrations: Object.freeze([...migrations]) });
|
|
}
|
|
|
|
const peopleInitialSql = `CREATE TABLE "people_records" (
|
|
"id" bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
|
|
"display_name" text NOT NULL
|
|
);\n`;
|
|
const calendarInitialSql = `CREATE TABLE "calendar_events" (
|
|
"id" bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
|
|
"owner_id" bigint NOT NULL,
|
|
"title" text NOT NULL
|
|
);\n`;
|
|
const peopleLaterSql = `ALTER TABLE "people_records" ADD COLUMN "email" text;\n`;
|
|
|
|
function peopleInitial(sql = peopleInitialSql): DiscoveredModuleMigration {
|
|
return migration("people", 0, { sql, tag: "0000_initial" });
|
|
}
|
|
|
|
function calendarInitial(): DiscoveredModuleMigration {
|
|
return migration("calendar", 0, { sql: calendarInitialSql, tag: "0000_initial" });
|
|
}
|
|
|
|
function peopleLater(sql = peopleLaterSql): DiscoveredModuleMigration {
|
|
return migration("people", 1, { sql, tag: "0001_add_email" });
|
|
}
|
|
|
|
interface RelationExistenceRow extends QueryResultRow {
|
|
readonly relation_name: string | null;
|
|
}
|
|
|
|
interface UserObjectRow extends QueryResultRow {
|
|
readonly object_kind: string;
|
|
readonly object_name: string;
|
|
readonly schema_name: string;
|
|
}
|
|
|
|
interface ColumnRow extends QueryResultRow {
|
|
readonly column_default: string | null;
|
|
readonly column_name: string;
|
|
readonly data_type: string;
|
|
readonly is_nullable: string;
|
|
readonly ordinal_position: number;
|
|
readonly table_name: string;
|
|
readonly table_schema: string;
|
|
}
|
|
|
|
interface ConstraintRow extends QueryResultRow {
|
|
readonly constraint_definition: string;
|
|
readonly constraint_name: string;
|
|
readonly table_name: string;
|
|
readonly table_schema: string;
|
|
}
|
|
|
|
interface SequenceRow extends QueryResultRow {
|
|
readonly cache_size: string;
|
|
readonly cycle: boolean;
|
|
readonly increment_by: string;
|
|
readonly last_value: string | null;
|
|
readonly sequence_name: string;
|
|
readonly sequence_schema: string;
|
|
}
|
|
|
|
interface HistoryRow extends QueryResultRow {
|
|
readonly applied_at: Date;
|
|
readonly applied_order: number;
|
|
readonly checksum: string;
|
|
readonly identity: string;
|
|
readonly migration_index: number;
|
|
readonly migration_tag: string;
|
|
readonly module_id: string;
|
|
}
|
|
|
|
interface BooleanRow extends QueryResultRow {
|
|
readonly value: boolean;
|
|
}
|
|
|
|
interface CountRow extends QueryResultRow {
|
|
readonly value: number;
|
|
}
|
|
|
|
interface DatabaseState {
|
|
readonly columns: readonly ColumnRow[];
|
|
readonly constraints: readonly ConstraintRow[];
|
|
readonly history: readonly HistoryRow[];
|
|
readonly objects: readonly UserObjectRow[];
|
|
readonly sequences: readonly SequenceRow[];
|
|
readonly supplementalObjects: readonly UserObjectRow[];
|
|
}
|
|
|
|
async function relationExists(client: Client, qualifiedName: string): Promise<boolean> {
|
|
const result = await safeQuery<RelationExistenceRow>(
|
|
client,
|
|
"SELECT to_regclass($1) AS relation_name",
|
|
[qualifiedName],
|
|
);
|
|
return result.rows[0]?.relation_name !== null && result.rows[0]?.relation_name !== undefined;
|
|
}
|
|
|
|
async function readHistory(client: Client): Promise<readonly HistoryRow[]> {
|
|
if (!(await relationExists(client, historyTable))) {
|
|
return Object.freeze([]);
|
|
}
|
|
const result = await safeQuery<HistoryRow>(
|
|
client,
|
|
`
|
|
SELECT
|
|
applied_order::integer AS applied_order,
|
|
identity,
|
|
module_id,
|
|
migration_tag,
|
|
migration_index,
|
|
checksum,
|
|
applied_at
|
|
FROM public.ekklesia_core_migrations
|
|
ORDER BY applied_order ASC
|
|
`,
|
|
);
|
|
return Object.freeze(result.rows.map((row) => Object.freeze({ ...row })));
|
|
}
|
|
|
|
async function databaseState(client: Client): Promise<DatabaseState> {
|
|
const objects = await safeQuery<UserObjectRow>(
|
|
client,
|
|
`
|
|
SELECT
|
|
namespace.nspname AS schema_name,
|
|
relation.relname AS object_name,
|
|
relation.relkind::text AS object_kind
|
|
FROM pg_catalog.pg_class AS relation
|
|
INNER JOIN pg_catalog.pg_namespace AS namespace
|
|
ON namespace.oid = relation.relnamespace
|
|
WHERE namespace.nspname NOT LIKE 'pg\\_%' ESCAPE '\\'
|
|
AND namespace.nspname <> 'information_schema'
|
|
ORDER BY namespace.nspname, relation.relname, relation.relkind
|
|
`,
|
|
[],
|
|
"Integration user-object inspection failed.",
|
|
);
|
|
const columns = await safeQuery<ColumnRow>(
|
|
client,
|
|
`
|
|
SELECT
|
|
table_schema,
|
|
table_name,
|
|
column_name,
|
|
ordinal_position,
|
|
data_type,
|
|
is_nullable,
|
|
column_default
|
|
FROM information_schema.columns
|
|
WHERE table_schema NOT LIKE 'pg\\_%' ESCAPE '\\'
|
|
AND table_schema <> 'information_schema'
|
|
ORDER BY table_schema, table_name, ordinal_position
|
|
`,
|
|
[],
|
|
"Integration column inspection failed.",
|
|
);
|
|
const constraints = await safeQuery<ConstraintRow>(
|
|
client,
|
|
`
|
|
SELECT
|
|
namespace.nspname AS table_schema,
|
|
relation.relname AS table_name,
|
|
catalog_constraint.conname AS constraint_name,
|
|
pg_catalog.pg_get_constraintdef(catalog_constraint.oid, true) AS constraint_definition
|
|
FROM pg_catalog.pg_constraint AS catalog_constraint
|
|
INNER JOIN pg_catalog.pg_class AS relation
|
|
ON relation.oid = catalog_constraint.conrelid
|
|
INNER JOIN pg_catalog.pg_namespace AS namespace ON namespace.oid = relation.relnamespace
|
|
WHERE namespace.nspname NOT LIKE 'pg\\_%' ESCAPE '\\'
|
|
AND namespace.nspname <> 'information_schema'
|
|
ORDER BY namespace.nspname, relation.relname, catalog_constraint.conname
|
|
`,
|
|
[],
|
|
"Integration constraint inspection failed.",
|
|
);
|
|
const sequences = await safeQuery<SequenceRow>(
|
|
client,
|
|
`
|
|
SELECT
|
|
schemaname AS sequence_schema,
|
|
sequencename AS sequence_name,
|
|
increment_by::text,
|
|
cycle,
|
|
cache_size::text,
|
|
last_value::text
|
|
FROM pg_catalog.pg_sequences
|
|
WHERE schemaname NOT LIKE 'pg\\_%' ESCAPE '\\'
|
|
AND schemaname <> 'information_schema'
|
|
ORDER BY schemaname, sequencename
|
|
`,
|
|
[],
|
|
"Integration sequence inspection failed.",
|
|
);
|
|
const supplementalObjects = await safeQuery<UserObjectRow>(
|
|
client,
|
|
`
|
|
SELECT
|
|
'schema' AS object_kind,
|
|
namespace.nspname AS object_name,
|
|
'' AS schema_name
|
|
FROM pg_catalog.pg_namespace AS namespace
|
|
WHERE namespace.nspname NOT LIKE 'pg\\_%' ESCAPE '\\'
|
|
AND namespace.nspname <> 'information_schema'
|
|
UNION ALL
|
|
SELECT
|
|
'extension' AS object_kind,
|
|
catalog_extension.extname AS object_name,
|
|
namespace.nspname AS schema_name
|
|
FROM pg_catalog.pg_extension AS catalog_extension
|
|
INNER JOIN pg_catalog.pg_namespace AS namespace
|
|
ON namespace.oid = catalog_extension.extnamespace
|
|
UNION ALL
|
|
SELECT
|
|
'routine' AS object_kind,
|
|
catalog_routine.oid::regprocedure::text AS object_name,
|
|
namespace.nspname AS schema_name
|
|
FROM pg_catalog.pg_proc AS catalog_routine
|
|
INNER JOIN pg_catalog.pg_namespace AS namespace
|
|
ON namespace.oid = catalog_routine.pronamespace
|
|
WHERE namespace.nspname NOT LIKE 'pg\\_%' ESCAPE '\\'
|
|
AND namespace.nspname <> 'information_schema'
|
|
UNION ALL
|
|
SELECT
|
|
CASE user_type.typtype
|
|
WHEN 'd' THEN 'domain'
|
|
WHEN 'e' THEN 'enum'
|
|
WHEN 'm' THEN 'multirange'
|
|
WHEN 'r' THEN 'range'
|
|
END AS object_kind,
|
|
user_type.typname AS object_name,
|
|
namespace.nspname AS schema_name
|
|
FROM pg_catalog.pg_type AS user_type
|
|
INNER JOIN pg_catalog.pg_namespace AS namespace
|
|
ON namespace.oid = user_type.typnamespace
|
|
WHERE namespace.nspname NOT LIKE 'pg\\_%' ESCAPE '\\'
|
|
AND namespace.nspname <> 'information_schema'
|
|
AND user_type.typtype IN ('d', 'e', 'm', 'r')
|
|
ORDER BY object_kind, schema_name, object_name
|
|
`,
|
|
[],
|
|
"Integration supplemental-object inspection failed.",
|
|
);
|
|
const history = await readHistory(client);
|
|
|
|
return Object.freeze({
|
|
columns: Object.freeze(columns.rows.map((row) => Object.freeze({ ...row }))),
|
|
constraints: Object.freeze(constraints.rows.map((row) => Object.freeze({ ...row }))),
|
|
history,
|
|
objects: Object.freeze(objects.rows.map((row) => Object.freeze({ ...row }))),
|
|
sequences: Object.freeze(sequences.rows.map((row) => Object.freeze({ ...row }))),
|
|
supplementalObjects: Object.freeze(
|
|
supplementalObjects.rows.map((row) => Object.freeze({ ...row })),
|
|
),
|
|
});
|
|
}
|
|
|
|
async function columnExists(
|
|
client: Client,
|
|
tableName: string,
|
|
columnName: string,
|
|
): Promise<boolean> {
|
|
const result = await safeQuery<BooleanRow>(
|
|
client,
|
|
`
|
|
SELECT EXISTS (
|
|
SELECT 1
|
|
FROM information_schema.columns
|
|
WHERE table_schema = 'public' AND table_name = $1 AND column_name = $2
|
|
) AS value
|
|
`,
|
|
[tableName, columnName],
|
|
);
|
|
return result.rows[0]?.value === true;
|
|
}
|
|
|
|
async function captureRunnerError(
|
|
operation: Promise<unknown>,
|
|
code: MigrationRunnerErrorCode,
|
|
): Promise<MigrationRunnerError> {
|
|
try {
|
|
await operation;
|
|
} catch (error) {
|
|
if (!(error instanceof MigrationRunnerError)) {
|
|
throw harnessError("A migration operation returned an unsafe error type.");
|
|
}
|
|
const migrationError = error as MigrationRunnerError;
|
|
expect(migrationError.code).toBe(code);
|
|
return migrationError;
|
|
}
|
|
throw harnessError("A migration operation unexpectedly succeeded.");
|
|
}
|
|
|
|
function approvalFor(migrationFixture: DiscoveredModuleMigration): MigrationApproval {
|
|
return Object.freeze({
|
|
checksum: migrationFixture.checksum,
|
|
identity: migrationFixture.identity,
|
|
moduleId: migrationFixture.moduleId,
|
|
});
|
|
}
|
|
|
|
function approvalProvider(approvals: readonly MigrationApproval[]): MigrationApprovalProvider {
|
|
const frozenApprovals = Object.freeze(
|
|
approvals.map((approval) => Object.freeze({ ...approval })),
|
|
);
|
|
return Object.freeze({
|
|
getApprovals: () => Promise.resolve(frozenApprovals),
|
|
});
|
|
}
|
|
|
|
async function acquireAdministrativeLock(
|
|
client: Client,
|
|
namespace: number,
|
|
key: number,
|
|
): Promise<void> {
|
|
await safeQuery(client, "SELECT pg_advisory_lock($1::integer, $2::integer)", [namespace, key]);
|
|
}
|
|
|
|
async function releaseAdministrativeLock(
|
|
client: Client,
|
|
namespace: number,
|
|
key: number,
|
|
): Promise<void> {
|
|
const result = await safeQuery<BooleanRow>(
|
|
client,
|
|
"SELECT pg_advisory_unlock($1::integer, $2::integer) AS value",
|
|
[namespace, key],
|
|
);
|
|
if (result.rows[0]?.value !== true) {
|
|
throw harnessError("An integration advisory lock was not released.");
|
|
}
|
|
}
|
|
|
|
async function expectMigrationLockAvailable(client: Client): Promise<void> {
|
|
const result = await safeQuery<BooleanRow>(
|
|
client,
|
|
"SELECT pg_try_advisory_lock($1::integer, $2::integer) AS value",
|
|
[migrationLockNamespace, migrationLockKey],
|
|
);
|
|
expect(result.rows[0]?.value).toBe(true);
|
|
if (result.rows[0]?.value === true) {
|
|
await releaseAdministrativeLock(client, migrationLockNamespace, migrationLockKey);
|
|
}
|
|
}
|
|
|
|
async function waitForDatabaseCondition(
|
|
check: () => Promise<boolean>,
|
|
timeoutMs = 5_000,
|
|
): Promise<void> {
|
|
const deadline = performance.now() + timeoutMs;
|
|
while (performance.now() < deadline) {
|
|
if (await check()) {
|
|
return;
|
|
}
|
|
await new Promise<void>((resolve) => setTimeout(resolve, 25));
|
|
}
|
|
throw harnessError("A bounded integration database condition was not observed.");
|
|
}
|
|
|
|
describe.sequential("live PostgreSQL module migration execution", () => {
|
|
it("keeps fresh planning read-only", async () => {
|
|
await withIsolatedDatabase("fresh_plan", async (database) => {
|
|
const people = peopleInitial();
|
|
const calendar = calendarInitial();
|
|
const inventory = aggregate(
|
|
moduleFixture("people", [people]),
|
|
moduleFixture("calendar", [calendar]),
|
|
);
|
|
const before = await databaseState(database.client);
|
|
|
|
const plan = await planModuleMigrations({
|
|
aggregate: inventory,
|
|
database: createPostgresMigrationDatabase(database.url),
|
|
});
|
|
|
|
expect(plan.entries.map((entry) => [entry.identity, entry.status])).toEqual([
|
|
[people.identity, "pending"],
|
|
[calendar.identity, "pending"],
|
|
]);
|
|
expect(plan).toMatchObject({ appliedCount: 0, approvalRequiredCount: 0, pendingCount: 2 });
|
|
expect(await databaseState(database.client)).toEqual(before);
|
|
expect(await relationExists(database.client, historyTable)).toBe(false);
|
|
expect(await relationExists(database.client, "public.people_records")).toBe(false);
|
|
expect(await relationExists(database.client, "public.calendar_events")).toBe(false);
|
|
});
|
|
});
|
|
|
|
it("applies a fresh dependency-ordered inventory and records exact history", async () => {
|
|
await withIsolatedDatabase("fresh_apply", async (database) => {
|
|
const people = peopleInitial();
|
|
const calendar = calendarInitial();
|
|
const inventory = aggregate(
|
|
moduleFixture("people", [people]),
|
|
moduleFixture("calendar", [calendar]),
|
|
);
|
|
|
|
const result = await applyModuleMigrations({
|
|
aggregate: inventory,
|
|
database: createPostgresMigrationDatabase(database.url),
|
|
});
|
|
|
|
expect(result.applied.map((entry) => entry.identity)).toEqual([
|
|
people.identity,
|
|
calendar.identity,
|
|
]);
|
|
expect(await relationExists(database.client, "public.people_records")).toBe(true);
|
|
expect(await relationExists(database.client, "public.calendar_events")).toBe(true);
|
|
const history = await readHistory(database.client);
|
|
expect(history.map(({ applied_at: _appliedAt, ...row }) => row)).toEqual([
|
|
{
|
|
applied_order: 1,
|
|
checksum: people.checksum,
|
|
identity: people.identity,
|
|
migration_index: 0,
|
|
migration_tag: people.tag,
|
|
module_id: people.moduleId,
|
|
},
|
|
{
|
|
applied_order: 2,
|
|
checksum: calendar.checksum,
|
|
identity: calendar.identity,
|
|
migration_index: 0,
|
|
migration_tag: calendar.tag,
|
|
module_id: calendar.moduleId,
|
|
},
|
|
]);
|
|
expect(history).toHaveLength(2);
|
|
expect(
|
|
history.every(
|
|
(row) => row.applied_at instanceof Date && !Number.isNaN(row.applied_at.getTime()),
|
|
),
|
|
).toBe(true);
|
|
});
|
|
});
|
|
|
|
it("performs a genuine no-op without changing schema, data, sequences, or history", async () => {
|
|
await withIsolatedDatabase("no_op", async (database) => {
|
|
const initial = peopleInitial();
|
|
const inventory = aggregate(moduleFixture("people", [initial]));
|
|
const migrationDatabase = createPostgresMigrationDatabase(database.url);
|
|
await applyModuleMigrations({ aggregate: inventory, database: migrationDatabase });
|
|
await safeQuery(
|
|
database.client,
|
|
"INSERT INTO public.people_records (display_name) VALUES ($1)",
|
|
["No-op sentinel"],
|
|
);
|
|
const before = await databaseState(database.client);
|
|
const beforeData = await safeQuery<QueryResultRow>(
|
|
database.client,
|
|
"SELECT id::integer, display_name FROM public.people_records ORDER BY id",
|
|
);
|
|
|
|
const result = await applyModuleMigrations({
|
|
aggregate: inventory,
|
|
database: migrationDatabase,
|
|
});
|
|
|
|
expect(result).toEqual({ applied: [], appliedCount: 0, noOp: true });
|
|
expect(await databaseState(database.client)).toEqual(before);
|
|
const afterData = await safeQuery<QueryResultRow>(
|
|
database.client,
|
|
"SELECT id::integer, display_name FROM public.people_records ORDER BY id",
|
|
);
|
|
expect(afterData.rows).toEqual(beforeData.rows);
|
|
});
|
|
});
|
|
|
|
it("executes only a newly presented later migration", async () => {
|
|
await withIsolatedDatabase("later", async (database) => {
|
|
const initial = peopleInitial();
|
|
const later = peopleLater();
|
|
const initialInventory = aggregate(moduleFixture("people", [initial]));
|
|
const laterInventory = aggregate(moduleFixture("people", [initial, later]));
|
|
const migrationDatabase = createPostgresMigrationDatabase(database.url);
|
|
await applyModuleMigrations({ aggregate: initialInventory, database: migrationDatabase });
|
|
const originalHistory = await readHistory(database.client);
|
|
|
|
const result = await applyModuleMigrations({
|
|
aggregate: laterInventory,
|
|
database: migrationDatabase,
|
|
});
|
|
|
|
expect(result.applied.map((entry) => entry.identity)).toEqual([later.identity]);
|
|
expect(await columnExists(database.client, "people_records", "email")).toBe(true);
|
|
const history = await readHistory(database.client);
|
|
expect(history).toHaveLength(2);
|
|
expect(history[0]).toEqual(originalHistory[0]);
|
|
expect(history[1]).toMatchObject({
|
|
applied_order: 2,
|
|
checksum: later.checksum,
|
|
identity: later.identity,
|
|
migration_index: 1,
|
|
migration_tag: later.tag,
|
|
module_id: later.moduleId,
|
|
});
|
|
});
|
|
});
|
|
|
|
it("rejects edited applied contents during both planning and applying", async () => {
|
|
await withIsolatedDatabase("edited", async (database) => {
|
|
const initial = peopleInitial();
|
|
const edited = peopleInitial(
|
|
`CREATE TABLE "people_records" ("id" bigint PRIMARY KEY, "edited_marker" text);\n`,
|
|
);
|
|
const originalInventory = aggregate(moduleFixture("people", [initial]));
|
|
const editedInventory = aggregate(moduleFixture("people", [edited]));
|
|
const migrationDatabase = createPostgresMigrationDatabase(database.url);
|
|
await applyModuleMigrations({ aggregate: originalInventory, database: migrationDatabase });
|
|
const before = await databaseState(database.client);
|
|
|
|
const planError = await captureRunnerError(
|
|
planModuleMigrations({ aggregate: editedInventory, database: migrationDatabase }),
|
|
"MIGRATION_CHECKSUM_MISMATCH",
|
|
);
|
|
expect(planError.migrationIdentity).toBe(initial.identity);
|
|
expect(await databaseState(database.client)).toEqual(before);
|
|
|
|
const applyError = await captureRunnerError(
|
|
applyModuleMigrations({ aggregate: editedInventory, database: migrationDatabase }),
|
|
"MIGRATION_CHECKSUM_MISMATCH",
|
|
);
|
|
expect(applyError.migrationIdentity).toBe(initial.identity);
|
|
expect(await databaseState(database.client)).toEqual(before);
|
|
expect(await columnExists(database.client, "people_records", "edited_marker")).toBe(false);
|
|
});
|
|
});
|
|
|
|
it("rejects an unknown applied identity without repair or further execution", async () => {
|
|
await withIsolatedDatabase("unknown_history", async (database) => {
|
|
const initial = peopleInitial();
|
|
const later = peopleLater();
|
|
const initialInventory = aggregate(moduleFixture("people", [initial]));
|
|
const completeInventory = aggregate(moduleFixture("people", [initial, later]));
|
|
const migrationDatabase = createPostgresMigrationDatabase(database.url);
|
|
await applyModuleMigrations({ aggregate: initialInventory, database: migrationDatabase });
|
|
await safeQuery(
|
|
database.client,
|
|
`
|
|
INSERT INTO public.ekklesia_core_migrations
|
|
(identity, module_id, migration_tag, migration_index, checksum)
|
|
VALUES ($1, $2, $3, $4, $5)
|
|
`,
|
|
["ghost:0000_unknown", "ghost", "0000_unknown", 0, "f".repeat(64)],
|
|
);
|
|
const corrupted = await databaseState(database.client);
|
|
|
|
await captureRunnerError(
|
|
planModuleMigrations({ aggregate: completeInventory, database: migrationDatabase }),
|
|
"MIGRATION_UNKNOWN_APPLIED",
|
|
);
|
|
await captureRunnerError(
|
|
applyModuleMigrations({ aggregate: completeInventory, database: migrationDatabase }),
|
|
"MIGRATION_UNKNOWN_APPLIED",
|
|
);
|
|
|
|
expect(await databaseState(database.client)).toEqual(corrupted);
|
|
expect(await columnExists(database.client, "people_records", "email")).toBe(false);
|
|
expect((await readHistory(database.client)).map((row) => row.identity)).toEqual([
|
|
initial.identity,
|
|
"ghost:0000_unknown",
|
|
]);
|
|
});
|
|
});
|
|
|
|
it("rejects an invalid per-module history prefix without silent repair", async () => {
|
|
await withIsolatedDatabase("history_gap", async (database) => {
|
|
const initial = peopleInitial();
|
|
const later = peopleLater();
|
|
const unreachable = migration("people", 2, {
|
|
sql: `CREATE TABLE "people_unreachable" ("id" bigint PRIMARY KEY);\n`,
|
|
tag: "0002_unreachable",
|
|
});
|
|
const appliedInventory = aggregate(moduleFixture("people", [initial, later]));
|
|
const completeInventory = aggregate(moduleFixture("people", [initial, later, unreachable]));
|
|
const migrationDatabase = createPostgresMigrationDatabase(database.url);
|
|
await applyModuleMigrations({ aggregate: appliedInventory, database: migrationDatabase });
|
|
await safeQuery(
|
|
database.client,
|
|
"DELETE FROM public.ekklesia_core_migrations WHERE identity = $1",
|
|
[initial.identity],
|
|
);
|
|
const corrupted = await databaseState(database.client);
|
|
|
|
await captureRunnerError(
|
|
planModuleMigrations({ aggregate: completeInventory, database: migrationDatabase }),
|
|
"MIGRATION_HISTORY_INVALID",
|
|
);
|
|
await captureRunnerError(
|
|
applyModuleMigrations({ aggregate: completeInventory, database: migrationDatabase }),
|
|
"MIGRATION_HISTORY_INVALID",
|
|
);
|
|
|
|
expect(await databaseState(database.client)).toEqual(corrupted);
|
|
expect(await relationExists(database.client, "public.people_unreachable")).toBe(false);
|
|
expect((await readHistory(database.client)).map((row) => row.identity)).toEqual([
|
|
later.identity,
|
|
]);
|
|
});
|
|
});
|
|
|
|
it("rolls back the entire current pending batch after a real PostgreSQL failure", async () => {
|
|
await withIsolatedDatabase("rollback", async (database) => {
|
|
const committed = migration("rollback", 0, {
|
|
sql: `CREATE TABLE "rollback_committed" ("id" bigint PRIMARY KEY);\n`,
|
|
tag: "0000_committed",
|
|
});
|
|
const pendingFirst = migration("rollback", 1, {
|
|
sql: `CREATE TABLE "rollback_pending_one" ("id" bigint PRIMARY KEY);\n`,
|
|
tag: "0001_pending_one",
|
|
});
|
|
const pendingFailure = migration("rollback", 2, {
|
|
sql: `CREATE TABLE "rollback_pending_two" ("id" bigint PRIMARY KEY);\n--> statement-breakpoint\nSELECT 1 / 0;\n`,
|
|
tag: "0002_pending_failure",
|
|
});
|
|
const committedInventory = aggregate(moduleFixture("rollback", [committed]));
|
|
const completeInventory = aggregate(
|
|
moduleFixture("rollback", [committed, pendingFirst, pendingFailure]),
|
|
);
|
|
const migrationDatabase = createPostgresMigrationDatabase(database.url);
|
|
await applyModuleMigrations({ aggregate: committedInventory, database: migrationDatabase });
|
|
|
|
const error = await captureRunnerError(
|
|
applyModuleMigrations({ aggregate: completeInventory, database: migrationDatabase }),
|
|
"MIGRATION_EXECUTION_FAILED",
|
|
);
|
|
|
|
expect(error.migrationIdentity).toBe(pendingFailure.identity);
|
|
expect(await relationExists(database.client, "public.rollback_committed")).toBe(true);
|
|
expect(await relationExists(database.client, "public.rollback_pending_one")).toBe(false);
|
|
expect(await relationExists(database.client, "public.rollback_pending_two")).toBe(false);
|
|
expect((await readHistory(database.client)).map((row) => row.identity)).toEqual([
|
|
committed.identity,
|
|
]);
|
|
await expectMigrationLockAvailable(database.client);
|
|
});
|
|
});
|
|
|
|
it("honors Drizzle statement breakpoints without splitting procedural semicolons", async () => {
|
|
await withIsolatedDatabase("breakpoints", async (database) => {
|
|
const sql = `CREATE TABLE "breakpoints_steps" (
|
|
"id" integer PRIMARY KEY,
|
|
"note" text NOT NULL
|
|
);
|
|
--> statement-breakpoint
|
|
DO $body$
|
|
BEGIN
|
|
INSERT INTO "breakpoints_steps" ("id", "note") VALUES (1, 'first;still-one-statement');
|
|
END
|
|
$body$;
|
|
--> statement-breakpoint
|
|
ALTER TABLE "breakpoints_steps" ADD COLUMN "sequence_value" integer;
|
|
UPDATE "breakpoints_steps" SET "sequence_value" = 1 WHERE "id" = 1;
|
|
INSERT INTO "breakpoints_steps" ("id", "note", "sequence_value")
|
|
SELECT 2, 'second', "sequence_value" + 1 FROM "breakpoints_steps" WHERE "id" = 1;
|
|
`;
|
|
const initial = migration("breakpoints", 0, { sql, tag: "0000_ordered_steps" });
|
|
|
|
await applyModuleMigrations({
|
|
aggregate: aggregate(moduleFixture("breakpoints", [initial])),
|
|
database: createPostgresMigrationDatabase(database.url),
|
|
});
|
|
|
|
const rows = await safeQuery<QueryResultRow>(
|
|
database.client,
|
|
`SELECT id, note, sequence_value FROM public.breakpoints_steps ORDER BY id`,
|
|
);
|
|
expect(rows.rows).toEqual([
|
|
{ id: 1, note: "first;still-one-statement", sequence_value: 1 },
|
|
{ id: 2, note: "second", sequence_value: 2 },
|
|
]);
|
|
expect((await readHistory(database.client)).map((row) => row.identity)).toEqual([
|
|
initial.identity,
|
|
]);
|
|
});
|
|
});
|
|
|
|
it("serializes two real migrators with the session advisory lock", async () => {
|
|
await withIsolatedDatabase("concurrency", async (database) => {
|
|
await acquireAdministrativeLock(
|
|
database.client,
|
|
concurrencyBarrierNamespace,
|
|
concurrencyBarrierKey,
|
|
);
|
|
let barrierHeld = true;
|
|
const sql = `SELECT pg_advisory_xact_lock(${concurrencyBarrierNamespace}::integer, ${concurrencyBarrierKey}::integer);\n--> statement-breakpoint\nCREATE TABLE "concurrency_records" ("id" bigint PRIMARY KEY);\n`;
|
|
const initial = migration("concurrency", 0, { sql, tag: "0000_serialized" });
|
|
const inventory = aggregate(moduleFixture("concurrency", [initial]));
|
|
const applicationName = "ekklesia_migration_test_concurrency";
|
|
const productionUrl = databaseUrl(database.name, applicationName);
|
|
const firstApply = applyModuleMigrations({
|
|
aggregate: inventory,
|
|
database: createPostgresMigrationDatabase(productionUrl),
|
|
lockTimeoutMs: 5_000,
|
|
});
|
|
|
|
try {
|
|
await waitForDatabaseCondition(async () => {
|
|
const result = await safeQuery<CountRow>(
|
|
database.client,
|
|
`
|
|
SELECT count(*)::integer AS value
|
|
FROM pg_catalog.pg_locks AS lock
|
|
INNER JOIN pg_catalog.pg_stat_activity AS activity ON activity.pid = lock.pid
|
|
WHERE lock.locktype = 'advisory'
|
|
AND lock.classid = $1::oid
|
|
AND lock.objid = $2::oid
|
|
AND lock.objsubid = 2
|
|
AND lock.granted
|
|
AND activity.datname = current_database()
|
|
AND activity.application_name = $3
|
|
`,
|
|
[migrationLockNamespace, migrationLockKey, applicationName],
|
|
);
|
|
return result.rows[0]?.value === 1;
|
|
});
|
|
|
|
const secondError = await captureRunnerError(
|
|
applyModuleMigrations({
|
|
aggregate: inventory,
|
|
database: createPostgresMigrationDatabase(productionUrl),
|
|
lockTimeoutMs: 250,
|
|
}),
|
|
"MIGRATION_LOCK_TIMEOUT",
|
|
);
|
|
expect(secondError.migrationIdentity).toBeUndefined();
|
|
expect(await relationExists(database.client, historyTable)).toBe(false);
|
|
expect(await relationExists(database.client, "public.concurrency_records")).toBe(false);
|
|
|
|
await releaseAdministrativeLock(
|
|
database.client,
|
|
concurrencyBarrierNamespace,
|
|
concurrencyBarrierKey,
|
|
);
|
|
barrierHeld = false;
|
|
const firstResult = await firstApply;
|
|
expect(firstResult.applied.map((entry) => entry.identity)).toEqual([initial.identity]);
|
|
} finally {
|
|
if (barrierHeld) {
|
|
await releaseAdministrativeLock(
|
|
database.client,
|
|
concurrencyBarrierNamespace,
|
|
concurrencyBarrierKey,
|
|
);
|
|
}
|
|
await Promise.allSettled([firstApply]);
|
|
}
|
|
|
|
expect(await relationExists(database.client, "public.concurrency_records")).toBe(true);
|
|
expect((await readHistory(database.client)).map((row) => row.identity)).toEqual([
|
|
initial.identity,
|
|
]);
|
|
});
|
|
});
|
|
|
|
it("times out on the real lock, releases it, and permits a later apply", async () => {
|
|
await withIsolatedDatabase("lock_timeout", async (database) => {
|
|
const initial = migration("locks", 0, {
|
|
sql: `CREATE TABLE "locks_records" ("id" bigint PRIMARY KEY);\n`,
|
|
tag: "0000_initial",
|
|
});
|
|
const inventory = aggregate(moduleFixture("locks", [initial]));
|
|
const migrationDatabase = createPostgresMigrationDatabase(database.url);
|
|
await acquireAdministrativeLock(database.client, migrationLockNamespace, migrationLockKey);
|
|
let lockHeld = true;
|
|
const before = await databaseState(database.client);
|
|
const startedAt = performance.now();
|
|
try {
|
|
await captureRunnerError(
|
|
applyModuleMigrations({
|
|
aggregate: inventory,
|
|
database: migrationDatabase,
|
|
lockTimeoutMs: 250,
|
|
}),
|
|
"MIGRATION_LOCK_TIMEOUT",
|
|
);
|
|
} finally {
|
|
await releaseAdministrativeLock(database.client, migrationLockNamespace, migrationLockKey);
|
|
lockHeld = false;
|
|
}
|
|
const elapsedMs = performance.now() - startedAt;
|
|
expect(elapsedMs).toBeGreaterThanOrEqual(150);
|
|
expect(elapsedMs).toBeLessThan(2_000);
|
|
expect(await databaseState(database.client)).toEqual(before);
|
|
expect(lockHeld).toBe(false);
|
|
|
|
const result = await applyModuleMigrations({
|
|
aggregate: inventory,
|
|
database: migrationDatabase,
|
|
});
|
|
expect(result.applied.map((entry) => entry.identity)).toEqual([initial.identity]);
|
|
expect(await relationExists(database.client, "public.locks_records")).toBe(true);
|
|
});
|
|
});
|
|
|
|
it("does not strand the migration lock after execution failure", async () => {
|
|
await withIsolatedDatabase("failed_lock_release", async (database) => {
|
|
const failing = migration("lock_failure", 0, {
|
|
sql: `CREATE TABLE "lock_failure_records" ("id" bigint PRIMARY KEY);\n--> statement-breakpoint\nSELECT 1 / 0;\n`,
|
|
tag: "0000_initial",
|
|
});
|
|
const corrected = migration("lock_failure", 0, {
|
|
sql: `CREATE TABLE "lock_failure_records" ("id" bigint PRIMARY KEY);\n`,
|
|
tag: "0000_initial",
|
|
});
|
|
const migrationDatabase = createPostgresMigrationDatabase(database.url);
|
|
|
|
await captureRunnerError(
|
|
applyModuleMigrations({
|
|
aggregate: aggregate(moduleFixture("lock_failure", [failing])),
|
|
database: migrationDatabase,
|
|
}),
|
|
"MIGRATION_EXECUTION_FAILED",
|
|
);
|
|
await expectMigrationLockAvailable(database.client);
|
|
expect(await relationExists(database.client, historyTable)).toBe(false);
|
|
expect(await relationExists(database.client, "public.lock_failure_records")).toBe(false);
|
|
|
|
const result = await applyModuleMigrations({
|
|
aggregate: aggregate(moduleFixture("lock_failure", [corrected])),
|
|
database: migrationDatabase,
|
|
});
|
|
expect(result.applied.map((entry) => entry.identity)).toEqual([corrected.identity]);
|
|
});
|
|
});
|
|
|
|
it("requires an exact destructive approval before any persistent mutation", async () => {
|
|
await withIsolatedDatabase("approvals", async (database) => {
|
|
const destructive = migration("destructive", 0, {
|
|
approvalRequired: true,
|
|
sql: `CREATE TABLE "destructive_records" ("id" bigint PRIMARY KEY);\n`,
|
|
tag: "0000_replace_records",
|
|
});
|
|
const inventory = aggregate(moduleFixture("destructive", [destructive]));
|
|
const migrationDatabase = createPostgresMigrationDatabase(database.url);
|
|
const exact = approvalFor(destructive);
|
|
const rejectedApprovals: readonly (readonly MigrationApproval[] | undefined)[] = [
|
|
undefined,
|
|
[Object.freeze({ ...exact, moduleId: "wrong-module" })],
|
|
[Object.freeze({ ...exact, identity: "destructive:0000_wrong" })],
|
|
[Object.freeze({ ...exact, checksum: "0".repeat(64) })],
|
|
];
|
|
|
|
for (const supplied of rejectedApprovals) {
|
|
const before = await databaseState(database.client);
|
|
const operation =
|
|
supplied === undefined
|
|
? applyModuleMigrations({ aggregate: inventory, database: migrationDatabase })
|
|
: applyModuleMigrations({
|
|
aggregate: inventory,
|
|
approvals: approvalProvider(supplied),
|
|
database: migrationDatabase,
|
|
});
|
|
const error = await captureRunnerError(operation, "MIGRATION_APPROVAL_REQUIRED");
|
|
expect(error.migrationIdentity).toBe(destructive.identity);
|
|
expect(await databaseState(database.client)).toEqual(before);
|
|
expect(await relationExists(database.client, historyTable)).toBe(false);
|
|
expect(await relationExists(database.client, "public.destructive_records")).toBe(false);
|
|
}
|
|
|
|
const result = await applyModuleMigrations({
|
|
aggregate: inventory,
|
|
approvals: approvalProvider([exact]),
|
|
database: migrationDatabase,
|
|
});
|
|
expect(result.applied.map((entry) => entry.identity)).toEqual([destructive.identity]);
|
|
expect(await relationExists(database.client, "public.destructive_records")).toBe(true);
|
|
expect((await readHistory(database.client)).map((row) => row.identity)).toEqual([
|
|
destructive.identity,
|
|
]);
|
|
});
|
|
});
|
|
|
|
it("sanitizes a real PostgreSQL failure and normal command reporting", async () => {
|
|
await withIsolatedDatabase("sanitization", async (database) => {
|
|
const hostilePassword = "fake-password-7f7e620d";
|
|
const hostileUrl = `postgresql://hostile:${hostilePassword}@private-db.invalid/secret`;
|
|
const hostileHostname = "private-db.invalid";
|
|
const hostilePath = "/srv/ekklesia/private/migration.sql";
|
|
const hostileSqlMarker = "HOSTILE_RAW_SQL_MARKER_45f28a";
|
|
const sql = `DO $body$
|
|
BEGIN
|
|
RAISE EXCEPTION '${hostilePassword} ${hostileUrl} ${hostileHostname} ${hostilePath} ${hostileSqlMarker}';
|
|
END
|
|
$body$;\n`;
|
|
const failing = migration("sanitization", 0, {
|
|
sql,
|
|
tag: "0000_hostile_failure",
|
|
});
|
|
const capturedOutput: string[] = [];
|
|
const errorSpy = vi
|
|
.spyOn(console, "error")
|
|
.mockImplementation((...values: unknown[]) =>
|
|
capturedOutput.push(values.map(String).join(" ")),
|
|
);
|
|
const logSpy = vi
|
|
.spyOn(console, "log")
|
|
.mockImplementation((...values: unknown[]) =>
|
|
capturedOutput.push(values.map(String).join(" ")),
|
|
);
|
|
let publicError: MigrationRunnerError;
|
|
try {
|
|
publicError = await captureRunnerError(
|
|
applyModuleMigrations({
|
|
aggregate: aggregate(moduleFixture("sanitization", [failing])),
|
|
database: createPostgresMigrationDatabase(database.url),
|
|
}),
|
|
"MIGRATION_EXECUTION_FAILED",
|
|
);
|
|
reportMigrationCommandError(publicError);
|
|
} finally {
|
|
errorSpy.mockRestore();
|
|
logSpy.mockRestore();
|
|
}
|
|
|
|
const publicText = [
|
|
String(publicError),
|
|
publicError.message,
|
|
publicError.stack ?? "",
|
|
JSON.stringify(publicError),
|
|
...capturedOutput,
|
|
].join("\n");
|
|
expect(publicError.migrationIdentity).toBe(failing.identity);
|
|
expect(publicText.includes("MIGRATION_EXECUTION_FAILED")).toBe(true);
|
|
expect(publicText.includes(failing.identity)).toBe(true);
|
|
const hostileValues = [
|
|
hostilePassword,
|
|
hostileUrl,
|
|
hostileHostname,
|
|
hostilePath,
|
|
hostileSqlMarker,
|
|
"RAISE EXCEPTION",
|
|
database.url,
|
|
new URL(database.url).password,
|
|
new URL(database.url).hostname,
|
|
];
|
|
expect(hostileValues.some((hostileValue) => publicText.includes(hostileValue))).toBe(false);
|
|
expect(await relationExists(database.client, historyTable)).toBe(false);
|
|
});
|
|
});
|
|
});
|
|
|
|
const coreIdentityTables = Object.freeze([
|
|
"ekklesia_core_user",
|
|
"ekklesia_core_session",
|
|
"ekklesia_core_account",
|
|
"ekklesia_core_verification",
|
|
]);
|
|
|
|
async function loadRealCoreMigrations(): Promise<readonly DiscoveredModuleMigration[]> {
|
|
const migrations = await discoverCoreMigrations({ coreMigrationsDirectory });
|
|
if (migrations.length === 0) {
|
|
throw harnessError("The committed Core identity migration is missing.");
|
|
}
|
|
return migrations;
|
|
}
|
|
|
|
interface IndexRow extends QueryResultRow {
|
|
readonly indexname: string;
|
|
readonly tablename: string;
|
|
}
|
|
|
|
async function coreIdentityIndexNames(client: Client): Promise<readonly string[]> {
|
|
const result = await safeQuery<IndexRow>(
|
|
client,
|
|
`SELECT tablename, indexname FROM pg_catalog.pg_indexes
|
|
WHERE schemaname = 'public' AND tablename = ANY($1::text[])
|
|
ORDER BY tablename, indexname`,
|
|
[coreIdentityTables],
|
|
);
|
|
return Object.freeze(result.rows.map((row) => row.indexname));
|
|
}
|
|
|
|
describe.sequential("live PostgreSQL Core identity migration execution", () => {
|
|
it("plans the committed Core identity migration on a fresh database without mutation", async () => {
|
|
await withIsolatedDatabase("core_fresh_plan", async (database) => {
|
|
const coreMigrations = await loadRealCoreMigrations();
|
|
const inventory = aggregate(moduleFixture("core", coreMigrations));
|
|
const before = await databaseState(database.client);
|
|
|
|
const plan = await planModuleMigrations({
|
|
aggregate: inventory,
|
|
database: createPostgresMigrationDatabase(database.url),
|
|
});
|
|
|
|
expect(plan.entries.map((entry) => entry.status)).toEqual(
|
|
coreMigrations.map(() => "pending"),
|
|
);
|
|
expect(await databaseState(database.client)).toEqual(before);
|
|
expect(await relationExists(database.client, historyTable)).toBe(false);
|
|
for (const table of coreIdentityTables) {
|
|
expect(await relationExists(database.client, `public.${table}`)).toBe(false);
|
|
}
|
|
});
|
|
});
|
|
|
|
it("applies the committed Core identity migration and creates exactly the expected tables and indexes", async () => {
|
|
await withIsolatedDatabase("core_apply", async (database) => {
|
|
const coreMigrations = await loadRealCoreMigrations();
|
|
const inventory = aggregate(moduleFixture("core", coreMigrations));
|
|
|
|
const result = await applyModuleMigrations({
|
|
aggregate: inventory,
|
|
database: createPostgresMigrationDatabase(database.url),
|
|
});
|
|
|
|
expect(result.applied.map((entry) => entry.identity)).toEqual(
|
|
coreMigrations.map((entry) => entry.identity),
|
|
);
|
|
for (const table of coreIdentityTables) {
|
|
expect(await relationExists(database.client, `public.${table}`)).toBe(true);
|
|
}
|
|
expect(await coreIdentityIndexNames(database.client)).toEqual([
|
|
"ekklesia_core_account_pkey",
|
|
"ekklesia_core_account_user_id_idx",
|
|
"ekklesia_core_session_pkey",
|
|
"ekklesia_core_session_token_unique",
|
|
"ekklesia_core_session_user_id_idx",
|
|
"ekklesia_core_user_email_unique",
|
|
"ekklesia_core_user_pkey",
|
|
"ekklesia_core_verification_identifier_idx",
|
|
"ekklesia_core_verification_pkey",
|
|
]);
|
|
|
|
const history = await readHistory(database.client);
|
|
expect(history).toHaveLength(coreMigrations.length);
|
|
expect(history.map(({ applied_at: _appliedAt, ...row }) => row)).toEqual(
|
|
coreMigrations.map((entry, index) => ({
|
|
applied_order: index + 1,
|
|
checksum: entry.checksum,
|
|
identity: entry.identity,
|
|
migration_index: entry.index,
|
|
migration_tag: entry.tag,
|
|
module_id: "core",
|
|
})),
|
|
);
|
|
});
|
|
});
|
|
|
|
it("performs a genuine no-op on a second Core identity apply", async () => {
|
|
await withIsolatedDatabase("core_no_op", async (database) => {
|
|
const coreMigrations = await loadRealCoreMigrations();
|
|
const inventory = aggregate(moduleFixture("core", coreMigrations));
|
|
const migrationDatabase = createPostgresMigrationDatabase(database.url);
|
|
await applyModuleMigrations({ aggregate: inventory, database: migrationDatabase });
|
|
const before = await databaseState(database.client);
|
|
|
|
const result = await applyModuleMigrations({
|
|
aggregate: inventory,
|
|
database: migrationDatabase,
|
|
});
|
|
|
|
expect(result).toEqual({ applied: [], appliedCount: 0, noOp: true });
|
|
expect(await databaseState(database.client)).toEqual(before);
|
|
});
|
|
});
|
|
|
|
it("coexists with a fixture module migration, executing Core first as documented", async () => {
|
|
await withIsolatedDatabase("core_and_module", async (database) => {
|
|
const coreMigrations = await loadRealCoreMigrations();
|
|
const people = peopleInitial();
|
|
const inventory = aggregate(
|
|
moduleFixture("core", coreMigrations),
|
|
moduleFixture("people", [people]),
|
|
);
|
|
|
|
const result = await applyModuleMigrations({
|
|
aggregate: inventory,
|
|
database: createPostgresMigrationDatabase(database.url),
|
|
});
|
|
|
|
expect(result.applied.map((entry) => entry.identity)).toEqual([
|
|
...coreMigrations.map((entry) => entry.identity),
|
|
people.identity,
|
|
]);
|
|
for (const table of coreIdentityTables) {
|
|
expect(await relationExists(database.client, `public.${table}`)).toBe(true);
|
|
}
|
|
expect(await relationExists(database.client, "public.people_records")).toBe(true);
|
|
const history = await readHistory(database.client);
|
|
expect(history.map((row) => row.module_id)).toEqual([
|
|
...coreMigrations.map(() => "core"),
|
|
"people",
|
|
]);
|
|
});
|
|
});
|
|
|
|
it("does not corrupt previously valid module-only history when Core is introduced later", async () => {
|
|
await withIsolatedDatabase("core_added_later", async (database) => {
|
|
const people = peopleInitial();
|
|
const migrationDatabase = createPostgresMigrationDatabase(database.url);
|
|
await applyModuleMigrations({
|
|
aggregate: aggregate(moduleFixture("people", [people])),
|
|
database: migrationDatabase,
|
|
});
|
|
const historyBeforeCore = await readHistory(database.client);
|
|
|
|
const coreMigrations = await loadRealCoreMigrations();
|
|
const result = await applyModuleMigrations({
|
|
aggregate: aggregate(
|
|
moduleFixture("core", coreMigrations),
|
|
moduleFixture("people", [people]),
|
|
),
|
|
database: migrationDatabase,
|
|
});
|
|
|
|
expect(result.applied.map((entry) => entry.identity)).toEqual(
|
|
coreMigrations.map((entry) => entry.identity),
|
|
);
|
|
const historyAfterCore = await readHistory(database.client);
|
|
const peopleRowBefore = historyBeforeCore.find((row) => row.module_id === "people");
|
|
const peopleRowAfter = historyAfterCore.find((row) => row.module_id === "people");
|
|
expect(peopleRowAfter).toEqual(peopleRowBefore);
|
|
expect(historyAfterCore).toHaveLength(historyBeforeCore.length + coreMigrations.length);
|
|
});
|
|
});
|
|
|
|
it("does not damage the committed Core identity migration when a later module migration fails", async () => {
|
|
await withIsolatedDatabase("core_rollback_safety", async (database) => {
|
|
const coreMigrations = await loadRealCoreMigrations();
|
|
const migrationDatabase = createPostgresMigrationDatabase(database.url);
|
|
await applyModuleMigrations({
|
|
aggregate: aggregate(moduleFixture("core", coreMigrations)),
|
|
database: migrationDatabase,
|
|
});
|
|
const historyAfterCore = await readHistory(database.client);
|
|
|
|
const failing = migration("broken", 0, {
|
|
sql: `CREATE TABLE "broken_records" ("id" bigint PRIMARY KEY);\n--> statement-breakpoint\nSELECT 1 / 0;\n`,
|
|
tag: "0000_broken",
|
|
});
|
|
|
|
const error = await captureRunnerError(
|
|
applyModuleMigrations({
|
|
aggregate: aggregate(
|
|
moduleFixture("core", coreMigrations),
|
|
moduleFixture("broken", [failing]),
|
|
),
|
|
database: migrationDatabase,
|
|
}),
|
|
"MIGRATION_EXECUTION_FAILED",
|
|
);
|
|
|
|
expect(error.migrationIdentity).toBe(failing.identity);
|
|
for (const table of coreIdentityTables) {
|
|
expect(await relationExists(database.client, `public.${table}`)).toBe(true);
|
|
}
|
|
expect(await relationExists(database.client, "public.broken_records")).toBe(false);
|
|
expect(await readHistory(database.client)).toEqual(historyAfterCore);
|
|
await expectMigrationLockAvailable(database.client);
|
|
});
|
|
});
|
|
});
|