import { createHash } from 'node:crypto';
import { readdirSync, readFileSync } from 'node:fs';
import { dirname, join } from 'node:path';
import { fileURLToPath } from 'node:url';
import { captureCaught } from '@/server/observability/capture.server';

export type MigrationStatement = { file: string; statement: string };
export type MigrationFile = { file: string; sql: string; hash: string; createdAt: number };
export type MigrationSkipPredicate = (entry: MigrationStatement) => boolean;
export type MigrationExecutor = { query(sql: string, values?: unknown[]): Promise<unknown> };
export type MigrationPool = { connect(): Promise<MigrationExecutor & { release(): void }> };

type ConstraintExpectation = {
  type?: string;
  deleteAction?: string;
  validated?: boolean;
};

type MigrationCatalog = {
  tables: Set<string>;
  columns: Set<string>;
  enums: Map<string, string[]>;
  indexes: Set<string>;
  constraints: Map<string, ConstraintExpectation>;
  extensions: Set<string>;
  functions: Set<string>;
  triggers: Set<string>;
  absentIndexes: Set<string>;
  absentConstraints: Set<string>;
};

type CatalogInspection = { diff: string[]; positivePresent: number };

const MIGRATIONS_DIR = join(dirname(fileURLToPath(import.meta.url)), 'migrations');
const MIGRATION_FILE = /^(0000_baseline|\d{14}_[a-z0-9_]+)\.sql$/;

function migrationPrefix(file: string): number {
  const match = /^(\d+)/.exec(file);
  if (!match) throw new Error(`migration file has no numeric prefix: ${file}`);
  return Number(match[1]);
}

function compareMigrationFiles(a: string, b: string): number {
  const prefixDelta = migrationPrefix(a) - migrationPrefix(b);
  if (prefixDelta !== 0) return prefixDelta;
  return a.localeCompare(b);
}

function createdAt(file: string): number {
  if (file === '0000_baseline.sql') return 0;
  const stamp = file.split('_', 1)[0];
  if (!stamp || !/^\d{14}$/.test(stamp)) {
    throw new Error(`migration file has invalid timestamp: ${file}`);
  }
  const utc = Date.UTC(
    Number(stamp.slice(0, 4)),
    Number(stamp.slice(4, 6)) - 1,
    Number(stamp.slice(6, 8)),
    Number(stamp.slice(8, 10)),
    Number(stamp.slice(10, 12)),
    Number(stamp.slice(12, 14)),
  );
  if (!Number.isFinite(utc)) throw new Error(`migration file has invalid timestamp: ${file}`);
  return utc;
}

export function migrationDirectory(): string {
  return MIGRATIONS_DIR;
}

export function orderedMigrationFiles(directory = MIGRATIONS_DIR): MigrationFile[] {
  const entries = readdirSync(directory).filter((file) => file.endsWith('.sql'));
  const invalid = entries.filter((file) => !MIGRATION_FILE.test(file));
  if (invalid.length > 0) {
    throw new Error(
      `migration files not applied by authoritative chain: ${invalid.sort().join(', ')}`,
    );
  }
  const duplicates = entries
    .map(migrationPrefix)
    .filter((prefix, index, prefixes) => prefixes.indexOf(prefix) !== index);
  if (duplicates.length > 0)
    throw new Error(`duplicate migration prefixes: ${[...new Set(duplicates)].join(', ')}`);

  return entries.sort(compareMigrationFiles).map((file) => {
    const sql = readFileSync(join(directory, file), 'utf8');
    return {
      file,
      sql,
      hash: createHash('sha256').update(sql).digest('hex'),
      createdAt: createdAt(file),
    };
  });
}

function stripSqlComments(raw: string): string {
  let output = '';
  let dollarTag: string | null = null;
  let inSingleQuote = false;
  let inDoubleQuote = false;
  for (let i = 0; i < raw.length; i++) {
    const ch = raw[i];
    const next = raw[i + 1];
    if (dollarTag) {
      if (raw.startsWith(dollarTag, i)) {
        output += dollarTag;
        i += dollarTag.length - 1;
        dollarTag = null;
      } else output += ch;
      continue;
    }
    if (!inSingleQuote && !inDoubleQuote && ch === '$') {
      const match = raw.slice(i).match(/^\$[A-Za-z_][A-Za-z0-9_]*\$|^\$\$/);
      if (match) {
        dollarTag = match[0];
        output += dollarTag;
        i += dollarTag.length - 1;
        continue;
      }
    }
    if (!inDoubleQuote && ch === "'") {
      output += ch;
      if (inSingleQuote && next === "'") {
        output += next;
        i++;
      } else inSingleQuote = !inSingleQuote;
      continue;
    }
    if (!inSingleQuote && ch === '"') {
      inDoubleQuote = !inDoubleQuote;
      output += ch;
      continue;
    }
    if (!inSingleQuote && !inDoubleQuote && ch === '-' && next === '-') {
      while (i < raw.length && raw[i] !== '\n' && raw[i] !== '\r') i++;
      output += '\n';
      continue;
    }
    if (!inSingleQuote && !inDoubleQuote && ch === '/' && next === '*') {
      i += 2;
      while (i < raw.length && !(raw[i] === '*' && raw[i + 1] === '/')) i++;
      i++;
      output += ' ';
      continue;
    }
    output += ch;
  }
  return output;
}

export function splitMigrationStatements(raw: string): string[] {
  return raw.includes('--> statement-breakpoint')
    ? raw
        .split('--> statement-breakpoint')
        .map(stripSqlComments)
        .map((statement) => statement.trim())
        .filter(Boolean)
    : [stripSqlComments(raw).trim()].filter(Boolean);
}

function emptyCatalog(): MigrationCatalog {
  return {
    tables: new Set(),
    columns: new Set(),
    enums: new Map(),
    indexes: new Set(),
    constraints: new Map(),
    extensions: new Set(),
    functions: new Set(),
    triggers: new Set(),
    absentIndexes: new Set(),
    absentConstraints: new Set(),
  };
}

function identifier(raw: string): string {
  return raw.trim().replaceAll('"', '').split('.').at(-1) ?? raw;
}

function catalogKey(raw: string): string {
  const parts = raw.trim().replaceAll('"', '').split('.');
  return `${parts.length > 1 ? parts.at(-2) : 'public'}.${parts.at(-1)}`;
}

function parenthesized(raw: string, start: number): string {
  let depth = 0;
  let quote = false;
  for (let index = start; index < raw.length; index++) {
    const char = raw[index];
    if (char === "'") {
      if (quote && raw[index + 1] === "'") index++;
      else quote = !quote;
      continue;
    }
    if (quote) continue;
    if (char === '(') depth++;
    if (char === ')') {
      depth--;
      if (depth === 0) return raw.slice(start + 1, index);
    }
  }
  return '';
}

function topLevelParts(raw: string): string[] {
  const parts: string[] = [];
  let start = 0;
  let depth = 0;
  let quote = false;
  for (let index = 0; index < raw.length; index++) {
    const char = raw[index];
    if (char === "'") {
      if (quote && raw[index + 1] === "'") index++;
      else quote = !quote;
      continue;
    }
    if (quote) continue;
    if (char === '(') depth++;
    else if (char === ')') depth--;
    else if (char === ',' && depth === 0) {
      parts.push(raw.slice(start, index));
      start = index + 1;
    }
  }
  parts.push(raw.slice(start));
  return parts;
}

const SQL_IDENTIFIER = String.raw`(?:"[^"]+"|[A-Za-z_][A-Za-z0-9_$]*)(?:\.(?:"[^"]+"|[A-Za-z_][A-Za-z0-9_$]*))?`;

function addTableColumns(catalog: MigrationCatalog, table: string, body: string): void {
  catalog.tables.add(catalogKey(table));
  for (const part of topLevelParts(body)) {
    const column = part.trim();
    if (/^(CONSTRAINT|PRIMARY\s+KEY|FOREIGN\s+KEY|UNIQUE|CHECK|EXCLUDE)\b/i.test(column)) {
      continue;
    }
    const match = new RegExp(`^(${SQL_IDENTIFIER})\\s+`).exec(column);
    if (match?.[1]) catalog.columns.add(`${catalogKey(table)}.${identifier(match[1])}`);
  }
}

function constraintExpectation(definition: string): ConstraintExpectation {
  if (/\bFOREIGN\s+KEY\b/i.test(definition)) {
    const action =
      /\bON\s+DELETE\s+(CASCADE|SET\s+NULL|SET\s+DEFAULT|RESTRICT|NO\s+ACTION)\b/i.exec(
        definition,
      )?.[1];
    return {
      type: 'f',
      deleteAction:
        action?.replace(/\s+/g, ' ').toUpperCase() === 'CASCADE'
          ? 'c'
          : action?.replace(/\s+/g, ' ').toUpperCase() === 'SET NULL'
            ? 'n'
            : action?.replace(/\s+/g, ' ').toUpperCase() === 'SET DEFAULT'
              ? 'd'
              : action?.replace(/\s+/g, ' ').toUpperCase() === 'RESTRICT'
                ? 'r'
                : 'a',
      validated: !/\bNOT\s+VALID\b/i.test(definition),
    };
  }
  if (/\bPRIMARY\s+KEY\b/i.test(definition)) return { type: 'p' };
  if (/\bUNIQUE\b/i.test(definition)) return { type: 'u' };
  if (/\bCHECK\b/i.test(definition)) return { type: 'c' };
  return {};
}

function catalogForStatements(statements: string[]): MigrationCatalog {
  const catalog = emptyCatalog();
  for (const statement of statements) {
    for (const table of statement.matchAll(
      new RegExp(`CREATE\\s+TABLE(?:\\s+IF\\s+NOT\\s+EXISTS)?\\s+(${SQL_IDENTIFIER})\\s*\\(`, 'gi'),
    )) {
      if (!table[1] || table.index === undefined) continue;
      const open = statement.indexOf('(', table.index);
      addTableColumns(catalog, table[1], parenthesized(statement, open));
    }

    const enumMatch = new RegExp(
      `CREATE\\s+TYPE\\s+(${SQL_IDENTIFIER})\\s+AS\\s+ENUM\\s*\\(`,
      'i',
    ).exec(statement);
    if (enumMatch?.[1]) {
      const open = statement.indexOf('(', enumMatch.index);
      const labels = [...parenthesized(statement, open).matchAll(/'((?:''|[^'])*)'/g)].map(
        (match) => match[1]!.replaceAll("''", "'"),
      );
      catalog.enums.set(catalogKey(enumMatch[1]), labels);
    }

    for (const match of statement.matchAll(
      new RegExp(
        `ALTER\\s+TABLE(?:\\s+ONLY)?\\s+(${SQL_IDENTIFIER})\\s+ADD\\s+COLUMN(?:\\s+IF\\s+NOT\\s+EXISTS)?\\s+(${SQL_IDENTIFIER})`,
        'gi',
      ),
    )) {
      if (match[1] && match[2])
        catalog.columns.add(`${catalogKey(match[1])}.${identifier(match[2])}`);
    }

    for (const match of statement.matchAll(
      new RegExp(
        `CREATE\\s+(?:UNIQUE\\s+)?INDEX(?:\\s+CONCURRENTLY)?(?:\\s+IF\\s+NOT\\s+EXISTS)?\\s+(${SQL_IDENTIFIER})`,
        'gi',
      ),
    )) {
      if (match[1]) {
        const key = catalogKey(match[1]);
        catalog.indexes.add(key);
        catalog.absentIndexes.delete(key);
      }
    }
    for (const match of statement.matchAll(
      new RegExp(`DROP\\s+INDEX(?:\\s+IF\\s+EXISTS)?\\s+(${SQL_IDENTIFIER})`, 'gi'),
    )) {
      if (match[1]) {
        const key = catalogKey(match[1]);
        catalog.indexes.delete(key);
        catalog.absentIndexes.add(key);
      }
    }

    for (const match of statement.matchAll(
      new RegExp(
        `(?:DROP\\s+CONSTRAINT(?:\\s+IF\\s+EXISTS)?\\s+(${SQL_IDENTIFIER})|ADD\\s+CONSTRAINT\\s+(${SQL_IDENTIFIER})\\s+([^;]+))`,
        'gi',
      ),
    )) {
      if (match[1]) {
        const key = catalogKey(match[1]);
        catalog.constraints.delete(key);
        catalog.absentConstraints.add(key);
      } else if (match[2] && match[3]) {
        const key = catalogKey(match[2]);
        catalog.constraints.set(key, constraintExpectation(match[3]));
        catalog.absentConstraints.delete(key);
      }
    }
    for (const match of statement.matchAll(
      new RegExp(`VALIDATE\\s+CONSTRAINT\\s+(${SQL_IDENTIFIER})`, 'gi'),
    )) {
      if (match[1]) {
        const key = catalogKey(match[1]);
        const expected = catalog.constraints.get(key);
        if (expected) catalog.constraints.set(key, { ...expected, validated: true });
      }
    }
    for (const match of statement.matchAll(
      new RegExp(`CREATE\\s+EXTENSION(?:\\s+IF\\s+NOT\\s+EXISTS)?\\s+(${SQL_IDENTIFIER})`, 'gi'),
    )) {
      if (match[1]) catalog.extensions.add(identifier(match[1]));
    }
    for (const match of statement.matchAll(
      new RegExp(`CREATE\\s+(?:OR\\s+REPLACE\\s+)?FUNCTION\\s+(${SQL_IDENTIFIER})`, 'gi'),
    )) {
      if (match[1]) catalog.functions.add(catalogKey(match[1]));
    }
    for (const match of statement.matchAll(
      new RegExp(`CREATE\\s+TRIGGER\\s+(${SQL_IDENTIFIER})`, 'gi'),
    )) {
      if (match[1]) catalog.triggers.add(identifier(match[1]));
    }
  }
  return catalog;
}

function mergeCatalog(target: MigrationCatalog, source: MigrationCatalog): void {
  for (const table of source.tables) target.tables.add(table);
  for (const column of source.columns) target.columns.add(column);
  for (const [name, labels] of source.enums) target.enums.set(name, labels);
  for (const index of source.absentIndexes) {
    target.indexes.delete(index);
    target.absentIndexes.add(index);
  }
  for (const name of source.absentConstraints) {
    target.constraints.delete(name);
    target.absentConstraints.add(name);
  }
  for (const index of source.indexes) {
    target.indexes.add(index);
    target.absentIndexes.delete(index);
  }
  for (const [name, constraint] of source.constraints) {
    target.constraints.set(name, constraint);
    target.absentConstraints.delete(name);
  }
  for (const extension of source.extensions) target.extensions.add(extension);
  for (const fn of source.functions) target.functions.add(fn);
  for (const trigger of source.triggers) target.triggers.add(trigger);
}

function catalogEffectCount(catalog: MigrationCatalog): number {
  return (
    catalog.tables.size +
    catalog.columns.size +
    catalog.enums.size +
    catalog.indexes.size +
    catalog.constraints.size +
    catalog.extensions.size +
    catalog.functions.size +
    catalog.triggers.size +
    catalog.absentIndexes.size +
    catalog.absentConstraints.size
  );
}

function rows<T>(result: unknown): T[] {
  if (!result || typeof result !== 'object' || !('rows' in result)) return [];
  const value = (result as { rows?: unknown }).rows;
  return Array.isArray(value) ? (value as T[]) : [];
}

function expectedNames(keys: Iterable<string>): string[] {
  return [...keys].map((key) => key.split('.').at(-1) ?? key);
}

function orderedSubset(expected: string[], actual: string[]): boolean {
  let cursor = 0;
  return expected.every((label) => {
    const at = actual.indexOf(label, cursor);
    if (at === -1) return false;
    cursor = at + 1;
    return true;
  });
}

async function inspectCatalog(
  client: MigrationExecutor,
  catalog: MigrationCatalog,
): Promise<CatalogInspection> {
  const tables = expectedNames(catalog.tables);
  const tableRows = tables.length
    ? rows<{ table_name: string }>(
        await client.query(
          `SELECT table_name FROM information_schema.tables WHERE table_schema = 'public' AND table_name = ANY($1::text[])`,
          [tables],
        ),
      )
    : [];
  const presentTables = new Set(tableRows.map((row) => `public.${row.table_name}`));
  const columnTables = [...new Set([...catalog.columns].map((column) => column.split('.')[1]))];
  const columnRows = columnTables.length
    ? rows<{ table_name: string; column_name: string }>(
        await client.query(
          `SELECT table_name, column_name FROM information_schema.columns WHERE table_schema = 'public' AND table_name = ANY($1::text[])`,
          [columnTables],
        ),
      )
    : [];
  const presentColumns = new Set(
    columnRows.map((row) => `public.${row.table_name}.${row.column_name}`),
  );
  const enumNames = expectedNames(catalog.enums.keys());
  const enumRows = enumNames.length
    ? rows<{ typname: string; enumlabel: string }>(
        await client.query(
          `SELECT t.typname, e.enumlabel FROM pg_type t JOIN pg_namespace n ON n.oid = t.typnamespace JOIN pg_enum e ON e.enumtypid = t.oid WHERE n.nspname = 'public' AND t.typname = ANY($1::text[]) ORDER BY t.typname, e.enumsortorder`,
          [enumNames],
        ),
      )
    : [];
  const presentEnums = new Map<string, string[]>();
  for (const row of enumRows) {
    const key = `public.${row.typname}`;
    presentEnums.set(key, [...(presentEnums.get(key) ?? []), row.enumlabel]);
  }
  const indexNames = expectedNames([...catalog.indexes, ...catalog.absentIndexes]);
  const indexRows = indexNames.length
    ? rows<{ indexname: string }>(
        await client.query(
          `SELECT indexname FROM pg_indexes WHERE schemaname = 'public' AND indexname = ANY($1::text[])`,
          [indexNames],
        ),
      )
    : [];
  const presentIndexes = new Set(indexRows.map((row) => `public.${row.indexname}`));
  const constraintNames = expectedNames([
    ...catalog.constraints.keys(),
    ...catalog.absentConstraints,
  ]);
  const constraintRows = constraintNames.length
    ? rows<{ conname: string; contype: string; confdeltype: string; convalidated: boolean }>(
        await client.query(
          `SELECT conname, contype, confdeltype, convalidated FROM pg_constraint WHERE connamespace = 'public'::regnamespace AND conname = ANY($1::text[])`,
          [constraintNames],
        ),
      )
    : [];
  const presentConstraints = new Map(constraintRows.map((row) => [`public.${row.conname}`, row]));
  const extensionNames = [...catalog.extensions];
  const extensionRows = extensionNames.length
    ? rows<{ extname: string }>(
        await client.query(`SELECT extname FROM pg_extension WHERE extname = ANY($1::text[])`, [
          extensionNames,
        ]),
      )
    : [];
  const presentExtensions = new Set(extensionRows.map((row) => row.extname));
  const functionNames = expectedNames(catalog.functions);
  const functionRows = functionNames.length
    ? rows<{ proname: string }>(
        await client.query(
          `SELECT proname FROM pg_proc WHERE pronamespace = 'public'::regnamespace AND proname = ANY($1::text[])`,
          [functionNames],
        ),
      )
    : [];
  const presentFunctions = new Set(functionRows.map((row) => `public.${row.proname}`));
  const triggerNames = [...catalog.triggers];
  const triggerRows = triggerNames.length
    ? rows<{ tgname: string }>(
        await client.query(
          `SELECT tgname FROM pg_trigger WHERE NOT tgisinternal AND tgname = ANY($1::text[])`,
          [triggerNames],
        ),
      )
    : [];
  const presentTriggers = new Set(triggerRows.map((row) => row.tgname));

  const diff: string[] = [];
  let positivePresent = 0;
  for (const table of catalog.tables) {
    if (presentTables.has(table)) positivePresent++;
    else diff.push(`table ${table}`);
  }
  for (const column of catalog.columns) {
    if (presentColumns.has(column)) positivePresent++;
    else diff.push(`column ${column}`);
  }
  for (const [name, expected] of catalog.enums) {
    const actual = presentEnums.get(name) ?? [];
    if (actual.length > 0) positivePresent++;
    if (!orderedSubset(expected, actual)) {
      diff.push(
        `enum ${name} expected ordered labels ${expected.join(', ')} got ${actual.join(', ')}`,
      );
    }
  }
  for (const index of catalog.indexes) {
    if (presentIndexes.has(index)) positivePresent++;
    else diff.push(`index ${index}`);
  }
  for (const [name, expected] of catalog.constraints) {
    const actual = presentConstraints.get(name);
    if (actual) positivePresent++;
    if (!actual) {
      diff.push(`constraint ${name}`);
      continue;
    }
    if (expected.type && actual.contype !== expected.type) {
      diff.push(`constraint ${name} expected type ${expected.type} got ${actual.contype}`);
    }
    if (expected.deleteAction && actual.confdeltype !== expected.deleteAction) {
      diff.push(
        `constraint ${name} expected delete action ${expected.deleteAction} got ${actual.confdeltype}`,
      );
    }
    if (expected.validated !== undefined && actual.convalidated !== expected.validated) {
      diff.push(
        `constraint ${name} expected validated=${expected.validated} got ${actual.convalidated}`,
      );
    }
  }
  for (const extension of catalog.extensions) {
    if (presentExtensions.has(extension)) positivePresent++;
    else diff.push(`extension ${extension}`);
  }
  for (const fn of catalog.functions) {
    if (presentFunctions.has(fn)) positivePresent++;
    else diff.push(`function ${fn}`);
  }
  for (const trigger of catalog.triggers) {
    if (presentTriggers.has(trigger)) positivePresent++;
    else diff.push(`trigger ${trigger}`);
  }
  for (const index of catalog.absentIndexes) {
    if (presentIndexes.has(index)) diff.push(`index ${index} must be absent`);
  }
  for (const constraint of catalog.absentConstraints) {
    if (presentConstraints.has(constraint)) diff.push(`constraint ${constraint} must be absent`);
  }
  return { diff, positivePresent };
}

export type ApplyMigrationChainOptions = { directory?: string; skip?: MigrationSkipPredicate };

export type MigrationChainResult = {
  applied: string[];
  reconciled: string[];
  skipped: string[];
  legacyLedgerHashes: string[];
};

export async function applyMigrationChain(
  pool: MigrationPool,
  options: ApplyMigrationChainOptions = {},
): Promise<MigrationChainResult> {
  const migrations = orderedMigrationFiles(options.directory);
  const client = await pool.connect();
  try {
    await client.query('BEGIN');
    await client.query(
      "SELECT pg_advisory_xact_lock(hashtextextended('multideal:migration-chain', 0))",
    );
    await client.query('CREATE SCHEMA IF NOT EXISTS drizzle');
    await client.query(
      'CREATE TABLE IF NOT EXISTS drizzle.__drizzle_migrations (id serial PRIMARY KEY, hash text NOT NULL, created_at bigint)',
    );
    const hashes = migrations.map((migration) => migration.hash);
    const result = (await client.query('SELECT hash FROM drizzle.__drizzle_migrations')) as {
      rows?: Array<{ hash: string }>;
    };
    const ledgerHashes = new Set(result.rows?.map((row) => row.hash) ?? []);
    const appliedHashes = new Set(hashes.filter((hash) => ledgerHashes.has(hash)));
    const legacyLedgerHashes = [...ledgerHashes].filter((hash) => !hashes.includes(hash));
    const applied: string[] = [];
    const reconciled: string[] = [];
    const skipped: string[] = [];
    const pending = migrations.filter((migration) => !appliedHashes.has(migration.hash));
    const catalog = emptyCatalog();
    const migrationCatalogs = new Map<string, MigrationCatalog>();
    for (const migration of migrations) {
      const statements = splitMigrationStatements(migration.sql).filter(
        (statement) => !options.skip?.({ file: migration.file, statement }),
      );
      const migrationCatalog = catalogForStatements(statements);
      migrationCatalogs.set(migration.file, migrationCatalog);
      mergeCatalog(catalog, migrationCatalog);
    }
    const inspection = await inspectCatalog(client, catalog);
    if (pending.length > 0 && appliedHashes.size > 0) {
      const appliedCatalog = emptyCatalog();
      for (const migration of migrations) {
        if (!appliedHashes.has(migration.hash)) continue;
        const statements = splitMigrationStatements(migration.sql).filter(
          (statement) => !options.skip?.({ file: migration.file, statement }),
        );
        mergeCatalog(appliedCatalog, catalogForStatements(statements));
      }
      const appliedInspection = await inspectCatalog(client, appliedCatalog);
      if (appliedInspection.diff.length > 0) {
        throw new Error(
          `migration reconciliation refused; catalog drifts from recorded migrations:\n${appliedInspection.diff.map((entry) => `- ${entry}`).join('\n')}`,
        );
      }
    }
    if (pending.length > 0 && inspection.positivePresent > 0 && inspection.diff.length === 0) {
      for (const migration of pending) {
        await client.query(
          'INSERT INTO drizzle.__drizzle_migrations (hash, created_at) VALUES ($1, $2)',
          [migration.hash, migration.createdAt],
        );
        reconciled.push(migration.file);
      }
      await client.query('COMMIT');
      return { applied, reconciled, skipped, legacyLedgerHashes };
    }
    if (pending.length > 0 && inspection.positivePresent > 0 && ledgerHashes.size === 0) {
      throw new Error(
        `migration reconciliation refused; live catalog differs from migration chain:\n${inspection.diff.map((entry) => `- ${entry}`).join('\n')}`,
      );
    }
    let prefixLength = 0;
    if (inspection.positivePresent > 0) {
      let pendingGap = false;
      for (const migration of pending) {
        const migrationCatalog = migrationCatalogs.get(migration.file);
        if (!migrationCatalog || catalogEffectCount(migrationCatalog) === 0) {
          throw new Error(
            `migration reconciliation refused; migration ${migration.file} has no catalog-verifiable effect`,
          );
        }
        const migrationInspection = await inspectCatalog(client, migrationCatalog);
        const present = migrationInspection.diff.length === 0;
        if (present && pendingGap) {
          throw new Error(
            `migration reconciliation refused; live catalog differs from migration chain:\n${inspection.diff.map((entry) => `- ${entry}`).join('\n')}`,
          );
        }
        if (present) prefixLength++;
        else pendingGap = true;
      }
    }
    const reconciledPending = new Set(
      pending.slice(0, prefixLength).map((migration) => migration.hash),
    );
    for (const migration of pending.slice(0, prefixLength)) {
      await client.query(
        'INSERT INTO drizzle.__drizzle_migrations (hash, created_at) VALUES ($1, $2)',
        [migration.hash, migration.createdAt],
      );
      reconciled.push(migration.file);
    }
    try {
      for (const migration of migrations) {
        if (appliedHashes.has(migration.hash)) {
          skipped.push(migration.file);
          continue;
        }
        if (reconciledPending.has(migration.hash)) continue;
        const statements = splitMigrationStatements(migration.sql).filter(
          (statement) => !options.skip?.({ file: migration.file, statement }),
        );
        for (const statement of statements) await client.query(statement);
        await client.query(
          'INSERT INTO drizzle.__drizzle_migrations (hash, created_at) VALUES ($1, $2)',
          [migration.hash, migration.createdAt],
        );
        applied.push(migration.file);
      }
    } catch (error) {
      await client.query('ROLLBACK');
      const conflictInspection = await inspectCatalog(client, catalog);
      throw new Error(
        `migration reconciliation refused; live catalog differs from migration chain:\n${conflictInspection.diff.map((entry) => `- ${entry}`).join('\n')}\nsuffix execution failed: ${error instanceof Error ? error.message : String(error)}`,
        { cause: error },
      );
    }
    await client.query('COMMIT');
    return { applied, reconciled, skipped, legacyLedgerHashes };
  } catch (error) {
    await client.query('ROLLBACK').catch((rollbackError) => {
      captureCaught(rollbackError, { scope: 'server.db.migration-chain.rollback' });
    });
    throw error;
  } finally {
    client.release();
  }
}