import { createHash } from 'node:crypto';
import type { Pool, PoolClient } from 'pg';
import { databaseMigrations, type DatabaseMigration } from './migration-catalog';

const MIGRATION_LOCK_ID = 817_726_471;

export interface AppliedMigration {
  version: string;
  name: string;
  checksum: string;
}

function checksum(migration: DatabaseMigration): string {
  return createHash('sha256').update(migration.sql, 'utf8').digest('hex');
}

async function ensureMigrationTable(client: PoolClient): Promise<void> {
  await client.query(`
    CREATE TABLE IF NOT EXISTS assessor_schema_migrations (
      version text PRIMARY KEY,
      name text NOT NULL,
      checksum text NOT NULL,
      applied_at timestamptz NOT NULL DEFAULT now()
    )
  `);
}

export async function runDatabaseMigrations(
  pool: Pool,
  migrations: readonly DatabaseMigration[] = databaseMigrations
): Promise<AppliedMigration[]> {
  const client = await pool.connect();
  const applied: AppliedMigration[] = [];
  try {
    await client.query('SELECT pg_advisory_lock($1)', [MIGRATION_LOCK_ID]);
    await ensureMigrationTable(client);
    const existing = await client.query<{ version: string; checksum: string }>(
      'SELECT version, checksum FROM assessor_schema_migrations ORDER BY version'
    );
    const existingByVersion = new Map(existing.rows.map(row => [row.version, row.checksum]));
    for (const migration of migrations) {
      const migrationChecksum = checksum(migration);
      const storedChecksum = existingByVersion.get(migration.version);
      if (storedChecksum && storedChecksum !== migrationChecksum) {
        throw new Error(`Migration ${migration.version} wurde nach ihrer Anwendung verändert.`);
      }
      if (storedChecksum) continue;
      await client.query('BEGIN');
      try {
        await client.query(migration.sql);
        await client.query(
          'INSERT INTO assessor_schema_migrations (version, name, checksum) VALUES ($1, $2, $3)',
          [migration.version, migration.name, migrationChecksum]
        );
        await client.query('COMMIT');
        applied.push({ version: migration.version, name: migration.name, checksum: migrationChecksum });
      } catch (error) {
        await client.query('ROLLBACK');
        throw error;
      }
    }
    return applied;
  } finally {
    try {
      await client.query('SELECT pg_advisory_unlock($1)', [MIGRATION_LOCK_ID]);
    } finally {
      client.release();
    }
  }
}