diff --git a/apps/server/src/persistence/CustomMigrationCompatibility.test.ts b/apps/server/src/persistence/CustomMigrationCompatibility.test.ts new file mode 100644 index 00000000000..9be7476268a --- /dev/null +++ b/apps/server/src/persistence/CustomMigrationCompatibility.test.ts @@ -0,0 +1,162 @@ +import { assert, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +import { + CUSTOM_WORK_LANES_MIGRATION_ID, + LEGACY_WORK_LANES_MIGRATION_NAME, + runCustomMigrationCompatibilityBridge, +} from "./CustomMigrationCompatibility.ts"; +import { runMigrations } from "./Migrations.ts"; +import * as NodeSqliteClient from "./NodeSqliteClient.ts"; + +const createLegacyPrivateTables = (sql: SqlClient.SqlClient) => + Effect.gen(function* () { + yield* sql`CREATE TABLE projection_work_lanes (id TEXT PRIMARY KEY)`; + yield* sql`CREATE TABLE projection_source_truth_revisions (id TEXT PRIMARY KEY)`; + yield* sql`CREATE TABLE projection_lane_acceptance_criteria (id TEXT PRIMARY KEY)`; + }); + +const seedLegacyMigration35 = (sql: SqlClient.SqlClient) => + sql` + INSERT INTO effect_sql_migrations (migration_id, name) + VALUES (35, ${LEGACY_WORK_LANES_MIGRATION_NAME}) + `; + +const assertIntegrity = Effect.fn("customMigrationCompatibility.assertIntegrity")(function* () { + const sql = yield* SqlClient.SqlClient; + const rows = yield* sql<{ readonly integrity_check: string }>`PRAGMA integrity_check`; + assert.equal(rows[0]?.integrity_check, "ok"); +}); + +const freshOfficialLayer = it.layer(NodeSqliteClient.layerMemory()); +freshOfficialLayer("custom migration compatibility: fresh official database", (it) => { + it.effect("preserves the official migration-35 path", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + yield* runMigrations(); + + const migration35 = yield* sql<{ readonly name: string }>` + SELECT name FROM effect_sql_migrations WHERE migration_id = 35 + `; + assert.equal(migration35[0]?.name, "ProjectionThreadTitleRegeneration"); + + const columns = yield* sql<{ readonly name: string }>` + PRAGMA table_info(projection_threads) + `; + const names = new Set(columns.map((column) => column.name)); + assert.ok(names.has("title_regeneration_request_id")); + assert.ok(names.has("title_regeneration_started_at")); + + const customRows = yield* sql<{ readonly count: number }>` + SELECT COUNT(*) AS count FROM t3_custom_schema_migrations + `; + assert.equal(customRows[0]?.count, 0); + yield* assertIntegrity(); + }), + ); +}); + +const legacyCustomLayer = it.layer(NodeSqliteClient.layerMemory()); +legacyCustomLayer("custom migration compatibility: legacy custom database", (it) => { + it.effect("keeps historical row 35, repairs upstream 35, and continues through 43", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + yield* runMigrations({ toMigrationInclusive: 34 }); + yield* createLegacyPrivateTables(sql); + yield* seedLegacyMigration35(sql); + + yield* runMigrations(); + + const migration35 = yield* sql<{ readonly name: string }>` + SELECT name FROM effect_sql_migrations WHERE migration_id = 35 + `; + assert.equal(migration35[0]?.name, LEGACY_WORK_LANES_MIGRATION_NAME); + + const latestMigration = yield* sql<{ readonly migration_id: number; readonly name: string }>` + SELECT migration_id, name + FROM effect_sql_migrations + ORDER BY migration_id DESC + LIMIT 1 + `; + assert.equal(latestMigration[0]?.migration_id, 43); + assert.equal(latestMigration[0]?.name, "ProjectionThreadsUnsettledAt"); + + const columns = yield* sql<{ readonly name: string }>` + PRAGMA table_info(projection_threads) + `; + const names = new Set(columns.map((column) => column.name)); + assert.ok(names.has("title_regeneration_request_id")); + assert.ok(names.has("title_regeneration_started_at")); + + const adopted = yield* sql<{ readonly name: string }>` + SELECT name + FROM t3_custom_schema_migrations + WHERE migration_id = ${CUSTOM_WORK_LANES_MIGRATION_ID} + `; + assert.equal(adopted[0]?.name, LEGACY_WORK_LANES_MIGRATION_NAME); + yield* assertIntegrity(); + }), + ); +}); + +const alreadyBridgedLayer = it.layer(NodeSqliteClient.layerMemory()); +alreadyBridgedLayer("custom migration compatibility: already bridged database", (it) => { + it.effect("is idempotent", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + yield* runMigrations({ toMigrationInclusive: 34 }); + yield* createLegacyPrivateTables(sql); + yield* seedLegacyMigration35(sql); + + const first = yield* runCustomMigrationCompatibilityBridge(); + const second = yield* runCustomMigrationCompatibilityBridge(); + + assert.deepEqual(first.addedTitleRegenerationColumns, [ + "title_regeneration_request_id", + "title_regeneration_started_at", + ]); + assert.deepEqual(second.addedTitleRegenerationColumns, []); + assert.equal(second.legacyWorkLanesMigrationDetected, true); + assert.equal(second.adoptedLegacyWorkLanesMigration, true); + + const columns = yield* sql<{ readonly name: string }>` + PRAGMA table_info(projection_threads) + `; + assert.equal( + columns.filter((column) => column.name === "title_regeneration_request_id").length, + 1, + ); + assert.equal( + columns.filter((column) => column.name === "title_regeneration_started_at").length, + 1, + ); + + const customRows = yield* sql<{ readonly count: number }>` + SELECT COUNT(*) AS count + FROM t3_custom_schema_migrations + WHERE migration_id = ${CUSTOM_WORK_LANES_MIGRATION_ID} + `; + assert.equal(customRows[0]?.count, 1); + yield* assertIntegrity(); + }), + ); +}); + +const malformedLegacyLayer = it.layer(NodeSqliteClient.layerMemory()); +malformedLegacyLayer("custom migration compatibility: malformed legacy database", (it) => { + it.effect("fails closed when the historical ledger row exists without its private tables", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + yield* runMigrations({ toMigrationInclusive: 34 }); + yield* seedLegacyMigration35(sql); + + const error = yield* runCustomMigrationCompatibilityBridge().pipe(Effect.flip); + assert.match(String(error), /required private tables are missing/i); + }), + ); +}); diff --git a/apps/server/src/persistence/CustomMigrationCompatibility.ts b/apps/server/src/persistence/CustomMigrationCompatibility.ts new file mode 100644 index 00000000000..c10e17e3dd4 --- /dev/null +++ b/apps/server/src/persistence/CustomMigrationCompatibility.ts @@ -0,0 +1,159 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +export const CUSTOM_MIGRATIONS_TABLE = "t3_custom_schema_migrations"; +export const LEGACY_WORK_LANES_MIGRATION_ID = 35; +export const LEGACY_WORK_LANES_MIGRATION_NAME = "WorkLanesAndSourceTruth"; +export const CUSTOM_WORK_LANES_MIGRATION_ID = 1; + +const REQUIRED_LEGACY_WORK_LANE_TABLES = [ + "projection_work_lanes", + "projection_source_truth_revisions", + "projection_lane_acceptance_criteria", +] as const; + +export interface CustomMigrationCompatibilityResult { + readonly legacyWorkLanesMigrationDetected: boolean; + readonly addedTitleRegenerationColumns: ReadonlyArray; + readonly adoptedLegacyWorkLanesMigration: boolean; +} + +const tableExists = Effect.fn("customMigrationCompatibility.tableExists")(function* (name: string) { + const sql = yield* SqlClient.SqlClient; + const rows = yield* sql<{ readonly name: string }>` + SELECT name + FROM sqlite_master + WHERE type = 'table' AND name = ${name} + LIMIT 1 + `; + return rows.length > 0; +}); + +const ensureCustomMigrationTable = Effect.fn( + "customMigrationCompatibility.ensureCustomMigrationTable", +)(function* () { + const sql = yield* SqlClient.SqlClient; + yield* sql` + CREATE TABLE IF NOT EXISTS t3_custom_schema_migrations ( + migration_id INTEGER PRIMARY KEY NOT NULL, + name VARCHAR(255) NOT NULL, + created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP + ) + `; +}); + +/** + * Bridge the one known collision between the historical custom fork and upstream. + * + * The old custom fork recorded migration 35 as WorkLanesAndSourceTruth in + * effect_sql_migrations. Upstream v0.0.35 also uses id 35, for the title-regeneration + * columns. Effect's migrator skips ids <= the latest recorded id, so an old custom + * database would otherwise skip upstream 35 and later fail when code reads the + * missing columns. + * + * This bridge deliberately leaves effect_sql_migrations row 35 untouched. When that row + * is the legacy custom migration, it installs only the idempotent upstream-35 schema + * additions, verifies the private tables implied by the historical row, and adopts the + * private migration into a separate custom namespace for future fork-only migrations. + */ +export const runCustomMigrationCompatibilityBridge = Effect.fn( + "runCustomMigrationCompatibilityBridge", +)(function* () { + const sql = yield* SqlClient.SqlClient; + + yield* ensureCustomMigrationTable(); + + if (!(yield* tableExists("effect_sql_migrations"))) { + return { + legacyWorkLanesMigrationDetected: false, + addedTitleRegenerationColumns: [], + adoptedLegacyWorkLanesMigration: false, + } satisfies CustomMigrationCompatibilityResult; + } + + const legacyRows = yield* sql<{ readonly name: string }>` + SELECT name + FROM effect_sql_migrations + WHERE migration_id = ${LEGACY_WORK_LANES_MIGRATION_ID} + LIMIT 1 + `; + const legacyName = legacyRows[0]?.name; + if (legacyName !== LEGACY_WORK_LANES_MIGRATION_NAME) { + return { + legacyWorkLanesMigrationDetected: false, + addedTitleRegenerationColumns: [], + adoptedLegacyWorkLanesMigration: false, + } satisfies CustomMigrationCompatibilityResult; + } + + if (!(yield* tableExists("projection_threads"))) { + return yield* Effect.fail( + new Error( + "Legacy migration 35_WorkLanesAndSourceTruth is recorded, but projection_threads is missing", + ), + ); + } + + const missingPrivateTables: Array = []; + for (const table of REQUIRED_LEGACY_WORK_LANE_TABLES) { + if (!(yield* tableExists(table))) { + missingPrivateTables.push(table); + } + } + if (missingPrivateTables.length > 0) { + return yield* Effect.fail( + new Error( + `Legacy migration 35_WorkLanesAndSourceTruth is recorded, but required private tables are missing: ${missingPrivateTables.join(", ")}`, + ), + ); + } + + const columns = yield* sql<{ readonly name: string }>` + PRAGMA table_info(projection_threads) + `; + const columnNames = new Set(columns.map((column) => column.name)); + const addedTitleRegenerationColumns: Array = []; + + if (!columnNames.has("title_regeneration_request_id")) { + yield* sql` + ALTER TABLE projection_threads + ADD COLUMN title_regeneration_request_id TEXT + `; + addedTitleRegenerationColumns.push("title_regeneration_request_id"); + } + + if (!columnNames.has("title_regeneration_started_at")) { + yield* sql` + ALTER TABLE projection_threads + ADD COLUMN title_regeneration_started_at TEXT + `; + addedTitleRegenerationColumns.push("title_regeneration_started_at"); + } + + yield* sql` + INSERT INTO t3_custom_schema_migrations (migration_id, name) + VALUES (${CUSTOM_WORK_LANES_MIGRATION_ID}, ${LEGACY_WORK_LANES_MIGRATION_NAME}) + ON CONFLICT (migration_id) DO NOTHING + `; + + const adoptedRows = yield* sql<{ readonly name: string }>` + SELECT name + FROM t3_custom_schema_migrations + WHERE migration_id = ${CUSTOM_WORK_LANES_MIGRATION_ID} + LIMIT 1 + `; + const adoptedName = adoptedRows[0]?.name; + if (adoptedName !== LEGACY_WORK_LANES_MIGRATION_NAME) { + return yield* Effect.fail( + new Error( + `Custom migration namespace collision at id ${CUSTOM_WORK_LANES_MIGRATION_ID}: expected ${LEGACY_WORK_LANES_MIGRATION_NAME}, found ${adoptedName ?? "missing"}`, + ), + ); + } + + return { + legacyWorkLanesMigrationDetected: true, + addedTitleRegenerationColumns, + adoptedLegacyWorkLanesMigration: true, + } satisfies CustomMigrationCompatibilityResult; +}); diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 8abbe87fce3..86a72cb31e1 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -12,6 +12,8 @@ import * as Migrator from "effect/unstable/sql/Migrator"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; +import { runCustomMigrationCompatibilityBridge } from "./CustomMigrationCompatibility.ts"; + // Import all migrations statically import Migration0001 from "./Migrations/001_OrchestrationEvents.ts"; import Migration0002 from "./Migrations/002_OrchestrationCommandReceipts.ts"; @@ -147,6 +149,7 @@ export interface RunMigrationsOptions { export const runMigrations = Effect.fn("runMigrations")(function* ({ toMigrationInclusive, }: RunMigrationsOptions = {}) { + yield* runCustomMigrationCompatibilityBridge(); const executedMigrations = yield* run({ loader: makeMigrationLoader(toMigrationInclusive) }); const migrations = executedMigrations.map(([id, name]) => `${id}_${name}`); yield* migrations.length === 0