Custodial financial storage
This commit is contained in:
Executable
+89
@@ -0,0 +1,89 @@
|
||||
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(); }
|
||||
}
|
||||
Executable
+45
@@ -0,0 +1,45 @@
|
||||
import { DatabaseSync } from 'node:sqlite';
|
||||
import { readFile, open, lstat, mkdir } from 'node:fs/promises';
|
||||
import { join } from 'node:path';
|
||||
import { CustodyMarket } from './custody.ts';
|
||||
import { PostgresCustody, type CustodyOptions } from './postgres-custody.ts';
|
||||
import { custodyStorage, postgresConfig } from './postgres-config.ts';
|
||||
|
||||
/** No automatic SQLite reinterpretation or fallback after a database connection/configuration failure. */
|
||||
export async function openCustodyStorage(directory: string, options: CustodyOptions, env = process.env): Promise<CustodyMarket | PostgresCustody> {
|
||||
const storage = custodyStorage(env), marker = join(directory, 'testnet', 'custody-postgres.json');
|
||||
await mkdir(join(directory, 'testnet'), { recursive: true, mode: 0o700 });
|
||||
let existing: { version: number; custodianId: string } | undefined;
|
||||
try { existing = JSON.parse(await readFile(marker, 'utf8')); }
|
||||
catch (e) { if ((e as NodeJS.ErrnoException).code !== 'ENOENT') throw e; }
|
||||
if (storage === 'sqlite') {
|
||||
if (existing) throw new Error('Custody data directory is bound to PostgreSQL; SQLite fallback prohibited');
|
||||
return CustodyMarket.open(directory, options);
|
||||
}
|
||||
const source = join(directory, 'testnet', 'custody-sandbox.sqlite');
|
||||
let frozen: string | undefined;
|
||||
try {
|
||||
const info = await lstat(source);
|
||||
if (!info.isFile() || info.isSymbolicLink()) throw new Error('Invalid custody source file');
|
||||
const db = new DatabaseSync(source, { readOnly: true });
|
||||
try {
|
||||
frozen = (db.prepare("SELECT value FROM custody_meta WHERE key='postgres_cutover'").get() as { value: string } | undefined)?.value;
|
||||
if (!frozen) throw new Error('Existing SQLite custody requires explicit stopped-node import before PostgreSQL activation');
|
||||
} finally { db.close(); }
|
||||
} catch (e) { if ((e as NodeJS.ErrnoException).code !== 'ENOENT') throw e; }
|
||||
const market = await PostgresCustody.open(await postgresConfig(env), options);
|
||||
try {
|
||||
if (frozen && await market.importSource() !== frozen) throw new Error('PostgreSQL import does not match local frozen custody source');
|
||||
if (existing && (existing.version !== 1 || existing.custodianId !== market.id)) throw new Error('PostgreSQL custodian differs from data-directory binding');
|
||||
if (!existing) {
|
||||
const handle = await open(marker, 'wx', 0o600);
|
||||
try { await handle.writeFile(JSON.stringify({ version: 1, custodianId: market.id })); await handle.sync(); }
|
||||
finally { await handle.close(); }
|
||||
if (process.platform !== 'win32') {
|
||||
const parent = await open(join(directory, 'testnet'), 'r');
|
||||
try { await parent.sync(); } finally { await parent.close(); }
|
||||
}
|
||||
}
|
||||
return market;
|
||||
} catch (e) { await market.close(); throw e; }
|
||||
}
|
||||
@@ -76,6 +76,9 @@ export class CustodyMarket {
|
||||
db.exec('BEGIN IMMEDIATE');
|
||||
let id: string;
|
||||
try {
|
||||
if (db.prepare("SELECT 1 FROM custody_meta WHERE key='postgres_cutover'").get()) {
|
||||
throw new Error('SQLite custody was frozen for PostgreSQL cutover; resume the import, never restart SQLite');
|
||||
}
|
||||
const row = db.prepare('SELECT value FROM custody_meta WHERE key=?').get('identity') as { value: string } | undefined;
|
||||
if (row) {
|
||||
const identity = JSON.parse(row.value);
|
||||
|
||||
Executable
+34
@@ -0,0 +1,34 @@
|
||||
import { lstat, readFile } from 'node:fs/promises';
|
||||
import type { PoolConfig } from 'pg';
|
||||
|
||||
export function custodyStorage(env: NodeJS.ProcessEnv = process.env): 'sqlite' | 'postgres' {
|
||||
const value = env.WIKIDECK_CUSTODY_STORAGE ?? 'sqlite';
|
||||
if (value !== 'sqlite' && value !== 'postgres') throw new Error('Invalid custody storage backend');
|
||||
return value;
|
||||
}
|
||||
async function secretFile(path: string | undefined): Promise<string> {
|
||||
if (!path) throw new Error('Custody PostgreSQL secret file is required');
|
||||
const info = await lstat(path);
|
||||
if (!info.isFile() || info.isSymbolicLink() || info.size > 16384 || info.size === 0) throw new Error('Invalid PostgreSQL secret file');
|
||||
return (await readFile(path, 'utf8')).trim();
|
||||
}
|
||||
/** No URL logging, no TLS options in a connection string overriding verified transport. */
|
||||
export async function postgresConfig(env: NodeJS.ProcessEnv = process.env): Promise<PoolConfig> {
|
||||
try {
|
||||
const url = new URL(await secretFile(env.WIKIDECK_CUSTODY_PG_URL_FILE));
|
||||
if (!['postgres:', 'postgresql:'].includes(url.protocol) || !url.hostname || url.search || url.hash ||
|
||||
!/^\/[a-zA-Z0-9_-]+$/.test(url.pathname)) throw new Error('Invalid PostgreSQL URL');
|
||||
const insecure = env.WIKIDECK_CUSTODY_PG_LOCAL_TEST === '1';
|
||||
if (env.WIKIDECK_CUSTODY_PG_LOCAL_TEST !== undefined && !insecure) throw new Error('Invalid local-test switch');
|
||||
if (insecure && !['127.0.0.1', 'localhost', '[::1]'].includes(url.hostname)) throw new Error('Non-TLS test connections must be loopback');
|
||||
if (!insecure && (!url.username || !url.password)) throw new Error('PostgreSQL credentials required');
|
||||
if (url.port && (!/^\d+$/.test(url.port) || Number(url.port) < 1 || Number(url.port) > 65535)) throw new Error('Invalid PostgreSQL port');
|
||||
const ca = env.WIKIDECK_CUSTODY_PG_CA_FILE ? await secretFile(env.WIKIDECK_CUSTODY_PG_CA_FILE) : undefined;
|
||||
return { host: url.hostname.replace(/^\[|\]$/g, ''), port: url.port ? Number(url.port) : 5432,
|
||||
database: url.pathname.slice(1), user: decodeURIComponent(url.username), password: decodeURIComponent(url.password),
|
||||
ssl: insecure ? false : { rejectUnauthorized: true, ...(ca ? { ca } : {}) },
|
||||
max: 4, connectionTimeoutMillis: 5000, idleTimeoutMillis: 10000,
|
||||
statement_timeout: 10000, lock_timeout: 3000, idle_in_transaction_session_timeout: 10000,
|
||||
application_name: 'wikideck-custody', options: '-c search_path=wk_custody,pg_catalog' };
|
||||
} catch { throw new Error('Invalid or inaccessible custody PostgreSQL configuration; check secret files and verified TLS settings'); }
|
||||
}
|
||||
Executable
+241
@@ -0,0 +1,241 @@
|
||||
import { Pool, type PoolClient, type PoolConfig } from 'pg';
|
||||
import { createHash, randomUUID } from 'node:crypto';
|
||||
import { IS_MAINNET, NETWORK } from '../../shared/network.ts';
|
||||
import { CHAIN_ID, verifySignature } from '../testnet/consensus.ts';
|
||||
import { ADDITIONAL_MARKET_FEE } from '../../shared/market.ts';
|
||||
import { allocateXmrSale } from '../../shared/xmr-pricing.ts';
|
||||
import { parseAtomicXmr } from '../../shared/xmr-amount.ts';
|
||||
import { CUSTODY_POLICY, MOCK_CREDIT_ATOMIC, custodyCommandSchema, custodyMessage,
|
||||
type CustodyActivity, type CustodyCommand, type CustodyInfo, type CustodyListing,
|
||||
type CustodyOrder, type CustodyPage, type CustodyWithdrawal } from '../../shared/custody.ts';
|
||||
import { CustodyError } from './custody.ts';
|
||||
import { CUSTODY_WRITE_LOCK, verifyCustodySchema } from './postgres-schema.ts';
|
||||
|
||||
export type CustodyOptions = { feeBps: number; card(hash: string, owner: string, sequence: number): { title: string } | undefined; now?: () => number };
|
||||
type Result = CustodyActivity | CustodyListing | CustodyOrder | CustodyWithdrawal;
|
||||
export type CustodyAudit = { journals: number; assetsAtomicXmr: string; claimsAndRevenueAtomicXmr: string };
|
||||
type Line = { account: string; normal?: 'debit' | 'credit'; debit?: bigint; credit?: bigint };
|
||||
const available = (actor: string) => `customer:${actor}:available`;
|
||||
const reserved = (o: CustodyOrder) => `customer:${o.buyer}:purchase:${o.id}`;
|
||||
const held = (w: CustodyWithdrawal) => `customer:${w.actor}:withdrawal:${w.id}`;
|
||||
function check(ok: unknown, message: string): asserts ok { if (!ok) throw new CustodyError(message); }
|
||||
|
||||
/** PostgreSQL-backed SAME sandbox protocol. No real XMR, NFT witness, or automatic schema migration. */
|
||||
export class PostgresCustody {
|
||||
readonly id: string;
|
||||
private readonly pool: Pool;
|
||||
private readonly options: CustodyOptions;
|
||||
private active = 0;
|
||||
private constructor(pool: Pool, options: CustodyOptions, id: string) { this.pool = pool; this.options = options; this.id = id; }
|
||||
static async open(config: PoolConfig, options: CustodyOptions): Promise<PostgresCustody> {
|
||||
if (IS_MAINNET) throw new Error('Custody sandbox cannot run on mainnet');
|
||||
if (!Number.isSafeInteger(options.feeBps) || options.feeBps < 0 || options.feeBps > 5000) throw new Error('Invalid custody fee');
|
||||
const pool = new Pool(config);
|
||||
// Never log connection details or let idle-client socket errors become uncaught process exceptions.
|
||||
pool.on('error', () => { console.error('Custody PostgreSQL connection unavailable; requests will fail closed.'); });
|
||||
try {
|
||||
const c = await pool.connect();
|
||||
let id: string;
|
||||
try {
|
||||
await c.query('BEGIN');
|
||||
await c.query('SELECT pg_advisory_xact_lock($1::bigint)', [CUSTODY_WRITE_LOCK]);
|
||||
await verifyCustodySchema(c);
|
||||
const privilege = await c.query(`SELECT r.rolsuper,r.rolcreatedb,r.rolcreaterole,r.rolreplication,r.rolbypassrls,
|
||||
pg_has_role(current_user,n.nspowner,'MEMBER') AS owns_schema
|
||||
FROM pg_roles r CROSS JOIN pg_namespace n WHERE r.rolname=current_user AND n.nspname='wk_custody'`);
|
||||
if (!privilege.rowCount || Object.values(privilege.rows[0]).some(Boolean)) {
|
||||
throw new Error('Custody runtime requires a non-administrative role separate from the schema owner');
|
||||
}
|
||||
const existing = await c.query("SELECT value FROM wk_custody.custody_meta WHERE key='identity'");
|
||||
if (existing.rowCount) {
|
||||
const v = JSON.parse(existing.rows[0].value);
|
||||
if (v.version !== 1 || v.network !== NETWORK || v.chainId !== CHAIN_ID || v.mode !== 'sandbox' ||
|
||||
v.policy !== CUSTODY_POLICY || !/^[a-f0-9]{8}(-[a-f0-9]{4}){3}-[a-f0-9]{12}$/.test(v.id)) throw new Error('Custody PostgreSQL identity mismatch');
|
||||
id = v.id;
|
||||
} else {
|
||||
id = randomUUID();
|
||||
await c.query('INSERT INTO wk_custody.custody_meta VALUES($1,$2)', ['identity', JSON.stringify({ version: 1,
|
||||
network: NETWORK, chainId: CHAIN_ID, mode: 'sandbox', policy: CUSTODY_POLICY, id })]);
|
||||
}
|
||||
await auditPostgres(c);
|
||||
await c.query('COMMIT');
|
||||
} catch (e) { await c.query('ROLLBACK').catch(() => {}); throw e; }
|
||||
finally { c.release(); }
|
||||
return new PostgresCustody(pool, options, id);
|
||||
} catch (e) { await pool.end(); throw e; }
|
||||
}
|
||||
close(): Promise<void> { return this.pool.end(); }
|
||||
info(): CustodyInfo {
|
||||
return { enabled: true, mode: 'sandbox', policy: CUSTODY_POLICY, network: NETWORK, chainId: CHAIN_ID,
|
||||
custodianId: this.id, feeBps: this.options.feeBps, liveDeposits: false, liveWithdrawals: false, nftSettlement: false };
|
||||
}
|
||||
private async transaction<T>(write: boolean, fn: (c: PoolClient) => Promise<T>): Promise<T> {
|
||||
check(this.active < 16, 'Custody database busy; retry the original command shortly');
|
||||
this.active++;
|
||||
let c: PoolClient | undefined;
|
||||
try {
|
||||
c = await this.pool.connect();
|
||||
await c.query(write ? 'BEGIN' : 'BEGIN ISOLATION LEVEL REPEATABLE READ READ ONLY');
|
||||
await c.query("SET LOCAL statement_timeout='10s'; SET LOCAL lock_timeout='3s'; SET LOCAL idle_in_transaction_session_timeout='10s'; SET LOCAL synchronous_commit=on");
|
||||
// One writer per financial domain initially; other readers remain nonblocking. Lock is database-scoped,
|
||||
// shared by every node/process, migrations and import, not an in-process promise queue.
|
||||
if (write) await c.query('SELECT pg_advisory_xact_lock($1::bigint)', [CUSTODY_WRITE_LOCK]);
|
||||
const result = await fn(c);
|
||||
await c.query('COMMIT');
|
||||
return result;
|
||||
} catch (e) {
|
||||
await c?.query('ROLLBACK').catch(() => {});
|
||||
// Never retry after an ambiguous COMMIT. Signed receipts resolve the SAME nonce on a client retry.
|
||||
if (e instanceof CustodyError) throw e;
|
||||
if ((e as { code?: string }).code === '23514') throw new CustodyError('Insufficient balance or custody amount constraint');
|
||||
throw new Error('Custody PostgreSQL operation failed; refresh or retry the original signed command');
|
||||
} finally { c?.release(); this.active--; }
|
||||
}
|
||||
async browse(offset = 0): Promise<CustodyPage> {
|
||||
check(Number.isSafeInteger(offset) && offset >= 0 && offset <= 10000, 'Invalid custody page');
|
||||
return this.transaction(false, async c => {
|
||||
const { rows } = await c.query("SELECT data FROM wk_custody.listings WHERE status IN ('open','reserved') ORDER BY sequence DESC LIMIT 25 OFFSET $1", [offset]);
|
||||
return { info: this.info(), listings: rows.slice(0, 24).map(r => r.data), next: rows.length > 24 ? offset + 24 : null };
|
||||
});
|
||||
}
|
||||
private async activity(c: PoolClient, actor: string, offset: number): Promise<CustodyActivity> {
|
||||
const orders = await c.query('SELECT data FROM wk_custody.orders WHERE buyer=$1 OR seller=$1 ORDER BY sequence DESC LIMIT 25 OFFSET $2', [actor, offset]);
|
||||
const withdrawals = await c.query('SELECT data FROM wk_custody.withdrawals WHERE actor=$1 ORDER BY sequence DESC LIMIT 25 OFFSET $2', [actor, offset]);
|
||||
const sums = await c.query(`SELECT COALESCE(sum(balance) FILTER(WHERE id=$1),0)::text AS available,
|
||||
COALESCE(sum(balance) FILTER(WHERE id LIKE $2),0)::text AS purchase,
|
||||
COALESCE(sum(balance) FILTER(WHERE id LIKE $3),0)::text AS withdrawal FROM wk_custody.accounts WHERE id LIKE $4`,
|
||||
[available(actor), `customer:${actor}:purchase:%`, `customer:${actor}:withdrawal:%`, `customer:${actor}:%`]);
|
||||
const claimed = await c.query('SELECT 1 FROM wk_custody.credits WHERE actor=$1', [actor]);
|
||||
return { availableAtomicXmr: sums.rows[0].available, purchaseReservedAtomicXmr: sums.rows[0].purchase,
|
||||
withdrawalReservedAtomicXmr: sums.rows[0].withdrawal, mockCreditClaimed: !!claimed.rowCount,
|
||||
orders: orders.rows.slice(0, 24).map(r => r.data), withdrawals: withdrawals.rows.slice(0, 24).map(r => r.data),
|
||||
next: orders.rows.length > 24 || withdrawals.rows.length > 24 ? offset + 24 : null };
|
||||
}
|
||||
private async post(c: PoolClient, id: string, kind: string, lines: Line[]) {
|
||||
const nonzero = lines.filter(l => (l.debit ?? 0n) !== 0n || (l.credit ?? 0n) !== 0n);
|
||||
if (!nonzero.length || nonzero.reduce((s, l) => s + (l.debit ?? 0n) - (l.credit ?? 0n), 0n) !== 0n) throw new Error('Unbalanced custody posting');
|
||||
await c.query('INSERT INTO wk_custody.journals(id,kind,created) VALUES($1,$2,$3)', [id, kind, (this.options.now ?? Date.now)()]);
|
||||
for (const [position, line] of nonzero.entries()) {
|
||||
const normal = line.normal ?? 'credit';
|
||||
await c.query('INSERT INTO wk_custody.accounts(id,normal) VALUES($1,$2) ON CONFLICT(id) DO NOTHING', [line.account, normal]);
|
||||
// Runtime intentionally has no UPDATE privilege on balances. The SECURITY DEFINER projection
|
||||
// trigger takes the account row lock; global writer admission protects the preceding normal check.
|
||||
const account = await c.query('SELECT normal FROM wk_custody.accounts WHERE id=$1', [line.account]);
|
||||
if (account.rows[0].normal !== normal) throw new Error('Custody account normal mismatch');
|
||||
// DB trigger updates projection; deferred trigger rejects an empty/unbalanced journal at COMMIT.
|
||||
await c.query('INSERT INTO wk_custody.entries VALUES($1,$2,$3,$4,$5)',
|
||||
[id, position, line.account, (line.debit ?? 0n).toString(), (line.credit ?? 0n).toString()]);
|
||||
}
|
||||
}
|
||||
private async record<T>(c: PoolClient, table: 'listings' | 'orders' | 'withdrawals', id: string): Promise<T> {
|
||||
const r = await c.query(`SELECT data FROM wk_custody.${table} WHERE id=$1 FOR UPDATE`, [id]);
|
||||
check(r.rowCount, 'Custody record not found'); return r.rows[0].data as T;
|
||||
}
|
||||
private async update(c: PoolClient, table: 'listings' | 'orders' | 'withdrawals', record: { id: string; status: string }) {
|
||||
await c.query(`UPDATE wk_custody.${table} SET status=$1,data=$2 WHERE id=$3`, [record.status, JSON.stringify(record), record.id]);
|
||||
}
|
||||
async execute(raw: unknown): Promise<Result> {
|
||||
const cmd = custodyCommandSchema.parse(raw), message = custodyMessage(cmd), now = (this.options.now ?? Date.now)();
|
||||
check(cmd.custodianId === this.id && cmd.chainId === CHAIN_ID && cmd.expiresAt >= now && cmd.expiresAt <= now + 300000 &&
|
||||
verifySignature(cmd.publicKey, cmd.actor, message, cmd.signature), 'Invalid or expired custody signature');
|
||||
const a = cmd.action;
|
||||
if (a.type === 'activity') return this.transaction(false, c => this.activity(c, cmd.actor, a.offset));
|
||||
return this.transaction(true, async c => {
|
||||
const key = `${cmd.actor}:${cmd.nonce}`, hash = createHash('sha256').update(message).digest('hex');
|
||||
const old = await c.query('SELECT digest,data FROM wk_custody.receipts WHERE id=$1', [key]);
|
||||
if (old.rowCount) { check(old.rows[0].digest === hash, 'Nonce already used for different custody terms'); return old.rows[0].data as Result; }
|
||||
// Sandbox admission budget matches SQLite. Receipt/history are NEVER pruned; cancellation stays available.
|
||||
if (!['cancel-listing', 'cancel-reservation', 'cancel-withdrawal'].includes(a.type)) {
|
||||
const count = await c.query('SELECT count(*)::integer AS n FROM (SELECT 1 FROM wk_custody.receipts LIMIT 10000) b');
|
||||
check(count.rows[0].n < 10000, 'Custody sandbox command budget reached');
|
||||
}
|
||||
const result = await this.mutate(c, cmd);
|
||||
await c.query('INSERT INTO wk_custody.receipts VALUES($1,$2,$3)', [key, hash, JSON.stringify(result)]);
|
||||
return result;
|
||||
});
|
||||
}
|
||||
private async mutate(c: PoolClient, cmd: CustodyCommand): Promise<Result> {
|
||||
const a = cmd.action, actor = cmd.actor, createdAt = (this.options.now ?? Date.now)();
|
||||
if (a.type === 'mock-credit') {
|
||||
check(!(await c.query('SELECT 1 FROM wk_custody.credits WHERE actor=$1', [actor])).rowCount, 'Sandbox credit already claimed');
|
||||
await this.post(c, `credit:${actor}`, 'mock-deposit', [
|
||||
{ account: 'treasury:mock', normal: 'debit', debit: BigInt(MOCK_CREDIT_ATOMIC) },
|
||||
{ account: available(actor), credit: BigInt(MOCK_CREDIT_ATOMIC) }]);
|
||||
await c.query('INSERT INTO wk_custody.credits VALUES($1)', [actor]); return this.activity(c, actor, 0);
|
||||
}
|
||||
if (a.type === 'list') {
|
||||
check(a.feeBps === this.options.feeBps, 'Custody hosting fee changed; review again');
|
||||
const card = this.options.card(a.cardHash, actor, a.sequence);
|
||||
check(card, 'Card is missing, stale, immature, or pending');
|
||||
check(!(await c.query("SELECT 1 FROM wk_custody.listings WHERE card=$1 AND status IN ('open','reserved')", [a.cardHash])).rowCount, 'Card already has a custody listing');
|
||||
const l: CustodyListing = { id: randomUUID(), seller: actor, cardHash: a.cardHash, sequence: a.sequence,
|
||||
title: card.title, priceAtomicXmr: a.priceAtomicXmr, feeBps: a.feeBps, status: 'open', createdAt };
|
||||
await c.query('INSERT INTO wk_custody.listings(id,card,status,data) VALUES($1,$2,$3,$4)', [l.id, l.cardHash, l.status, JSON.stringify(l)]); return l;
|
||||
}
|
||||
if (a.type === 'reserve' || a.type === 'cancel-listing') {
|
||||
const l = await this.record<CustodyListing>(c, 'listings', a.listingId);
|
||||
check(l.status === 'open', 'Custody listing is no longer available');
|
||||
if (a.type === 'cancel-listing') {
|
||||
check(l.seller === actor, 'Only the seller can cancel this listing'); l.status = 'cancelled'; await this.update(c, 'listings', l); return l;
|
||||
}
|
||||
check(actor !== l.seller, 'Cannot buy your own card');
|
||||
check(a.priceAtomicXmr === l.priceAtomicXmr && a.feeBps === l.feeBps, 'Custody sale terms changed');
|
||||
check(this.options.card(l.cardHash, l.seller, l.sequence), 'Seller card is no longer available');
|
||||
const allocation = allocateXmrSale(parseAtomicXmr(a.priceAtomicXmr), a.feeBps);
|
||||
const o: CustodyOrder = { id: randomUUID(), listingId: l.id, seller: l.seller, buyer: actor, cardHash: l.cardHash,
|
||||
sequence: l.sequence, grossAtomicXmr: a.priceAtomicXmr, sellerAtomicXmr: allocation.sellerBeforeNetworkCosts.toString(),
|
||||
authorAtomicXmr: allocation.author.toString(), hostAtomicXmr: allocation.hosting.toString(), authorAddress: ADDITIONAL_MARKET_FEE.address,
|
||||
status: 'awaiting-lock', createdAt };
|
||||
await this.post(c, `reserve:${o.id}`, 'purchase-reservation', [{ account: available(actor), debit: allocation.gross }, { account: reserved(o), credit: allocation.gross }]);
|
||||
l.status = 'reserved'; await this.update(c, 'listings', l);
|
||||
await c.query('INSERT INTO wk_custody.orders(id,buyer,seller,status,data) VALUES($1,$2,$3,$4,$5)', [o.id, actor, o.seller, o.status, JSON.stringify(o)]);
|
||||
await c.query('INSERT INTO wk_custody.outbox VALUES($1,$2,$3,$4)', [o.id, 'nft-lock', 'blocked', JSON.stringify(o)]); return o;
|
||||
}
|
||||
if (a.type === 'cancel-reservation') {
|
||||
const o = await this.record<CustodyOrder>(c, 'orders', a.orderId);
|
||||
const intent = await c.query('SELECT status FROM wk_custody.outbox WHERE id=$1 FOR UPDATE', [o.id]);
|
||||
check(o.buyer === actor && o.status === 'awaiting-lock' && intent.rows[0]?.status === 'blocked', 'Only an unstarted reservation can be cancelled by its buyer');
|
||||
await this.post(c, `cancel:${o.id}`, 'purchase-cancellation', [{ account: reserved(o), debit: BigInt(o.grossAtomicXmr) }, { account: available(actor), credit: BigInt(o.grossAtomicXmr) }]);
|
||||
o.status = 'cancelled'; await this.update(c, 'orders', o);
|
||||
const l = await this.record<CustodyListing>(c, 'listings', o.listingId); l.status = 'cancelled'; await this.update(c, 'listings', l);
|
||||
await c.query("UPDATE wk_custody.outbox SET status='cancelled' WHERE id=$1", [o.id]); return o;
|
||||
}
|
||||
if (a.type === 'withdraw') {
|
||||
const w: CustodyWithdrawal = { id: randomUUID(), actor, destination: a.destination, amountAtomicXmr: a.amountAtomicXmr,
|
||||
maxFeeAtomicXmr: a.maxFeeAtomicXmr, status: 'reserved', createdAt };
|
||||
const total = BigInt(a.amountAtomicXmr) + BigInt(a.maxFeeAtomicXmr);
|
||||
await this.post(c, `withdraw:${w.id}`, 'withdrawal-reservation', [{ account: available(actor), debit: total }, { account: held(w), credit: total }]);
|
||||
await c.query('INSERT INTO wk_custody.withdrawals(id,actor,status,data) VALUES($1,$2,$3,$4)', [w.id, actor, w.status, JSON.stringify(w)]);
|
||||
await c.query('INSERT INTO wk_custody.outbox VALUES($1,$2,$3,$4)', [w.id, 'withdrawal', 'blocked', JSON.stringify(w)]); return w;
|
||||
}
|
||||
if (a.type === 'cancel-withdrawal') {
|
||||
const w = await this.record<CustodyWithdrawal>(c, 'withdrawals', a.withdrawalId);
|
||||
check(w.actor === actor && w.status === 'reserved', 'Only an unprepared withdrawal can be cancelled by its owner');
|
||||
const total = BigInt(w.amountAtomicXmr) + BigInt(w.maxFeeAtomicXmr);
|
||||
await this.post(c, `withdraw-cancel:${w.id}`, 'withdrawal-cancellation', [{ account: held(w), debit: total }, { account: available(actor), credit: total }]);
|
||||
w.status = 'cancelled'; await this.update(c, 'withdrawals', w);
|
||||
await c.query("UPDATE wk_custody.outbox SET status='cancelled' WHERE id=$1", [w.id]); return w;
|
||||
}
|
||||
throw new CustodyError('Unsupported custody action');
|
||||
}
|
||||
audit(): Promise<CustodyAudit> { return this.transaction(false, auditPostgres); }
|
||||
importSource(): Promise<string | undefined> {
|
||||
return this.transaction(false, async c => (await c.query("SELECT value FROM wk_custody.custody_meta WHERE key='import_source'")).rows[0]?.value);
|
||||
}
|
||||
}
|
||||
|
||||
/** Exact NUMERIC/text aggregation in a coherent transaction; no Number conversion of money. */
|
||||
export async function auditPostgres(c: PoolClient): Promise<CustodyAudit> {
|
||||
const bad = await c.query(`SELECT 1 FROM wk_custody.journals j LEFT JOIN wk_custody.entries e ON e.journal=j.id
|
||||
GROUP BY j.id HAVING count(e.position)=0 OR sum(e.debit-e.credit)<>0 LIMIT 1`);
|
||||
if (bad.rowCount) throw new Error('Unbalanced PostgreSQL custody history');
|
||||
const mismatch = await c.query(`SELECT 1 FROM wk_custody.accounts a LEFT JOIN wk_custody.entries e ON e.account=a.id
|
||||
GROUP BY a.id HAVING a.balance<>COALESCE(sum(CASE WHEN a.normal='debit' THEN e.debit-e.credit ELSE e.credit-e.debit END),0) LIMIT 1`);
|
||||
if (mismatch.rowCount) throw new Error('PostgreSQL custody balance projection mismatch');
|
||||
const totals = await c.query(`SELECT COALESCE(sum(balance) FILTER(WHERE normal='debit'),0)::text AS assets,
|
||||
COALESCE(sum(balance) FILTER(WHERE normal='credit'),0)::text AS claims FROM wk_custody.accounts`);
|
||||
if (totals.rows[0].assets !== totals.rows[0].claims) throw new Error('PostgreSQL custody backing mismatch');
|
||||
const n = await c.query('SELECT count(*)::text AS n FROM wk_custody.journals');
|
||||
if (BigInt(n.rows[0].n) > BigInt(Number.MAX_SAFE_INTEGER)) throw new Error('Journal count exceeds supported range');
|
||||
return { journals: Number(n.rows[0].n), assetsAtomicXmr: totals.rows[0].assets, claimsAndRevenueAtomicXmr: totals.rows[0].claims };
|
||||
}
|
||||
Executable
+114
@@ -0,0 +1,114 @@
|
||||
import { createHash } from 'node:crypto';
|
||||
import type { Pool, PoolClient } from 'pg';
|
||||
|
||||
export const CUSTODY_PG_VERSION = 1;
|
||||
export const CUSTODY_WRITE_LOCK = '7140824361';
|
||||
export const CUSTODY_SCHEMA = `
|
||||
CREATE SCHEMA IF NOT EXISTS wk_custody;
|
||||
REVOKE ALL ON SCHEMA wk_custody FROM PUBLIC;
|
||||
SET LOCAL search_path = wk_custody, pg_catalog;
|
||||
CREATE TABLE schema_versions(version integer PRIMARY KEY, digest text NOT NULL);
|
||||
CREATE TABLE custody_meta(key text PRIMARY KEY, value text NOT NULL);
|
||||
CREATE TABLE accounts(id text PRIMARY KEY, normal text NOT NULL CHECK(normal IN ('debit','credit')),
|
||||
balance numeric NOT NULL DEFAULT 0 CHECK(balance>=0 AND balance<1e30 AND balance=trunc(balance)));
|
||||
CREATE TABLE journals(sequence bigint GENERATED ALWAYS AS IDENTITY UNIQUE, id text PRIMARY KEY,
|
||||
kind text NOT NULL, created bigint NOT NULL CHECK(created>=0), posting_tx bigint NOT NULL DEFAULT txid_current());
|
||||
CREATE TABLE entries(journal text NOT NULL REFERENCES journals(id), position integer NOT NULL CHECK(position>=0),
|
||||
account text NOT NULL REFERENCES accounts(id), debit numeric NOT NULL, credit numeric NOT NULL,
|
||||
PRIMARY KEY(journal,position), CHECK(debit>=0 AND credit>=0 AND debit<1e30 AND credit<1e30),
|
||||
CHECK(debit=trunc(debit) AND credit=trunc(credit) AND ((debit>0 AND credit=0) OR (credit>0 AND debit=0))));
|
||||
CREATE INDEX entries_account ON entries(account,journal);
|
||||
CREATE TABLE listings(sequence bigint GENERATED BY DEFAULT AS IDENTITY UNIQUE, id uuid PRIMARY KEY, card text NOT NULL,
|
||||
status text NOT NULL CHECK(status IN ('open','reserved','cancelled','sold')), data jsonb NOT NULL);
|
||||
CREATE UNIQUE INDEX one_active_card ON listings(card) WHERE status IN ('open','reserved');
|
||||
CREATE TABLE orders(sequence bigint GENERATED BY DEFAULT AS IDENTITY UNIQUE, id uuid PRIMARY KEY,
|
||||
buyer text NOT NULL, seller text NOT NULL, status text NOT NULL CHECK(status IN ('awaiting-lock','cancelled','settled')), data jsonb NOT NULL);
|
||||
CREATE INDEX orders_buyer ON orders(buyer,sequence DESC);
|
||||
CREATE INDEX orders_seller ON orders(seller,sequence DESC);
|
||||
CREATE TABLE withdrawals(sequence bigint GENERATED BY DEFAULT AS IDENTITY UNIQUE, id uuid PRIMARY KEY, actor text NOT NULL,
|
||||
status text NOT NULL CHECK(status IN ('reserved','prepared','complete','cancelled')), data jsonb NOT NULL);
|
||||
CREATE INDEX withdrawals_actor ON withdrawals(actor,sequence DESC);
|
||||
CREATE TABLE receipts(id text PRIMARY KEY, digest text NOT NULL, data jsonb NOT NULL);
|
||||
CREATE TABLE outbox(id uuid PRIMARY KEY, kind text NOT NULL CHECK(kind IN ('nft-lock','withdrawal')), status text NOT NULL, data text NOT NULL);
|
||||
CREATE TABLE credits(actor text PRIMARY KEY);
|
||||
CREATE FUNCTION immutable_history() RETURNS trigger LANGUAGE plpgsql SET search_path=wk_custody,pg_catalog AS $$
|
||||
BEGIN RAISE EXCEPTION 'Immutable financial history'; END $$;
|
||||
CREATE TRIGGER entries_immutable BEFORE UPDATE OR DELETE ON entries FOR EACH ROW EXECUTE FUNCTION immutable_history();
|
||||
CREATE TRIGGER journals_immutable BEFORE UPDATE OR DELETE ON journals FOR EACH ROW EXECUTE FUNCTION immutable_history();
|
||||
CREATE TRIGGER receipts_immutable BEFORE UPDATE OR DELETE ON receipts FOR EACH ROW EXECUTE FUNCTION immutable_history();
|
||||
CREATE FUNCTION apply_entry() RETURNS trigger LANGUAGE plpgsql SECURITY DEFINER SET search_path=wk_custody,pg_catalog AS $$
|
||||
DECLARE n text; amount numeric;
|
||||
BEGIN
|
||||
IF NOT EXISTS(SELECT 1 FROM journals WHERE id=NEW.journal AND posting_tx=txid_current()) THEN
|
||||
RAISE EXCEPTION 'Cannot append to a committed journal';
|
||||
END IF;
|
||||
SELECT normal INTO STRICT n FROM accounts WHERE id=NEW.account FOR UPDATE;
|
||||
amount := CASE WHEN n='debit' THEN NEW.debit-NEW.credit ELSE NEW.credit-NEW.debit END;
|
||||
UPDATE accounts SET balance=balance+amount WHERE id=NEW.account;
|
||||
RETURN NEW;
|
||||
END $$;
|
||||
CREATE TRIGGER entry_projection AFTER INSERT ON entries FOR EACH ROW EXECUTE FUNCTION apply_entry();
|
||||
CREATE FUNCTION balanced_journal() RETURNS trigger LANGUAGE plpgsql SET search_path=wk_custody,pg_catalog AS $$
|
||||
DECLARE total numeric; n bigint; journal_id text;
|
||||
BEGIN
|
||||
IF TG_TABLE_NAME='entries' THEN journal_id := NEW.journal; ELSE journal_id := NEW.id; END IF;
|
||||
SELECT count(*),sum(debit-credit) INTO n,total FROM entries WHERE journal=journal_id;
|
||||
IF n=0 OR total<>0 THEN RAISE EXCEPTION 'Unbalanced financial journal'; END IF;
|
||||
RETURN NULL;
|
||||
END $$;
|
||||
CREATE CONSTRAINT TRIGGER journal_balance AFTER INSERT ON journals DEFERRABLE INITIALLY DEFERRED
|
||||
FOR EACH ROW EXECUTE FUNCTION balanced_journal();
|
||||
CREATE CONSTRAINT TRIGGER entry_balance AFTER INSERT ON entries DEFERRABLE INITIALLY DEFERRED
|
||||
FOR EACH ROW EXECUTE FUNCTION balanced_journal();
|
||||
REVOKE ALL ON ALL TABLES IN SCHEMA wk_custody FROM PUBLIC;
|
||||
REVOKE ALL ON ALL SEQUENCES IN SCHEMA wk_custody FROM PUBLIC;
|
||||
REVOKE ALL ON ALL FUNCTIONS IN SCHEMA wk_custody FROM PUBLIC;
|
||||
`;
|
||||
export const CUSTODY_SCHEMA_DIGEST = createHash('sha256').update(CUSTODY_SCHEMA).digest('hex');
|
||||
|
||||
/** Explicit operator migration; the HTTP runtime never creates or upgrades schema. */
|
||||
export async function migrateCustody(pool: Pool): Promise<void> {
|
||||
const c = await pool.connect();
|
||||
try {
|
||||
await c.query('BEGIN');
|
||||
await c.query('SELECT pg_advisory_xact_lock($1::bigint)', [CUSTODY_WRITE_LOCK]);
|
||||
const exists = await c.query("SELECT to_regclass('wk_custody.schema_versions') AS name");
|
||||
if (exists.rows[0].name) await verifyCustodySchema(c);
|
||||
else {
|
||||
await c.query(CUSTODY_SCHEMA);
|
||||
await c.query('INSERT INTO wk_custody.schema_versions VALUES($1,$2)', [CUSTODY_PG_VERSION, CUSTODY_SCHEMA_DIGEST]);
|
||||
}
|
||||
await c.query('COMMIT');
|
||||
} catch (e) { await c.query('ROLLBACK').catch(() => {}); throw e; }
|
||||
finally { c.release(); }
|
||||
}
|
||||
export async function verifyCustodySchema(c: PoolClient): Promise<void> {
|
||||
const rows = await c.query('SELECT version,digest FROM wk_custody.schema_versions ORDER BY version');
|
||||
if (rows.rowCount !== 1 || rows.rows[0].version !== CUSTODY_PG_VERSION || rows.rows[0].digest !== CUSTODY_SCHEMA_DIGEST) {
|
||||
throw new Error('Custody PostgreSQL schema version/digest mismatch; explicit migration required');
|
||||
}
|
||||
}
|
||||
|
||||
/** Role must already exist (operator creates credentials out-of-band); never accept SQL identifiers from HTTP. */
|
||||
export async function grantCustodyRuntime(pool: Pool, role: string): Promise<void> {
|
||||
if (!/^[a-z][a-z0-9_]{0,62}$/.test(role)) throw new Error('Invalid runtime role identifier');
|
||||
const name = `"${role}"`;
|
||||
const c = await pool.connect();
|
||||
try {
|
||||
await c.query('BEGIN');
|
||||
await verifyCustodySchema(c);
|
||||
const r = await c.query('SELECT rolsuper,rolcreatedb,rolcreaterole,rolreplication,rolbypassrls FROM pg_roles WHERE rolname=$1', [role]);
|
||||
if (!r.rowCount || Object.values(r.rows[0]).some(Boolean)) throw new Error('Runtime role must exist without administrative privileges');
|
||||
const owner = await c.query("SELECT pg_get_userbyid(nspowner) AS owner FROM pg_namespace WHERE nspname='wk_custody'");
|
||||
if (owner.rows[0].owner === role) throw new Error('Runtime role must not own the custody schema');
|
||||
await c.query(`GRANT USAGE ON SCHEMA wk_custody TO ${name};
|
||||
GRANT SELECT ON ALL TABLES IN SCHEMA wk_custody TO ${name};
|
||||
GRANT USAGE,SELECT ON ALL SEQUENCES IN SCHEMA wk_custody TO ${name};
|
||||
GRANT INSERT(id,normal) ON wk_custody.accounts TO ${name};
|
||||
GRANT INSERT(id,kind,created) ON wk_custody.journals TO ${name};
|
||||
GRANT INSERT ON wk_custody.entries,wk_custody.receipts,wk_custody.credits,wk_custody.custody_meta TO ${name};
|
||||
GRANT INSERT,UPDATE ON wk_custody.listings,wk_custody.orders,wk_custody.withdrawals,wk_custody.outbox TO ${name};`);
|
||||
await c.query('COMMIT');
|
||||
} catch (e) { await c.query('ROLLBACK').catch(() => {}); throw e; }
|
||||
finally { c.release(); }
|
||||
}
|
||||
@@ -30,6 +30,9 @@ function inspect(file:string,name:typeof files[number]) {
|
||||
const meta=(key:string)=>(db.prepare('SELECT value FROM metadata WHERE key=?').get(key) as {value:string}|undefined)?.value;
|
||||
if(meta('schema_version')!=='1'||meta('network')!==NETWORK||meta('chain_id')!==CHAIN_ID)throw new Error('Ledger network/schema mismatch');
|
||||
}else if(name==='custody-sandbox.sqlite'){
|
||||
if(db.prepare("SELECT 1 FROM custody_meta WHERE key='postgres_cutover'").get()) {
|
||||
throw new Error('PostgreSQL cutover source cannot be backed up as current custody; coordinated database backup required');
|
||||
}
|
||||
const row=db.prepare("SELECT value FROM custody_meta WHERE key='identity'").get() as {value:string}|undefined;
|
||||
const identity=row?JSON.parse(row.value):undefined;
|
||||
if(identity?.version!==1||identity.network!==NETWORK||identity.chainId!==CHAIN_ID||identity.mode!=='sandbox'||
|
||||
@@ -53,6 +56,9 @@ export async function createBackup(dataDir:string,destination:string){
|
||||
const rel=relative(source,dest);
|
||||
if(!rel||(!rel.startsWith('..')&&!isAbsolute(rel)))throw new Error('Backup destination must be outside the node data directory');
|
||||
return offlineLock(source,async()=>{
|
||||
// File-only backups cannot represent an external financial database. Fail rather than omit liabilities.
|
||||
try { await lstat(join(source,PROFILE,'custody-postgres.json')); throw new Error('PostgreSQL custody requires coordinated database and node backup; SQLite-only backup refused'); }
|
||||
catch(e) { if ((e as NodeJS.ErrnoException).code !== 'ENOENT') throw e; }
|
||||
if(await realpath(join(source,PROFILE))!==join(source,PROFILE))throw new Error('Network data directory must not be symlinked');
|
||||
await mkdir(dest,{mode:0o700}); // Exclusive: never overwrite or recursively create a surprising destination.
|
||||
const entries=[];
|
||||
|
||||
+10
-8
@@ -19,6 +19,8 @@ import { marketCommandSchema } from '../../shared/market.ts';
|
||||
import { marketCommandV3Schema } from '../../shared/market-v3.ts';
|
||||
import { custodyCommandSchema } from '../../shared/custody.ts';
|
||||
import { CustodyMarket, CustodyError, custodyMode } from '../market/custody.ts';
|
||||
import { openCustodyStorage } from '../market/custody-storage.ts';
|
||||
import type { PostgresCustody } from '../market/postgres-custody.ts';
|
||||
import { IS_MAINNET, PROFILE } from '../../shared/network.ts';
|
||||
import { assertLaunchAllowed, launchManifest } from '../launch.ts';
|
||||
import { resourceConfig, RpcBudget, RpcError } from '../resources.ts';
|
||||
@@ -52,9 +54,9 @@ catch(e) {ledger.close();engine.close();await lock.close();await unlink(lockPath
|
||||
let market: Market;
|
||||
try { market = await Market.open(dataDir, ledger, paymentConfig); }
|
||||
catch(e) {peers.close();ledger.close();engine.close();await lock.close();await unlink(lockPath);throw e;}
|
||||
let custody: CustodyMarket | undefined;
|
||||
let custody: CustodyMarket | PostgresCustody | undefined;
|
||||
try {
|
||||
if (custodyConfig === 'sandbox') custody = await CustodyMarket.open(dataDir, { feeBps: paymentConfig.feeBps,
|
||||
if (custodyConfig === 'sandbox') custody = await openCustodyStorage(dataDir, { feeBps: paymentConfig.feeBps,
|
||||
card: (hash, owner, sequence) => {
|
||||
const nft = ledger.nft(hash);
|
||||
if (!nft || nft.owner !== owner || nft.confirmations < CONSENSUS.coinbaseMaturity || nft.sequence !== sequence ||
|
||||
@@ -96,19 +98,19 @@ app.get('/api/network',(_req,res)=>res.json(launchManifest()));
|
||||
app.get('/api/chain',(_req,res)=>res.json(ledger.legacySnapshot(limits.legacyBlocks)));
|
||||
app.get('/api/peers',(_req,res)=>res.json({peers:peers.status()}));
|
||||
app.get('/api/market', (req,res)=>res.json(market.browse(Number(req.query.offset ?? 0))));
|
||||
app.get('/api/custody', (req,res)=> {
|
||||
app.get('/api/custody', async (req,res)=> {
|
||||
if (!custody) { res.json({ enabled: false }); return; }
|
||||
try { res.json(custody.browse(Number(req.query.offset ?? 0))); }
|
||||
try { res.json(await custody.browse(Number(req.query.offset ?? 0))); }
|
||||
catch (error) {
|
||||
if (error instanceof CustodyError) res.status(400).json({ error: error.message });
|
||||
else res.status(503).json({ error: 'Custody sandbox unavailable; operator review required.' });
|
||||
}
|
||||
});
|
||||
app.post('/api/custody/command', rateLimit({windowMs:60000,limit:30,standardHeaders:'draft-8',legacyHeaders:false}), (req,res)=> {
|
||||
app.post('/api/custody/command', rateLimit({windowMs:60000,limit:30,standardHeaders:'draft-8',legacyHeaders:false}), async (req,res)=> {
|
||||
if (!custody) { res.status(503).json({ error: 'Custody sandbox is disabled on this host.' }); return; }
|
||||
const parsed = custodyCommandSchema.safeParse(req.body);
|
||||
if (!parsed.success) { res.status(400).json({ error: 'Invalid custody command schema.' }); return; }
|
||||
try { res.json(custody.execute(parsed.data)); }
|
||||
try { res.json(await custody.execute(parsed.data)); }
|
||||
catch (error) {
|
||||
if (error instanceof CustodyError) res.status(400).json({ error: error.message });
|
||||
else res.status(503).json({ error: 'Custody state conflict; refresh activity before retrying. Operator review may be required.' });
|
||||
@@ -234,6 +236,6 @@ let stopping=false;
|
||||
for(const signal of ['SIGTERM','SIGINT']as const)process.once(signal,()=>{
|
||||
if(stopping)return;stopping=true;peers.close();operations.close();clearInterval(marketTimer);
|
||||
const timeout=setTimeout(()=>process.exit(1),25000);timeout.unref();
|
||||
server.close(async()=>{custody?.close();market.close();ledger.close();engine.close();await lock.close();await unlink(lockPath).catch(()=>{});clearTimeout(timeout);process.exit(0);});
|
||||
server.close(async()=>{await custody?.close();market.close();ledger.close();engine.close();await lock.close();await unlink(lockPath).catch(()=>{});clearTimeout(timeout);process.exit(0);});
|
||||
});
|
||||
server.on('error',async()=>{clearInterval(marketTimer);operations.close();peers.close();custody?.close();market.close();ledger.close();engine.close();await lock.close();await unlink(lockPath).catch(()=>{});process.exit(1);});
|
||||
server.on('error',async()=>{clearInterval(marketTimer);operations.close();peers.close();await custody?.close();market.close();ledger.close();engine.close();await lock.close();await unlink(lockPath).catch(()=>{});process.exit(1);});
|
||||
|
||||
Reference in New Issue
Block a user