489 lines
17 KiB
JavaScript
489 lines
17 KiB
JavaScript
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();
|
|
}
|