90 lines
6.6 KiB
TypeScript
Executable File
90 lines
6.6 KiB
TypeScript
Executable File
import { DatabaseSync } from 'node:sqlite';
|
|
import { createHash } from 'node:crypto';
|
|
import { lstat } from 'node:fs/promises';
|
|
import type { Pool } from 'pg';
|
|
import { CUSTODY_POLICY } from '../../shared/custody.ts';
|
|
import { NETWORK } from '../../shared/network.ts';
|
|
import { CHAIN_ID } from '../testnet/consensus.ts';
|
|
import { auditPostgres } from './postgres-custody.ts';
|
|
import { CUSTODY_WRITE_LOCK, verifyCustodySchema } from './postgres-schema.ts';
|
|
|
|
/** Explicit stopped-node SANDBOX import. Source freezes BEFORE target commit; failure requires resume, not dual writes. */
|
|
export async function importSqliteCustody(file: string, pool: Pool): Promise<{ imported: true; sourceDigest: string; custodianId: string }> {
|
|
const stat = await lstat(file);
|
|
if (!stat.isFile() || stat.isSymbolicLink() || stat.size > 128 * 1024 * 1024) throw new Error('Invalid sandbox source database');
|
|
const db = new DatabaseSync(file);
|
|
db.exec('PRAGMA synchronous=FULL; PRAGMA busy_timeout=5000');
|
|
const names = ['accounts', 'journals', 'entries', 'listings', 'orders', 'withdrawals', 'receipts', 'outbox', 'credits'] as const;
|
|
type Row = Record<string, string | number>;
|
|
let snapshot: Record<typeof names[number], Row[]>;
|
|
let identityText: string;
|
|
let sourceDigest: string;
|
|
try {
|
|
db.exec('BEGIN IMMEDIATE');
|
|
try {
|
|
if (Object.values(db.prepare('PRAGMA quick_check').get()!)[0] !== 'ok') throw new Error('Invalid SQLite integrity');
|
|
identityText = (db.prepare("SELECT value FROM custody_meta WHERE key='identity'").get() as { value: string }).value;
|
|
const identity = JSON.parse(identityText);
|
|
if (identity.version !== 1 || identity.network !== NETWORK || identity.chainId !== CHAIN_ID || identity.mode !== 'sandbox' || identity.policy !== CUSTODY_POLICY) {
|
|
throw new Error('Source custody identity mismatch');
|
|
}
|
|
snapshot = Object.fromEntries(names.map(name => [name, db.prepare(`SELECT rowid AS import_sequence,* FROM ${name} ORDER BY rowid`).all()])) as typeof snapshot;
|
|
sourceDigest = createHash('sha256').update(JSON.stringify([identityText, snapshot])).digest('hex');
|
|
const existing = db.prepare("SELECT value FROM custody_meta WHERE key='postgres_cutover'").get() as { value: string } | undefined;
|
|
if (existing && existing.value !== sourceDigest) throw new Error('Frozen SQLite source changed; operator review required');
|
|
db.prepare('INSERT OR IGNORE INTO custody_meta VALUES(?,?)').run('postgres_cutover', sourceDigest);
|
|
db.exec('COMMIT'); // Source is now permanently ineligible for SQLite startup. Preserve for audit, NEVER delete/reset.
|
|
} catch (e) { db.exec('ROLLBACK'); throw e; }
|
|
} finally { db.close(); }
|
|
const c = await pool.connect();
|
|
try {
|
|
await c.query('BEGIN');
|
|
await c.query('SELECT pg_advisory_xact_lock($1::bigint)', [CUSTODY_WRITE_LOCK]);
|
|
await verifyCustodySchema(c);
|
|
const old = await c.query("SELECT value FROM wk_custody.custody_meta WHERE key='import_source'");
|
|
if (old.rowCount) {
|
|
if (old.rows[0].value !== sourceDigest) throw new Error('Target already imported a different source');
|
|
await auditPostgres(c); await c.query('COMMIT');
|
|
return { imported: true, sourceDigest, custodianId: JSON.parse(identityText).id };
|
|
}
|
|
for (const table of ['custody_meta', ...names]) {
|
|
if ((await c.query(`SELECT 1 FROM wk_custody.${table} LIMIT 1`)).rowCount) throw new Error('Import requires a migrated EMPTY PostgreSQL custody schema');
|
|
}
|
|
for (const a of snapshot.accounts) {
|
|
if (!/^(0|[1-9]\d{0,29})$/.test(String(a.balance))) throw new Error('Invalid source atomic balance');
|
|
await c.query('INSERT INTO wk_custody.accounts(id,normal) VALUES($1,$2)', [a.id, a.normal]);
|
|
}
|
|
// Rebuild balances from journal entries, never trust mutable projections as opening balances.
|
|
const byJournal = new Map<string, Row[]>();
|
|
for (const e of snapshot.entries) {
|
|
const rows = byJournal.get(String(e.journal)) ?? []; rows.push(e); byJournal.set(String(e.journal), rows);
|
|
}
|
|
for (const j of snapshot.journals) {
|
|
await c.query('INSERT INTO wk_custody.journals(id,kind,created) VALUES($1,$2,$3)', [j.id, j.kind, j.created]);
|
|
const lines = (byJournal.get(String(j.id)) ?? []).sort((a, b) => Number(a.position) - Number(b.position));
|
|
for (const e of lines) {
|
|
if (!/^(0|[1-9]\d{0,29})$/.test(String(e.debit)) || !/^(0|[1-9]\d{0,29})$/.test(String(e.credit))) throw new Error('Invalid source entry encoding');
|
|
await c.query('INSERT INTO wk_custody.entries VALUES($1,$2,$3,$4,$5)', [e.journal, e.position, e.account, e.debit, e.credit]);
|
|
}
|
|
}
|
|
const journalIds = new Set(snapshot.journals.map(j => j.id));
|
|
if (snapshot.entries.some(e => !journalIds.has(e.journal))) throw new Error('Orphan source entry');
|
|
for (const a of snapshot.accounts) {
|
|
const rebuilt = await c.query('SELECT balance::text AS balance FROM wk_custody.accounts WHERE id=$1', [a.id]);
|
|
if (rebuilt.rows[0].balance !== a.balance) throw new Error('Source balance projection mismatch');
|
|
}
|
|
for (const l of snapshot.listings) await c.query('INSERT INTO wk_custody.listings(sequence,id,card,status,data) VALUES($1,$2,$3,$4,$5)', [l.import_sequence, l.id, l.card, l.status, l.data]);
|
|
for (const o of snapshot.orders) await c.query('INSERT INTO wk_custody.orders(sequence,id,buyer,seller,status,data) VALUES($1,$2,$3,$4,$5,$6)', [o.import_sequence, o.id, o.buyer, o.seller, o.status, o.data]);
|
|
for (const w of snapshot.withdrawals) await c.query('INSERT INTO wk_custody.withdrawals(sequence,id,actor,status,data) VALUES($1,$2,$3,$4,$5)', [w.import_sequence, w.id, w.actor, w.status, w.data]);
|
|
for (const r of snapshot.receipts) await c.query('INSERT INTO wk_custody.receipts VALUES($1,$2,$3)', [r.id, r.digest, r.data]);
|
|
for (const o of snapshot.outbox) await c.query('INSERT INTO wk_custody.outbox VALUES($1,$2,$3,$4)', [o.id, o.kind, o.status, o.data]);
|
|
for (const a of snapshot.credits) await c.query('INSERT INTO wk_custody.credits VALUES($1)', [a.actor]);
|
|
for (const table of ['listings', 'orders', 'withdrawals']) await c.query(`SELECT setval(pg_get_serial_sequence('wk_custody.${table}','sequence'),COALESCE((SELECT max(sequence) FROM wk_custody.${table}),0)+1,false)`);
|
|
await c.query('INSERT INTO wk_custody.custody_meta VALUES($1,$2),($3,$4)', ['identity', identityText, 'import_source', sourceDigest]);
|
|
await auditPostgres(c);
|
|
await c.query('COMMIT');
|
|
return { imported: true, sourceDigest, custodianId: JSON.parse(identityText).id };
|
|
} catch (e) { await c.query('ROLLBACK').catch(() => {}); throw e; }
|
|
finally { c.release(); }
|
|
}
|