import { createHash } from "node:crypto"; import { readdir, readFile } from "node:fs/promises"; import path from "node:path"; import process from "node:process"; import { createInterface } from "node:readline"; import { fileURLToPath } from "node:url"; import pg from "pg"; const { Client } = pg; const EXPECTED_DATABASE = "booking_test"; const TARGET_SCHEMA = "condon"; const scriptDirectory = path.dirname(fileURLToPath(import.meta.url)); const migrationsDirectory = path.resolve(scriptDirectory, "..", "migrations"); async function readStandardInput() { const lines = createInterface({ input: process.stdin, terminal: false }); for await (const line of lines) { lines.close(); return line.trim(); } throw Object.assign(new Error("Missing connection input"), { code: "MISSING_CONNECTION_INPUT" }); } function sanitizeError(error) { const errorCode = typeof error?.code === "string" ? error.code : "MIGRATION_FAILED"; const connectionErrorCodes = new Set([ "ECONNREFUSED", "ECONNRESET", "ENOTFOUND", "ETIMEDOUT" ]); return { ok: false, errorCode, errorMessage: connectionErrorCodes.has(errorCode) ? "Database connection failed" : typeof error?.message === "string" ? error.message.replace(/\s+/g, " ").slice(0, 240) : "Migration failed" }; } async function loadMigrations() { const names = (await readdir(migrationsDirectory)) .filter(name => name.endsWith(".up.sql")) .sort(); if (names.length === 0) { throw Object.assign(new Error("No up migrations found"), { code: "NO_MIGRATIONS_FOUND" }); } return Promise.all(names.map(async fileName => { const sql = await readFile(path.join(migrationsDirectory, fileName), "utf8"); return { fileName, version: fileName.slice(0, -".up.sql".length), sql, checksum: createHash("sha256").update(sql).digest("hex") }; })); } let client; let transactionStarted = false; try { const [rawInput, migrations] = await Promise.all([ readStandardInput(), loadMigrations() ]); const input = JSON.parse(rawInput); client = new Client({ host: input.host, port: input.port, user: input.user, password: input.password, database: input.database, ssl: input.ssl ? { rejectUnauthorized: false } : undefined, application_name: "condon_guarded_migration", connectionTimeoutMillis: 10_000, query_timeout: 60_000, options: "-c statement_timeout=60000 -c lock_timeout=2000" }); await client.connect(); await client.query("BEGIN"); transactionStarted = true; const identityResult = await client.query(` SELECT current_database() AS database_name, current_user AS role_name, current_setting('transaction_read_only') AS transaction_read_only `); const identity = identityResult.rows[0]; if (identity.database_name !== EXPECTED_DATABASE) { throw Object.assign(new Error("Unexpected database"), { code: "UNEXPECTED_DATABASE" }); } if (identity.role_name !== input.user) { throw Object.assign(new Error("Unexpected database role"), { code: "UNEXPECTED_DATABASE_ROLE" }); } if (identity.transaction_read_only !== "off") { throw Object.assign(new Error("Migration transaction is read-only"), { code: "MIGRATION_TRANSACTION_READ_ONLY" }); } const schemaResult = await client.query(` SELECT EXISTS ( SELECT 1 FROM pg_catalog.pg_namespace WHERE nspname = $1 ) AS exists `, [TARGET_SCHEMA]); const schemaExists = schemaResult.rows[0].exists === true; if (schemaExists) { const migrationTableResult = await client.query(` SELECT to_regclass('condon.schema_migrations') IS NOT NULL AS exists `); if (!migrationTableResult.rows[0].exists) { throw Object.assign(new Error("Migration table is missing"), { code: "MIGRATION_TABLE_MISSING" }); } } const applied = []; const skipped = []; for (const migration of migrations) { let existing = null; if (schemaExists || migration !== migrations[0]) { const result = await client.query( "SELECT checksum FROM condon.schema_migrations WHERE version = $1", [migration.version] ); existing = result.rows[0] ?? null; } if (existing) { if (existing.checksum !== migration.checksum) { throw Object.assign(new Error(`Checksum mismatch for ${migration.version}`), { code: "MIGRATION_CHECKSUM_MISMATCH" }); } skipped.push(migration.version); continue; } await client.query(migration.sql); await client.query( `INSERT INTO condon.schema_migrations (version, checksum) VALUES ($1, $2)`, [migration.version, migration.checksum] ); applied.push(migration.version); } const verificationResult = await client.query(` SELECT (SELECT count(*)::integer FROM condon.room_types) AS room_type_count, (SELECT count(*)::integer FROM condon.owner_accounts) AS owner_account_count, (SELECT count(*)::integer FROM condon.entitlement_periods) AS entitlement_period_count, (SELECT count(*)::integer FROM condon.usage_records) AS usage_record_count, (SELECT count(*)::integer FROM condon.entitlement_ledger) AS ledger_count, (SELECT count(*)::integer FROM condon.schema_migrations) AS migration_count, (SELECT count(*)::integer FROM condon.bookings) AS booking_count `); const counts = verificationResult.rows[0]; await client.query("COMMIT"); transactionStarted = false; process.stdout.write(`${JSON.stringify({ ok: true, databaseVerified: true, targetSchema: TARGET_SCHEMA, applied, skipped, migrations: migrations.map(migration => ({ version: migration.version, checksum: migration.checksum })), counts }, null, 2)}\n`); } catch (error) { if (client && transactionStarted) { try { await client.query("ROLLBACK"); transactionStarted = false; } catch { // Keep the sanitized migration failure authoritative. } } process.stdout.write(`${JSON.stringify(sanitizeError(error), null, 2)}\n`); process.exitCode = 1; } finally { if (client) { await client.end().catch(() => { process.exitCode = 1; }); } }