115 lines
7.6 KiB
TypeScript
Executable File
115 lines
7.6 KiB
TypeScript
Executable File
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(); }
|
|
}
|