fix(db): enforce runtime external schema upgrades

src/server/db/runtimeSchemaBootstrap.live.test.ts

@@ -0,0 +1,111 @@
+import baselineContract from './generated/fixtures/2026-03-14-baseline.schemaContract.json' with { type: 'json' };
+import currentContract from './generated/schemaContract.json' with { type: 'json' };
+import mysql from 'mysql2/promise';
+import pg from 'pg';
+import { describe, expect, it } from 'vitest';
+import { generateBootstrapSql } from './schemaArtifactGenerator.js';
+import { __schemaIntrospectionTestUtils, introspectLiveSchema } from './schemaIntrospection.js';
+import { bootstrapRuntimeDatabaseSchema } from './runtimeSchemaBootstrap.js';
+
+const mysqlRuntime = process.env.DB_PARITY_MYSQL_URL ? it : it.skip;
+const postgresRuntime = process.env.DB_PARITY_POSTGRES_URL ? it : it.skip;
+
+async function resetMySqlSchema(connectionString: string): Promise<void> {
+  const connection = await mysql.createConnection({ uri: connectionString });
+  try {
+    await connection.query('SET FOREIGN_KEY_CHECKS = 0');
+    const [rows] = await connection.query(`
+      SELECT table_name AS table_name
+      FROM information_schema.tables
+      WHERE table_schema = DATABASE()
+        AND table_type = 'BASE TABLE'
+    `);
+    for (const row of rows as Array<Record<string, unknown>>) {
+      const tableName = String(row.table_name || '');
+      if (!tableName) continue;
+      await connection.query(`DROP TABLE IF EXISTS \`${tableName}\``);
+    }
+    await connection.query('SET FOREIGN_KEY_CHECKS = 1');
+  } finally {
+    await connection.end();
+  }
+}
+
+async function applyMySqlStatements(connectionString: string, statements: string[]): Promise<void> {
+  const connection = await mysql.createConnection({ uri: connectionString });
+  try {
+    for (const statement of statements) {
+      await connection.query(statement);
+    }
+  } finally {
+    await connection.end();
+  }
+}
+
+async function resetPostgresSchema(connectionString: string): Promise<void> {
+  const client = new pg.Client({ connectionString });
+  await client.connect();
+  try {
+    const result = await client.query(`
+      SELECT tablename
+      FROM pg_tables
+      WHERE schemaname = current_schema()
+      ORDER BY tablename ASC
+    `);
+    for (const row of result.rows as Array<{ tablename: string }>) {
+      await client.query(`DROP TABLE IF EXISTS "${row.tablename}" CASCADE`);
+    }
+  } finally {
+    await client.end();
+  }
+}
+
+async function applyPostgresStatements(connectionString: string, statements: string[]): Promise<void> {
+  const client = new pg.Client({ connectionString });
+  await client.connect();
+  try {
+    for (const statement of statements) {
+      await client.query(statement);
+    }
+  } finally {
+    await client.end();
+  }
+}
+
+describe('runtime schema bootstrap live upgrade path', () => {
+  mysqlRuntime('upgrades mysql runtime schemas from an older live contract', async () => {
+    const connectionString = process.env.DB_PARITY_MYSQL_URL!;
+    const baselineStatements = __schemaIntrospectionTestUtils.splitSqlStatements(
+      generateBootstrapSql('mysql', baselineContract),
+    );
+
+    await resetMySqlSchema(connectionString);
+    await applyMySqlStatements(connectionString, baselineStatements);
+
+    await bootstrapRuntimeDatabaseSchema({
+      dialect: 'mysql',
+      connectionString,
+    });
+
+    const live = await introspectLiveSchema({ dialect: 'mysql', connectionString });
+    expect(live).toEqual(currentContract);
+  });
+
+  postgresRuntime('upgrades postgres runtime schemas from an older live contract', async () => {
+    const connectionString = process.env.DB_PARITY_POSTGRES_URL!;
+    const baselineStatements = __schemaIntrospectionTestUtils.splitSqlStatements(
+      generateBootstrapSql('postgres', baselineContract),
+    );
+
+    await resetPostgresSchema(connectionString);
+    await applyPostgresStatements(connectionString, baselineStatements);
+
+    await bootstrapRuntimeDatabaseSchema({
+      dialect: 'postgres',
+      connectionString,
+    });
+
+    const live = await introspectLiveSchema({ dialect: 'postgres', connectionString });
+    expect(live).toEqual(currentContract);
+  });
+});

src/server/db/runtimeSchemaBootstrap.test.ts

@@ -1,3 +1,7 @@
+import baselineContract from './generated/fixtures/2026-03-14-baseline.schemaContract.json' with { type: 'json' };
+import currentContract from './generated/schemaContract.json' with { type: 'json' };
+import { classifyLegacyCompatMutation } from './legacySchemaCompat.js';
+import { generateUpgradeSql } from './schemaArtifactGenerator.js';
 import { describe, expect, it } from 'vitest';
 import {
   __runtimeSchemaBootstrapTestUtils,
@@ -30,73 +34,81 @@ function createStubClient(dialect: RuntimeSchemaDialect, executedSql: string[]):
 }
 
 describe('runtime schema bootstrap', () => {
-  it.each(['mysql', 'postgres'] as const)('loads generated bootstrap statements for %s', async (dialect) => {
+  it.each(['mysql', 'postgres'] as const)('executes live-schema upgrade statements for %s', async (dialect) => {
     const executedSql: string[] = [];
-    const expectedBootstrapSql = __runtimeSchemaBootstrapTestUtils.readGeneratedBootstrapStatements(dialect);
+    const expectedUpgradeSql = __runtimeSchemaBootstrapTestUtils.splitSqlStatements(
+      generateUpgradeSql(dialect, currentContract, baselineContract),
+    );
 
-    await ensureRuntimeDatabaseSchema(createStubClient(dialect, executedSql));
+    await ensureRuntimeDatabaseSchema(createStubClient(dialect, executedSql), {
+      currentContract,
+      liveContract: baselineContract,
+    });
 
-    expect(executedSql.slice(0, expectedBootstrapSql.length)).toEqual(expectedBootstrapSql);
+    expect(executedSql.slice(0, expectedUpgradeSql.length)).toEqual(expectedUpgradeSql);
   });
 
-  it('ignores duplicate mysql index errors when replaying bootstrap statements', async () => {
+  it('skips external schema execution when live schema already matches the current contract', async () => {
     const executedSql: string[] = [];
-    const targetSql = 'CREATE UNIQUE INDEX `model_availability_account_model_unique` ON `model_availability` (`account_id`, `model_name`(191))';
+    const expectedUpgradeSql = __runtimeSchemaBootstrapTestUtils.buildExternalUpgradeStatements(
+      'mysql',
+      currentContract,
+      currentContract,
+    );
+
+    await ensureRuntimeDatabaseSchema(createStubClient('mysql', executedSql), {
+      currentContract,
+      liveContract: currentContract,
+    });
+
+    expect(expectedUpgradeSql).toEqual([]);
+    expect(executedSql.every((sqlText) => classifyLegacyCompatMutation(sqlText) === 'legacy')).toBe(true);
+  });
+
+  it('ignores duplicate mysql index and column errors when replaying additive schema statements', async () => {
+    const executedSql: string[] = [];
+    const duplicateColumnSql = __runtimeSchemaBootstrapTestUtils.splitSqlStatements(
+      generateUpgradeSql('mysql', currentContract, baselineContract),
+    ).find((sqlText) => sqlText.includes('ALTER TABLE `model_availability` ADD COLUMN `is_manual`'));
+    const duplicateIndexSql = __runtimeSchemaBootstrapTestUtils.splitSqlStatements(
+      generateUpgradeSql('mysql', currentContract, baselineContract),
+    ).find((sqlText) => sqlText.includes('proxy_files_public_id_unique'));
+
+    expect(duplicateColumnSql).toBeDefined();
+    expect(duplicateIndexSql).toBeDefined();
 
     await ensureRuntimeDatabaseSchema({
       ...createStubClient('mysql', executedSql),
       execute: async (sqlText: string) => {
         executedSql.push(sqlText);
-        if (sqlText === targetSql) {
+        if (sqlText === duplicateColumnSql) {
+          const error = new Error("Duplicate column name 'is_manual'") as Error & { code?: string };
+          error.code = 'ER_DUP_FIELDNAME';
+          throw error;
+        }
+        if (sqlText === duplicateIndexSql) {
           const error = new Error("Duplicate key name 'model_availability_account_model_unique'") as Error & { code?: string };
           error.code = 'ER_DUP_KEYNAME';
           throw error;
         }
         return [];
       },
+    }, {
+      currentContract,
+      liveContract: baselineContract,
     });
 
-    expect(executedSql).toContain(targetSql);
+    expect(executedSql).toContain(duplicateColumnSql);
+    expect(executedSql).toContain(duplicateIndexSql);
   });
 
-  it('replays the mysql bootstrap against an already-initialized schema', async () => {
+  it('ignores postgres relation-already-exists errors when replaying additive schema statements', async () => {
     const executedSql: string[] = [];
-    const createdStatements = new Set<string>();
+    const targetSql = __runtimeSchemaBootstrapTestUtils.splitSqlStatements(
+      generateUpgradeSql('postgres', currentContract, baselineContract),
+    ).find((sqlText) => sqlText.includes('proxy_files_public_id_unique'));
 
-    const mysqlClient = createStubClient('mysql', executedSql);
-    mysqlClient.execute = async (sqlText: string) => {
-      executedSql.push(sqlText);
-      const normalized = sqlText.trim().toLowerCase();
-      const createsSchemaObject = normalized.startsWith('create table if not exists')
-        || normalized.startsWith('create index')
-        || normalized.startsWith('create unique index');
-
-      if (createsSchemaObject) {
-        if (createdStatements.has(sqlText)) {
-          const error = new Error(
-            normalized.startsWith('create table')
-              ? 'Table already exists'
-              : 'Duplicate key name during bootstrap replay',
-          ) as Error & { code?: string };
-          error.code = normalized.startsWith('create table') ? 'ER_TABLE_EXISTS_ERROR' : 'ER_DUP_KEYNAME';
-          throw error;
-        }
-        createdStatements.add(sqlText);
-      }
-
-      return [];
-    };
-
-    await ensureRuntimeDatabaseSchema(mysqlClient);
-    await ensureRuntimeDatabaseSchema(mysqlClient);
-
-    expect(executedSql.length).toBeGreaterThan(createdStatements.size);
-    expect(createdStatements.size).toBeGreaterThan(0);
-  });
-
-  it('ignores postgres relation-already-exists errors when replaying bootstrap statements', async () => {
-    const executedSql: string[] = [];
-    const targetSql = 'CREATE UNIQUE INDEX "model_availability_account_model_unique" ON "model_availability" ("account_id", "model_name")';
+    expect(targetSql).toBeDefined();
 
     await ensureRuntimeDatabaseSchema({
       ...createStubClient('postgres', executedSql),
@@ -109,6 +121,9 @@ describe('runtime schema bootstrap', () => {
         }
         return [];
       },
+    }, {
+      currentContract,
+      liveContract: baselineContract,
     });
 
     expect(executedSql).toContain(targetSql);

src/server/db/runtimeSchemaBootstrap.ts

@@ -7,13 +7,16 @@ import {
   ensureLegacySchemaCompatibility,
   type LegacySchemaCompatInspector,
 } from './legacySchemaCompat.js';
-import { generateBootstrapSql, resolveGeneratedArtifactPath } from './schemaArtifactGenerator.js';
-import type { SchemaContract } from './schemaContract.js';
+import { generateBootstrapSql, generateUpgradeSql } from './schemaArtifactGenerator.js';
+import { introspectLiveSchema } from './schemaIntrospection.js';
+import { resolveGeneratedSchemaContractPath, type SchemaContract } from './schemaContract.js';
 
 export type RuntimeSchemaDialect = 'sqlite' | 'mysql' | 'postgres';
 
 export interface RuntimeSchemaClient {
   dialect: RuntimeSchemaDialect;
+  connectionString: string;
+  ssl: boolean;
   begin(): Promise<void>;
   commit(): Promise<void>;
   rollback(): Promise<void>;
@@ -42,10 +45,13 @@ function isExistingSchemaObjectError(error: unknown): boolean {
     : '';
 
   return code === 'ER_DUP_KEYNAME'
+    || code === 'ER_DUP_FIELDNAME'
     || code === 'ER_TABLE_EXISTS_ERROR'
     || code === '42P07'
+    || code === '42701'
     || code === '42710'
     || lowered.includes('already exists')
+    || lowered.includes('duplicate column')
     || lowered.includes('duplicate key name')
     || lowered.includes('relation') && lowered.includes('already exists');
 }
@@ -144,12 +150,7 @@ function splitSqlStatements(sqlText: string): string[] {
 }
 
 function readSchemaContract(): SchemaContract {
-  return JSON.parse(readFileSync(resolveGeneratedArtifactPath('schemaContract.json'), 'utf8')) as SchemaContract;
-}
-
-function readGeneratedBootstrapStatements(dialect: Exclude<RuntimeSchemaDialect, 'sqlite'>): string[] {
-  const filename = dialect === 'mysql' ? 'mysql.bootstrap.sql' : 'postgres.bootstrap.sql';
-  return splitSqlStatements(readFileSync(resolveGeneratedArtifactPath(filename), 'utf8'));
+  return JSON.parse(readFileSync(resolveGeneratedSchemaContractPath(), 'utf8')) as SchemaContract;
 }
 
 async function createPostgresClient(connectionString: string, ssl: boolean): Promise<RuntimeSchemaClient> {
@@ -162,6 +163,8 @@ async function createPostgresClient(connectionString: string, ssl: boolean): Pro
 
   return {
     dialect: 'postgres',
+    connectionString,
+    ssl,
     begin: async () => { await client.query('BEGIN'); },
     commit: async () => { await client.query('COMMIT'); },
     rollback: async () => { await client.query('ROLLBACK'); },
@@ -185,6 +188,8 @@ async function createMySqlClient(connectionString: string, ssl: boolean): Promis
 
   return {
     dialect: 'mysql',
+    connectionString,
+    ssl,
     begin: async () => { await connection.beginTransaction(); },
     commit: async () => { await connection.commit(); },
     rollback: async () => { await connection.rollback(); },
@@ -210,6 +215,8 @@ async function createSqliteClient(connectionString: string): Promise<RuntimeSche
 
   return {
     dialect: 'sqlite',
+    connectionString,
+    ssl: false,
     begin: async () => { sqlite.exec('BEGIN'); },
     commit: async () => { sqlite.exec('COMMIT'); },
     rollback: async () => { sqlite.exec('ROLLBACK'); },
@@ -238,10 +245,43 @@ export async function createRuntimeSchemaClient(input: RuntimeSchemaConnectionIn
   return createSqliteClient(input.connectionString);
 }
 
-export async function ensureRuntimeDatabaseSchema(client: RuntimeSchemaClient): Promise<void> {
+type EnsureRuntimeDatabaseSchemaOptions = {
+  currentContract?: SchemaContract;
+  liveContract?: SchemaContract;
+};
+
+async function resolveLiveContract(client: RuntimeSchemaClient, liveContract?: SchemaContract): Promise<SchemaContract> {
+  if (liveContract) {
+    return liveContract;
+  }
+
+  return introspectLiveSchema({
+    dialect: client.dialect,
+    connectionString: client.connectionString,
+    ssl: client.ssl,
+  });
+}
+
+function buildExternalUpgradeStatements(
+  dialect: Exclude<RuntimeSchemaDialect, 'sqlite'>,
+  currentContract: SchemaContract,
+  liveContract: SchemaContract,
+): string[] {
+  return splitSqlStatements(generateUpgradeSql(dialect, currentContract, liveContract));
+}
+
+export async function ensureRuntimeDatabaseSchema(
+  client: RuntimeSchemaClient,
+  options: EnsureRuntimeDatabaseSchemaOptions = {},
+): Promise<void> {
+  const currentContract = options.currentContract ?? readSchemaContract();
   const statements = client.dialect === 'sqlite'
-    ? splitSqlStatements(generateBootstrapSql('sqlite', readSchemaContract()))
-    : readGeneratedBootstrapStatements(client.dialect);
+    ? splitSqlStatements(generateBootstrapSql('sqlite', currentContract))
+    : buildExternalUpgradeStatements(
+      client.dialect,
+      currentContract,
+      await resolveLiveContract(client, options.liveContract),
+    );
 
   for (const sqlText of statements) {
     await executeBootstrapStatement(client, sqlText);
@@ -260,5 +300,6 @@ export async function bootstrapRuntimeDatabaseSchema(input: RuntimeSchemaConnect
 }
 
 export const __runtimeSchemaBootstrapTestUtils = {
-  readGeneratedBootstrapStatements,
+  splitSqlStatements,
+  buildExternalUpgradeStatements,
 };