import { readFile } from 'node:fs/promises'; import path from 'node:path'; import { pathToFileURL } from 'node:url'; import pg from 'pg'; import { ulid } from 'ulid'; const ULID_RE = /^[0-9A-HJKMNP-TV-Z]{26}$/; const EXPECTED_TABLES = [ 'staff_audit_log', 'staff_attendance_events', 'staff_daily_metrics', 'staff_employees', 'staff_idempotency_keys', 'staff_inbox', 'staff_metric_definitions', 'staff_notification_deliveries', 'staff_org_unit_terms', 'staff_org_units', 'staff_outbox', 'staff_pay_assignments', 'staff_pay_rules', 'staff_payroll_ledger', 'staff_payroll_lines', 'staff_payroll_runs', 'staff_professions', 'staff_push_subscriptions', 'staff_standalone_employee_access', 'staff_sync_change_log', 'staff_sync_command_inbox', 'staff_sync_device_checkpoints', 'staff_sync_devices', 'staff_timesheet_entries', 'staff_work_items', 'staff_work_sessions', 'staff_payouts', ]; class SafeOperationError extends Error {} const identifier = (value) => `"${String(value).replaceAll('"', '""')}"`; const qualified = (schema, table) => `${identifier(schema)}.${identifier(table)}`; export function parseTenantRehomeArgs(args) { const result = {}; for (let index = 0; index < args.length; index += 1) { const key = args[index]; if (key === '--dry-run' || key === '--apply') { if (result.mode) throw new SafeOperationError('Choose exactly one of --dry-run or --apply.'); result.mode = key.slice(2); } else if (key === '--json') { if (result.json) throw new SafeOperationError('Duplicate option --json.'); result.json = true; } else if (['--source', '--from', '--target', '--to', '--actor', '--backup-evidence'].includes(key)) { const value = args[index + 1]; if (!value || value.startsWith('--')) throw new SafeOperationError(`Missing value for ${key}.`); const name = ({ '--source': 'source', '--from': 'source', '--target': 'target', '--to': 'target', '--actor': 'actor', '--backup-evidence': 'backup_evidence', })[key]; if (result[name]) throw new SafeOperationError(`Duplicate option for ${name.replaceAll('_', '-')}.`); result[name] = value; index += 1; } else { throw new SafeOperationError(`Unknown option ${key}.`); } } result.mode ??= 'dry-run'; for (const name of ['source', 'target']) { if (!ULID_RE.test(result[name] ?? '')) throw new SafeOperationError(`--${name} must be a valid ULID.`); } if (result.source === result.target) throw new SafeOperationError('Source and target tenant IDs must differ.'); if (result.mode === 'apply') { if (!ULID_RE.test(result.actor ?? '')) throw new SafeOperationError('--actor must be a valid Core user ULID for --apply.'); if (!result.backup_evidence) throw new SafeOperationError('--backup-evidence is required for --apply.'); } else if (result.actor || result.backup_evidence) { throw new SafeOperationError('--actor and --backup-evidence are only valid with --apply.'); } return result; } export function validateBackupEvidenceObject(evidence) { const backup = evidence?.backup; const restore = evidence?.restoreDrill; const sha256 = /^[a-f0-9]{64}$/i; const isoDate = (value) => typeof value === 'string' && Number.isFinite(Date.parse(value)); if ( backup?.format !== 'pg_dump -Fc' || backup?.encrypted !== true || backup?.offHost !== true || !sha256.test(backup?.sha256 ?? '') || !isoDate(backup?.createdAt) || restore?.status !== 'passed' || !isoDate(restore?.restoredAt) || !sha256.test(restore?.backupSha256 ?? '') || backup.sha256.toLowerCase() !== restore.backupSha256.toLowerCase() || typeof restore?.isolatedDatabase !== 'string' || restore.isolatedDatabase.trim() === '' || !isoDate(restore?.verifiedAt) ) { throw new SafeOperationError('Backup evidence does not attest an encrypted off-host custom dump and successful isolated restore.'); } return { backup: { format: 'pg_dump -Fc', encrypted: true, offHost: true, sha256: backup.sha256.toLowerCase(), createdAt: backup.createdAt, }, restoreDrill: { status: 'passed', backupSha256: restore.backupSha256.toLowerCase(), restoredAt: restore.restoredAt, verifiedAt: restore.verifiedAt, isolatedDatabase: restore.isolatedDatabase, }, }; } export async function validateBackupEvidence(filePath) { let evidence; try { evidence = JSON.parse(await readFile(path.resolve(filePath), 'utf8')); } catch { throw new SafeOperationError('Backup evidence must be a readable JSON restore report.'); } return validateBackupEvidenceObject(evidence); } async function discoverTenantTables(client) { const result = await client.query(` SELECT c.relname AS table_name, c.relrowsecurity AS rls_enabled, c.relforcerowsecurity AS rls_forced, pg_get_userbyid(c.relowner) AS table_owner, EXISTS ( SELECT 1 FROM pg_attribute a WHERE a.attrelid = c.oid AND a.attname = 'tenant_id' AND a.attnum > 0 AND NOT a.attisdropped ) AS has_tenant_id FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace WHERE n.nspname = 'public' AND c.relkind IN ('r', 'p') AND left(c.relname, 6) = 'staff_' ORDER BY c.relname `); const tables = result.rows; if (tables.length === 0) throw new SafeOperationError('No public staff tables found.'); const names = new Set(tables.map((table) => table.table_name)); const missing = EXPECTED_TABLES.filter((table) => !names.has(table)); if (missing.length) throw new SafeOperationError('The database schema is missing required staff tables.'); if (tables.some((table) => !table.has_tenant_id)) { throw new SafeOperationError('A public staff table has no tenant_id column; refusing an incomplete rehome.'); } if (tables.some((table) => !table.rls_enabled || !table.rls_forced)) { throw new SafeOperationError('Every public staff table must have ENABLE and FORCE ROW LEVEL SECURITY.'); } return tables; } async function assertAdministrator(client, tables) { const result = await client.query(` SELECT current_user AS role_name, r.rolsuper AS is_superuser, r.rolbypassrls AS bypasses_rls FROM pg_roles r WHERE r.rolname = current_user `); const role = result.rows[0]; if (!role || role.role_name === 'krona_staff_app' || (!role.is_superuser && !role.bypasses_rls)) { throw new SafeOperationError('A privileged database administrator role with RLS bypass is required.'); } if (!role.is_superuser && tables.some((table) => table.table_owner !== role.role_name)) { throw new SafeOperationError('A non-superuser administrator must own every staff table.'); } } async function assertSafeTriggersAndForeignKeys(client, tables) { const names = tables.map((table) => table.table_name); const triggers = await client.query( `SELECT c.relname AS table_name, t.tgname AS trigger_name FROM pg_trigger t JOIN pg_class c ON c.oid = t.tgrelid JOIN pg_namespace n ON n.oid = c.relnamespace WHERE n.nspname = 'public' AND c.relname = ANY($1::text[]) AND NOT t.tgisinternal ORDER BY c.relname, t.tgname`, [names], ); if (triggers.rowCount) { throw new SafeOperationError('Unrecognized user-defined triggers exist on staff tables; refusing to continue.'); } const fks = await client.query(` SELECT con.conname AS constraint_name, src_ns.nspname AS source_schema, src.relname AS source_table, dst_ns.nspname AS target_schema, dst.relname AS target_table, pg_get_constraintdef(con.oid, true) AS definition, con.convalidated AS is_validated FROM pg_constraint con JOIN pg_class src ON src.oid = con.conrelid JOIN pg_namespace src_ns ON src_ns.oid = src.relnamespace JOIN pg_class dst ON dst.oid = con.confrelid JOIN pg_namespace dst_ns ON dst_ns.oid = dst.relnamespace WHERE con.contype = 'f' AND (src_ns.nspname = 'public' AND src.relname = ANY($1::text[]) OR dst_ns.nspname = 'public' AND dst.relname = ANY($1::text[])) ORDER BY src_ns.nspname, src.relname, con.conname `, [names]); if (fks.rows.some((fk) => fk.source_schema !== 'public' || fk.target_schema !== 'public' || !names.includes(fk.source_table) || !names.includes(fk.target_table) )) { throw new SafeOperationError('A foreign key crosses the staff tenant-table boundary; refusing an incomplete rehome.'); } return fks.rows; } async function tableCounts(client, tables, tenantId) { const counts = {}; for (const { table_name: table } of tables) { const { rows } = await client.query( `SELECT count(*)::text AS row_count FROM ${qualified('public', table)} WHERE tenant_id = $1`, [tenantId], ); counts[table] = rows[0].row_count; } return counts; } async function ledgerTotals(client, tenantId) { const result = await client.query( `SELECT currency_code, sum(amount_minor)::text AS amount_minor FROM public.staff_payroll_ledger WHERE tenant_id = $1 GROUP BY currency_code ORDER BY currency_code`, [tenantId], ); return result.rows.map(({ currency_code, amount_minor }) => ({ currency: currency_code, amountMinor: amount_minor })); } async function lockTables(client, tables) { for (const { table_name: table } of tables) { await client.query(`LOCK TABLE ${qualified('public', table)} IN ACCESS EXCLUSIVE MODE`); } } export async function rehomeTenant(client, options) { const { source, target, mode } = options; if (!ULID_RE.test(source ?? '') || !ULID_RE.test(target ?? '') || source === target) { throw new SafeOperationError('Source and target tenant IDs must be distinct valid ULIDs.'); } if (mode !== 'dry-run' && mode !== 'apply') throw new SafeOperationError('Mode must be dry-run or apply.'); if (mode === 'apply') { if (!ULID_RE.test(options.actor ?? '')) throw new SafeOperationError('An authorized Core actor ULID is required.'); if (!options.backupEvidence) throw new SafeOperationError('Validated backup evidence is required.'); options.backupEvidence = validateBackupEvidenceObject(options.backupEvidence); } await client.query(mode === 'dry-run' ? 'BEGIN ISOLATION LEVEL REPEATABLE READ READ ONLY' : 'BEGIN ISOLATION LEVEL SERIALIZABLE'); let committed = false; try { if (mode === 'apply') { await client.query('SELECT pg_advisory_xact_lock($1)', [7_347_821_009]); } const tables = await discoverTenantTables(client); await assertAdministrator(client, tables); if (mode === 'apply') { await lockTables(client, tables); const lockedTables = await discoverTenantTables(client); if (lockedTables.map((table) => table.table_name).join('\n') !== tables.map((table) => table.table_name).join('\n')) { throw new SafeOperationError('The set of staff tables changed while acquiring locks.'); } } const sourceCounts = await tableCounts(client, tables, source); const targetCounts = await tableCounts(client, tables, target); const sourceTotal = Object.values(sourceCounts).reduce((total, count) => total + BigInt(count), 0n); const targetTotal = Object.values(targetCounts).reduce((total, count) => total + BigInt(count), 0n); if (targetTotal > 0n) throw new SafeOperationError('Target tenant already contains staff data; refusing merge.'); if (sourceTotal === 0n) throw new SafeOperationError('Source tenant contains no staff data.'); const report = { mode, source, target, tables: sourceCounts }; if (mode === 'dry-run') { await client.query('COMMIT'); committed = true; return report; } const foreignKeys = await assertSafeTriggersAndForeignKeys(client, tables); const ledgerBefore = await ledgerTotals(client, source); for (const fk of foreignKeys) { await client.query( `ALTER TABLE ${qualified(fk.source_schema, fk.source_table)} DROP CONSTRAINT ${identifier(fk.constraint_name)}`, ); } for (const { table_name: table } of tables) { await client.query( `UPDATE ${qualified('public', table)} SET tenant_id = $1 WHERE tenant_id = $2`, [target, source], ); } for (const fk of foreignKeys) { let definition = fk.definition; if (!fk.is_validated && !/\bNOT VALID\s*$/i.test(definition)) definition += ' NOT VALID'; await client.query( `ALTER TABLE ${qualified(fk.source_schema, fk.source_table)} ADD CONSTRAINT ${identifier(fk.constraint_name)} ${definition}`, ); } const restoredForeignKeys = await assertSafeTriggersAndForeignKeys(client, tables); const foreignKeySignature = (fk) => [ fk.source_schema, fk.source_table, fk.constraint_name, fk.target_schema, fk.target_table, fk.definition, fk.is_validated, ].join('\u0000'); if ( foreignKeys.map(foreignKeySignature).join('\n') !== restoredForeignKeys.map(foreignKeySignature).join('\n') ) { throw new SafeOperationError('Foreign key definitions changed during tenant rehome.'); } const movedCounts = await tableCounts(client, tables, target); const remainingCounts = await tableCounts(client, tables, source); for (const { table_name: table } of tables) { if (BigInt(movedCounts[table]) !== BigInt(sourceCounts[table]) || BigInt(remainingCounts[table]) !== 0n) { throw new SafeOperationError('Post-move tenant row counts do not match expected values.'); } } const ledgerAfter = await ledgerTotals(client, target); if (JSON.stringify(ledgerAfter) !== JSON.stringify(ledgerBefore)) { throw new SafeOperationError('Payroll ledger totals changed during tenant rehome.'); } const actor = options.actor; const correlationId = options.correlationId ?? ulid(); if (!ULID_RE.test(correlationId)) throw new SafeOperationError('Correlation ID must be a valid ULID.'); await client.query( `INSERT INTO public.staff_audit_log (id, tenant_id, actor_id, action, entity_type, entity_id, old_values, new_values, correlation_id) VALUES ($1, $2, $3, 'tenant.rehomed', 'tenant_rehome', NULL, $4::jsonb, $5::jsonb, $6)`, [ulid(), target, actor, JSON.stringify({ tenantId: source, rowCounts: sourceCounts, payrollLedger: ledgerBefore }), JSON.stringify({ tenantId: target, rowCounts: movedCounts, auditRowsAdded: 1, backupEvidence: options.backupEvidence, }), correlationId], ); const actualAuditCount = BigInt((await tableCounts(client, [{ table_name: 'staff_audit_log' }], target)).staff_audit_log); if (actualAuditCount !== BigInt(movedCounts.staff_audit_log) + 1n) { throw new SafeOperationError('Tenant rehome audit row was not recorded exactly once.'); } movedCounts.staff_audit_log = actualAuditCount.toString(); report.tables = movedCounts; report.auditRowsAdded = 1; report.payrollLedger = ledgerAfter; await client.query('COMMIT'); committed = true; return report; } catch (error) { if (!committed) await client.query('ROLLBACK').catch(() => undefined); throw error; } } function parseCli(args) { const options = parseTenantRehomeArgs(args); const connectionString = process.env.DATABASE_ADMIN_URL ?? process.env.DATABASE_REHOME_URL ?? process.env.DATABASE_MIGRATION_URL; if (!connectionString) throw new SafeOperationError('Set DATABASE_ADMIN_URL, DATABASE_REHOME_URL, or DATABASE_MIGRATION_URL.'); return { ...options, connectionString }; } async function main() { let options; let client; try { options = parseCli(process.argv.slice(2)); if (options.mode === 'apply') options.backupEvidence = await validateBackupEvidence(options.backup_evidence); client = new pg.Client({ connectionString: options.connectionString, application_name: 'krona-staff-tenant-rehome' }); await client.connect(); const report = await rehomeTenant(client, { source: options.source, target: options.target, actor: options.actor, mode: options.mode, backupEvidence: options.backupEvidence, }); process.stdout.write(`${JSON.stringify(report, null, 2)}\n`); } catch (error) { const code = typeof error?.code === 'string' && /^[A-Z0-9]{5}$/.test(error.code) ? ` (database code ${error.code})` : ''; const reason = error instanceof SafeOperationError ? ` ${error.message}` : ''; process.stderr.write(`Tenant rehome stopped.${reason}${code} No connection details or row contents were printed.\n`); process.exitCode = 1; } finally { await client?.end().catch(() => undefined); } } if (process.argv[1] && import.meta.url === pathToFileURL(path.resolve(process.argv[1])).href) { await main(); }