import test from 'node:test';
import assert from 'node:assert/strict';
import { Pool, type PoolConfig } from 'pg';
import { generateKeyPairSync, randomUUID, sign } from 'node:crypto';
import { mkdtemp, rm, writeFile, mkdir, readFile } from 'node:fs/promises';
import { join } from 'node:path';
import { tmpdir } from 'node:os';
import { spawn } from 'node:child_process';
import { execFile } from 'node:child_process';
import { promisify } from 'node:util';
import { DatabaseSync } from 'node:sqlite';
import { PostgresCustody, auditPostgres } from '../../server/market/postgres-custody.ts';
import { CustodyIntegrityError } from '../../server/market/custody-integrity.ts';
import { resolveCustodyIncident } from '../../server/market/custody-incident.ts';
import { CustodyMarket } from '../../server/market/custody.ts';
import { migrateCustody, grantCustodyRuntime } from '../../server/market/postgres-schema.ts';
import { importSqliteCustody } from '../../server/market/custody-import.ts';
import { postgresConfig, custodyStorage } from '../../server/market/postgres-config.ts';
import { openCustodyStorage } from '../../server/market/custody-storage.ts';
import { createBackup } from '../../server/recovery.ts';
import { CUSTODY_POLICY, custodyMessage, type CustodyAction, type CustodyCommand, type CustodyActivity,
  type CustodyListing, type CustodyOrder, type CustodyWithdrawal } from '../../shared/custody.ts';
import { NETWORK } from '../../shared/network.ts';
import { CHAIN_ID, address } from '../../server/testnet/consensus.ts';

if (process.env.WIKIDECK_PG_DISPOSABLE_TEST !== '1') throw new Error('PostgreSQL tests require the isolated disposable compose fixture');
const config: PoolConfig = { host: '127.0.0.1', database: 'custody_test', user: 'postgres', ssl: false, max: 4,
  connectionTimeoutMillis: 5000, statement_timeout: 10000, lock_timeout: 3000 };
function identity() {
  const keys = generateKeyPairSync('ec', { namedCurve: 'prime256v1' });
  const publicKey = keys.publicKey.export({ type: 'spki', format: 'der' }).toString('base64');
  return { publicKey, actor: address(publicKey), sign: (message: string) => sign('sha256', Buffer.from(message),
    { key: keys.privateKey, dsaEncoding: 'ieee-p1363' }).toString('base64') };
}
const buyer = identity(), seller = identity(), other = identity(), hash = 'a'.repeat(64);
const options = { feeBps: 250, card: (token: string, owner: string, sequence: number) =>
  token === hash && owner === seller.actor && sequence === 0 ? { title: 'PostgreSQL fixture' } : undefined };
function envelope(id: string, who: typeof buyer, action: CustodyAction, nonce: string = randomUUID()): CustodyCommand {
  const c: Omit<CustodyCommand, 'signature'> = { version: 1, mode: 'sandbox', policy: CUSTODY_POLICY, network: NETWORK,
    chainId: CHAIN_ID, custodianId: id, actor: who.actor, publicKey: who.publicKey, nonce, expiresAt: Date.now() + 240000, action };
  return { ...c, signature: who.sign(custodyMessage(c)) };
}
async function fixture(t: { after(fn: () => Promise<void>): void }) {
  const admin = new Pool(config);
  await admin.query('DROP SCHEMA IF EXISTS wk_custody CASCADE'); // This database exists ONLY in isolated tmpfs fixture.
  await migrateCustody(admin);
  await admin.query('DO $$ BEGIN IF NOT EXISTS(SELECT 1 FROM pg_roles WHERE rolname=\'custody_runtime\') THEN CREATE ROLE custody_runtime LOGIN; END IF; END $$');
  await grantCustodyRuntime(admin, 'custody_runtime');
  const runtime = { ...config, user: 'custody_runtime' };
  const market = await PostgresCustody.open(runtime, options);
  t.after(async () => { await market.close(); await admin.end(); });
  const execute = (who: typeof buyer, a: CustodyAction) => market.execute(envelope(market.id, who, a));
  const activity = async (who = buyer) => await execute(who, { type: 'activity', offset: 0 }) as CustodyActivity;
  return { admin, runtime, market, execute, activity };
}

test('explicit PostgreSQL migrations are idempotent and runtime role cannot alter financial history or schema', async t => {
  const f = await fixture(t); await migrateCustody(f.admin);
  await assert.rejects(PostgresCustody.open(config, options), /non-administrative/);
  await f.execute(buyer, { type: 'mock-credit' });
  const runtime = new Pool(f.runtime);
  try {
    for (const sql of ["UPDATE wk_custody.accounts SET balance=999", 'DELETE FROM wk_custody.entries', 'TRUNCATE wk_custody.journals',
      "ALTER TABLE wk_custody.accounts ADD COLUMN bad text", "UPDATE wk_custody.receipts SET digest='bad'"]) await assert.rejects(runtime.query(sql));
    await assert.rejects(grantCustodyRuntime(f.admin, 'postgres'), /administrative/);
    await assert.rejects(grantCustodyRuntime(f.admin, 'bad;DROP SCHEMA'), /identifier/);
  } finally { await runtime.end(); }
  assert.equal((await f.market.audit()).assetsAtomicXmr, '1000000000000');
});

test('database rejects unbalanced, fractional, negative, historical append and oversized postings at commit', async t => {
  const f = await fixture(t); await f.execute(buyer, { type: 'mock-credit' });
  const c = await f.admin.connect();
  try {
    await c.query('BEGIN');
    await c.query("INSERT INTO wk_custody.journals(id,kind,created) VALUES('bad','test',0)");
    await c.query("INSERT INTO wk_custody.entries VALUES('bad',0,'treasury:mock',1,0)");
    await assert.rejects(c.query('COMMIT'), /Unbalanced/); await c.query('ROLLBACK');
    await assert.rejects(c.query("INSERT INTO wk_custody.entries VALUES($1,9,'treasury:mock',1,0)", [`credit:${buyer.actor}`]), /committed/);
    for (const amount of ['-1', '1.5', '1000000000000000000000000000000']) await assert.rejects(c.query('INSERT INTO wk_custody.accounts VALUES($1,$2,$3)', [randomUUID(), 'debit', amount]));
  } finally { c.release(); }
  assert.equal((await f.market.audit()).journals, 1);
});

test('two independent pools serialize duplicate receipts, competing reservations and withdrawals without overspending', async t => {
  const f = await fixture(t), second = await PostgresCustody.open(f.runtime, options);
  try {
    const credit = envelope(f.market.id, buyer, { type: 'mock-credit' });
    const credits = await Promise.all([f.market.execute(credit), second.execute(credit)]);
    assert.deepEqual(credits[0], credits[1]);
    const listing = await f.execute(seller, { type: 'list', cardHash: hash, sequence: 0, priceAtomicXmr: '800000000000', feeBps: 250 }) as CustodyListing;
    await f.execute(other, { type: 'mock-credit' });
    const races = await Promise.allSettled([f.market.execute(envelope(f.market.id, buyer, { type: 'reserve', listingId: listing.id, priceAtomicXmr: listing.priceAtomicXmr, feeBps: 250 })),
      second.execute(envelope(f.market.id, other, { type: 'reserve', listingId: listing.id, priceAtomicXmr: listing.priceAtomicXmr, feeBps: 250 }))]);
    assert.equal(races.filter(r => r.status === 'fulfilled').length, 1);
    const winner = races.find(r => r.status === 'fulfilled') as PromiseFulfilledResult<CustodyOrder>;
    assert.equal(winner.value.sellerAtomicXmr, '772000000000');
    assert.equal((await f.activity(seller)).availableAtomicXmr, '0'); // No unimplemented NFT completion credit.
    const loser = winner.value.buyer === buyer.actor ? other : buyer;
    const withdrawals = await Promise.allSettled([f.market.execute(envelope(f.market.id, loser, { type: 'withdraw', destination: 'mock-external', amountAtomicXmr: '900000000000', maxFeeAtomicXmr: '1000' })),
      second.execute(envelope(f.market.id, loser, { type: 'withdraw', destination: 'mock-external', amountAtomicXmr: '900000000000', maxFeeAtomicXmr: '1000' }))]);
    assert.equal(withdrawals.filter(r => r.status === 'fulfilled').length, 1);
    assert.equal((await f.activity(loser)).withdrawalReservedAtomicXmr, '900000001000');
    assert.equal((await f.market.audit()).assetsAtomicXmr, '2000000000000');
    assert.equal((await f.activity(identity())).orders.length, 0);
  } finally { await second.close(); }
});

test('cancel/replay/reconnect preserves exact money, and receipt-insert failure rolls back balance and outbox', async t => {
  const f = await fixture(t); await f.execute(buyer, { type: 'mock-credit' });
  const w = await f.execute(buyer, { type: 'withdraw', destination: 'mock-external', amountAtomicXmr: '123', maxFeeAtomicXmr: '5' }) as CustodyWithdrawal;
  const cancel = envelope(f.market.id, buyer, { type: 'cancel-withdrawal', withdrawalId: w.id });
  await f.market.execute(cancel);
  const reopened = await PostgresCustody.open(f.runtime, options);
  try {
    assert.equal(reopened.id, f.market.id); assert.deepEqual(await reopened.execute(cancel), await f.market.execute(cancel));
    await assert.rejects(reopened.execute(envelope(f.market.id, buyer, { type: 'mock-credit' }, cancel.nonce)), /Nonce/);
    await f.admin.query("CREATE FUNCTION wk_custody.inject_failure() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN RAISE EXCEPTION 'fixture failure'; END $$; CREATE TRIGGER fail_receipt BEFORE INSERT ON wk_custody.receipts FOR EACH ROW EXECUTE FUNCTION wk_custody.inject_failure()");
    await assert.rejects(f.execute(buyer, { type: 'withdraw', destination: 'mock-external', amountAtomicXmr: '200', maxFeeAtomicXmr: '0' }));
    assert.equal((await f.activity()).availableAtomicXmr, '1000000000000');
    assert.equal((await f.activity()).withdrawals.length, 1);
    await f.admin.query('DROP TRIGGER fail_receipt ON wk_custody.receipts');
    assert.equal((await f.market.audit()).journals, 3);
  } finally { await reopened.close(); }
});

test('exact amounts above JS safe integer survive NUMERIC, JSON signatures and audit', async t => {
  const f = await fixture(t);
  // Seed a balanced synthetic fixture through SQL, not an HTTP mint/admin endpoint.
  const c = await f.admin.connect(), large = '9007199254740993';
  try {
    await c.query('BEGIN');
    await c.query("INSERT INTO wk_custody.accounts(id,normal) VALUES('treasury:mock','debit'),($1,'credit')", [`customer:${buyer.actor}:available`]);
    await c.query("INSERT INTO wk_custody.journals(id,kind,created) VALUES('large','fixture',0)");
    await c.query("INSERT INTO wk_custody.entries VALUES('large',0,'treasury:mock',$1,0),('large',1,$2,0,$1)", [large, `customer:${buyer.actor}:available`]);
    await c.query('COMMIT');
  } finally { c.release(); }
  assert.equal((await f.activity()).availableAtomicXmr, large);
  await f.execute(buyer, { type: 'withdraw', destination: 'mock-external', amountAtomicXmr: large, maxFeeAtomicXmr: '0' });
  assert.equal((await f.activity()).withdrawalReservedAtomicXmr, large); assert.equal((await f.market.audit()).assetsAtomicXmr, large);
});

test('SQLite cutover preserves identity, history, prepared attempts, ordering and replay; source cannot restart', async t => {
  const f = await fixture(t);
  // Import requires truly empty target, even a previously initialized empty custodian is rejected.
  await f.admin.query('DELETE FROM wk_custody.custody_meta');
  const dir = await mkdtemp(join(tmpdir(), 'custody-pg-import-'));
  t.after(() => rm(dir, { recursive: true, force: true }));
  const source = await CustodyMarket.open(dir, options);
  const originalId = source.id, credit = envelope(originalId, buyer, { type: 'mock-credit' });
  source.execute(credit);
  const w = source.execute(envelope(originalId, buyer, { type: 'withdraw', destination: 'mock-external', amountAtomicXmr: '12345', maxFeeAtomicXmr: '100' })) as CustodyWithdrawal;
  const attempt = randomUUID(), tx = `fixture:${'b'.repeat(64)}`;
  source.prepareWithdrawalFixture(w.id, attempt, tx, '50');
  const listing = source.execute(envelope(originalId, seller, { type: 'list', cardHash: hash, sequence: 0, priceAtomicXmr: '10000', feeBps: 250 })) as CustodyListing;
  const order = source.execute(envelope(originalId, buyer, { type: 'reserve', listingId: listing.id, priceAtomicXmr: listing.priceAtomicXmr, feeBps: 250 })) as CustodyOrder;
  const oldAudit = source.audit(); source.close();
  const file = join(dir, 'testnet', 'custody-sandbox.sqlite');
  const imported = await importSqliteCustody(file, f.admin);
  assert.equal(imported.custodianId, originalId);
  assert.deepEqual(await importSqliteCustody(file, f.admin), imported);
  await assert.rejects(CustodyMarket.open(dir, options), /frozen/);
  const pg = await PostgresCustody.open(f.runtime, options);
  try {
    assert.deepEqual(await pg.audit(), oldAudit);
    assert.equal((await pg.execute(credit) as CustodyActivity).availableAtomicXmr, '1000000000000'); // Original receipt, not current balance.
    const activity = await pg.execute(envelope(originalId, buyer, { type: 'activity', offset: 0 })) as CustodyActivity;
    assert.equal(activity.withdrawals[0].attemptId, attempt); assert.equal(activity.withdrawals[0].transactionId, tx);
    assert.equal(activity.orders[0].id, order.id);
    await pg.execute(envelope(originalId, buyer, { type: 'cancel-reservation', orderId: order.id }));
    assert.equal((await pg.execute(envelope(originalId, buyer, { type: 'activity', offset: 0 })) as CustodyActivity).purchaseReservedAtomicXmr, '0');
    await assert.rejects(pg.execute(envelope(originalId, buyer, { type: 'cancel-withdrawal', withdrawalId: w.id })), /unprepared/);
  } finally { await pg.close(); }
  const sqlite = new DatabaseSync(file); try { sqlite.prepare("UPDATE accounts SET balance='1' WHERE id=?").run(`customer:${buyer.actor}:available`); } finally { sqlite.close(); }
  await assert.rejects(importSqliteCustody(file, f.admin), /source changed/);
});

test('PostgreSQL HTTP custody integration, safe storage binding and backup refusal use the real runtime role', { timeout: 45000 }, async t => {
  const f = await fixture(t), dir = await mkdtemp(join(tmpdir(), 'custody-pg-http-'));
  const urlFile = join(dir, 'pg-url'); await writeFile(urlFile, 'postgresql://custody_runtime@127.0.0.1/custody_test', { mode: 0o600 });
  const env = { ...process.env, HOST: '127.0.0.1', PORT: '0', NODE_ENV: 'production', WIKIDECK_DATA_DIR: dir,
    WIKIDECK_CUSTODY_MODE: 'sandbox', WIKIDECK_CUSTODY_STORAGE: 'postgres', WIKIDECK_CUSTODY_PG_URL_FILE: urlFile,
    WIKIDECK_CUSTODY_PG_LOCAL_TEST: '1', WIKIDECK_PAYMENT_MODE: 'mock', WIKIDECK_PEERS: '' };
  const child = spawn(process.execPath, ['--experimental-strip-types', 'server/index.ts'], { env, stdio: ['ignore', 'pipe', 'pipe'] });
  const closed = new Promise<void>(resolve => child.once('close', () => resolve()));
  t.after(async () => { if (child.exitCode === null) child.kill('SIGTERM'); await closed; await rm(dir, { recursive: true, force: true }); });
  const base = await new Promise<string>((resolve, reject) => {
    let log = ''; const timer = setTimeout(() => reject(new Error('HTTP startup timeout')), 25000);
    const listener = (chunk: Buffer) => { log += chunk.toString(); const m = /Wikideck RandomX testnet: (http:\/\/127\.0\.0\.1:\d+)/.exec(log); if (m) { clearTimeout(timer); resolve(m[1]); } };
    child.stdout.on('data', listener); child.stderr.on('data', listener);
    child.once('exit', () => { clearTimeout(timer); reject(new Error('HTTP startup failed')); });
  });
  const page = await (await fetch(`${base}/api/custody`)).json(); assert.equal(page.info.custodianId, f.market.id);
  const command = envelope(f.market.id, buyer, { type: 'mock-credit' });
  const response = await fetch(`${base}/api/custody/command`, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify(command) });
  assert.equal(response.status, 200); assert.equal((await response.json()).availableAtomicXmr, '1000000000000');
  assert.match(response.headers.get('cache-control') ?? '', /no-store/);
  assert.equal(JSON.parse(await readFile(join(dir, 'testnet', 'custody-postgres.json'), 'utf8')).custodianId, f.market.id);
  child.kill('SIGTERM'); await closed;
  await assert.rejects(createBackup(dir, join(dir, '..', `refused-${randomUUID()}`)), /coordinated/);
  await assert.rejects(openCustodyStorage(dir, options, { WIKIDECK_CUSTODY_STORAGE: 'sqlite' }), /fallback prohibited/);
});

test('configuration requires secret files, verified TLS, literal backend, and loopback-only plaintext test transport', async t => {
  const dir = await mkdtemp(join(tmpdir(), 'custody-pg-config-')); t.after(() => rm(dir, { recursive: true, force: true }));
  const file = join(dir, 'connection');
  assert.equal(custodyStorage({}), 'sqlite'); assert.throws(() => custodyStorage({ WIKIDECK_CUSTODY_STORAGE: 'other' }));
  await assert.rejects(postgresConfig({}), /configuration/);
  await writeFile(file, 'postgresql://user:secret@database.example/custody');
  const secure = await postgresConfig({ WIKIDECK_CUSTODY_PG_URL_FILE: file }); assert.equal((secure.ssl as { rejectUnauthorized: boolean }).rejectUnauthorized, true);
  await assert.rejects(postgresConfig({ WIKIDECK_CUSTODY_PG_URL_FILE: file, WIKIDECK_CUSTODY_PG_LOCAL_TEST: '1' }), /configuration/);
  await writeFile(file, 'postgresql://user:secret@127.0.0.1/custody?sslmode=disable');
  await assert.rejects(postgresConfig({ WIKIDECK_CUSTODY_PG_URL_FILE: file }), /configuration/);
});

test('PostgreSQL market disable preserves existing claims and signed withdrawal cancellation', async t => {
  const f = await fixture(t); await f.execute(buyer, { type: 'mock-credit' });
  const listing = await f.execute(seller, { type: 'list', cardHash: hash, sequence: 0, priceAtomicXmr: '100', feeBps: 250 }) as CustodyListing;
  const order = await f.execute(buyer, { type: 'reserve', listingId: listing.id, priceAtomicXmr: '100', feeBps: 250 }) as CustodyOrder;
  const disabled = await PostgresCustody.open(f.runtime, { ...options, marketEnabled: false });
  try {
    assert.equal(disabled.info().marketEnabled, false); assert.deepEqual((await disabled.browse()).listings, []);
    await assert.rejects(disabled.execute(envelope(disabled.id, other, { type: 'mock-credit' })), /disabled/);
    await disabled.execute(envelope(disabled.id, buyer, { type: 'cancel-reservation', orderId: order.id }));
    const w = await disabled.execute(envelope(disabled.id, buyer, { type: 'withdraw', destination: 'mock-external', amountAtomicXmr: '1', maxFeeAtomicXmr: '0' })) as CustodyWithdrawal;
    await disabled.execute(envelope(disabled.id, buyer, { type: 'cancel-withdrawal', withdrawalId: w.id }));
    assert.equal((await disabled.audit()).assetsAtomicXmr, '1000000000000');
  } finally { await disabled.close(); }
});

test('full PostgreSQL dump/isolated restore preserves custody identity, receipts and journal backing', async t => {
  const f = await fixture(t); await f.execute(buyer, { type: 'mock-credit' });
  const cmd = envelope(f.market.id, buyer, { type: 'withdraw', destination: 'mock-external', amountAtomicXmr: '500', maxFeeAtomicXmr: '20' });
  const original = await f.market.execute(cmd), audit = await f.market.audit();
  const dir = await mkdtemp(join(tmpdir(), 'custody-pg-restore-')), archive = join(dir, 'custody.dump');
  const run = promisify(execFile), database = `restore_${randomUUID().replaceAll('-', '')}`;
  try {
    // Fixture role and TCP transport are local/trust-only on the isolated tmpfs network. No production credentials.
    await run('pg_dump', ['-h', '127.0.0.1', '-U', 'postgres', '-d', 'custody_test', '-Fc', '-f', archive]);
    await f.admin.query(`CREATE DATABASE "${database}"`);
    await run('pg_restore', ['-h', '127.0.0.1', '-U', 'postgres', '-d', database, '--exit-on-error', archive]);
    const restored = await PostgresCustody.open({ ...f.runtime, database }, options);
    try {
      assert.equal(restored.id, f.market.id); assert.deepEqual(await restored.audit(), audit);
      assert.deepEqual(await restored.execute(cmd), original);
      const balance = await restored.execute(envelope(restored.id, buyer, { type: 'activity', offset: 0 })) as CustodyActivity;
      assert.equal(balance.availableAtomicXmr, '999999999480'); assert.equal(balance.withdrawalReservedAtomicXmr, '520');
    } finally { await restored.close(); }
  } finally {
    await f.admin.query(`DROP DATABASE IF EXISTS "${database}" WITH (FORCE)`);
    await rm(dir, { recursive: true, force: true });
  }
});

test('failed cutover freezes the source, rolls back the entire target, and resumes without duplicate balances', async t => {
  const f = await fixture(t); await f.admin.query('DELETE FROM wk_custody.custody_meta');
  const dir = await mkdtemp(join(tmpdir(), 'custody-pg-failed-import-')); t.after(() => rm(dir, { recursive: true, force: true }));
  const source = await CustodyMarket.open(dir, options);
  source.execute(envelope(source.id, buyer, { type: 'mock-credit' })); source.close();
  const file = join(dir, 'testnet', 'custody-sandbox.sqlite');
  await f.admin.query("CREATE FUNCTION wk_custody.import_failure() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN RAISE EXCEPTION 'injected import failure'; END $$; CREATE TRIGGER fail_import BEFORE INSERT ON wk_custody.receipts FOR EACH ROW EXECUTE FUNCTION wk_custody.import_failure()");
  await assert.rejects(importSqliteCustody(file, f.admin), /injected import/);
  assert.equal((await f.admin.query('SELECT count(*)::integer AS n FROM wk_custody.accounts')).rows[0].n, 0);
  await assert.rejects(CustodyMarket.open(dir, options), /frozen/);
  await f.admin.query('DROP TRIGGER fail_import ON wk_custody.receipts');
  await importSqliteCustody(file, f.admin);
  const restored = await PostgresCustody.open(f.runtime, options);
  try { assert.equal((await restored.audit()).assetsAtomicXmr, '1000000000000'); }
  finally { await restored.close(); }
});

test('business audit rejects metadata corruption even when every journal and total remains balanced', async t => {
  const f = await fixture(t); await f.execute(buyer, { type: 'mock-credit' });
  const l = await f.execute(seller, { type: 'list', cardHash: hash, sequence: 0, priceAtomicXmr: '10000', feeBps: 250 }) as CustodyListing;
  const o = await f.execute(buyer, { type: 'reserve', listingId: l.id, priceAtomicXmr: l.priceAtomicXmr, feeBps: l.feeBps }) as CustodyOrder;
  const w = await f.execute(buyer, { type: 'withdraw', destination: 'mock-external', amountAtomicXmr: '500', maxFeeAtomicXmr: '20' }) as CustodyWithdrawal;
  const probes: [string, unknown[], string][] = [
    ["UPDATE wk_custody.orders SET data=jsonb_set(data,'{buyer}',to_jsonb($1::text)) WHERE id=$2", [other.actor, o.id], 'order-terms-mismatch'],
    ["UPDATE wk_custody.orders SET data=jsonb_set(data,'{sellerAtomicXmr}','\"1\"') WHERE id=$1", [o.id], 'order-allocation-mismatch'],
    ["UPDATE wk_custody.orders SET data=jsonb_set(data,'{grossAtomicXmr}','\"1.0\"') WHERE id=$1", [o.id], 'order-record-invalid'],
    ["UPDATE wk_custody.orders SET status='settled',data=jsonb_set(data,'{status}','\"settled\"') WHERE id=$1", [o.id], 'order-reservation-mismatch'],
    ["DELETE FROM wk_custody.outbox WHERE id=$1", [o.id], 'order-state-mismatch'],
    ["UPDATE wk_custody.outbox SET data='{}' WHERE id=$1", [o.id], 'order-outbox-invalid'],
    ["UPDATE wk_custody.listings SET data=jsonb_set(data,'{feeBps}','500') WHERE id=$1", [l.id], 'order-allocation-mismatch'],
    ["UPDATE wk_custody.withdrawals SET data=jsonb_set(data,'{maxFeeAtomicXmr}','\"21\"') WHERE id=$1", [w.id], 'withdrawal-reservation-mismatch'],
    ["UPDATE wk_custody.withdrawals SET data=jsonb_set(data,'{actor}',to_jsonb($1::text)) WHERE id=$2", [other.actor, w.id], 'withdrawal-identity-mismatch'],
    ["UPDATE wk_custody.outbox SET status='fixture-prepared' WHERE id=$1", [w.id], 'withdrawal-outbox-mismatch'],
    ["INSERT INTO wk_custody.outbox VALUES($1,'withdrawal','blocked','{}')", [randomUUID()], 'orphan-outbox'],
    ["INSERT INTO wk_custody.accounts(id,normal) VALUES($1,'credit')", [`customer:${buyer.actor}:purchase:${randomUUID()}`], 'orphan-purchase-reserve'],
  ];
  const c = await f.admin.connect();
  try {
    for (const [sql, args, code] of probes) {
      await c.query('BEGIN');
      try {
        await c.query(sql, args);
        const totals = await c.query("SELECT sum(CASE WHEN normal='debit' THEN balance ELSE -balance END)::text AS net FROM wk_custody.accounts");
        assert.equal(totals.rows[0].net, '0'); // This would pass the old aggregate-only accounting audit.
        await assert.rejects(auditPostgres(c), (e: unknown) => e instanceof CustodyIntegrityError && e.code === code, code);
      } finally { await c.query('ROLLBACK'); }
    }
  } finally { c.release(); }
  assert.equal((await f.market.audit()).journals, 3);
});

test('integrity incident blocks new mutations across instances/restarts but preserves private reads and original receipts', async t => {
  const f = await fixture(t), second = await PostgresCustody.open(f.runtime, options);
  const credit = envelope(f.market.id, buyer, { type: 'mock-credit' }); await f.market.execute(credit);
  const w = await f.execute(buyer, { type: 'withdraw', destination: 'mock-external', amountAtomicXmr: '100', maxFeeAtomicXmr: '0' }) as CustodyWithdrawal;
  try {
    await f.admin.query("UPDATE wk_custody.withdrawals SET data=jsonb_set(data,'{amountAtomicXmr}','\"101\"') WHERE id=$1", [w.id]);
    await assert.rejects(f.market.audit(), (e: unknown) => e instanceof CustodyIntegrityError && e.code === 'withdrawal-reservation-mismatch');
    const status = f.market.diagnostics(); assert.equal(status.integrityHold, true); assert.equal(status.mutationHold, true);
    const text = JSON.stringify(status); for (const secret of [buyer.actor, w.id, 'postgresql:', 'password']) assert.equal(text.includes(secret), false);
    const marker = await f.admin.query("SELECT value FROM wk_custody.custody_meta WHERE key='integrity_incident_v1'");
    assert.equal(JSON.parse(marker.rows[0].value).code, 'withdrawal-reservation-mismatch');
    await assert.rejects(second.execute(envelope(second.id, other, { type: 'mock-credit' })), /integrity incident/);
    assert.equal((await f.market.execute(credit) as CustodyActivity).availableAtomicXmr, '1000000000000');
    assert.equal((await f.activity()).withdrawalReservedAtomicXmr, '100');
    await assert.rejects(PostgresCustody.open(f.runtime, options), /unresolved-integrity-incident/);
    // Restore this disposable corruption to prove that repair alone does not silently clear the incident.
    await f.admin.query('UPDATE wk_custody.withdrawals SET data=$1 WHERE id=$2', [JSON.stringify(w), w.id]);
    await assert.rejects(f.market.audit(), /unresolved-integrity-incident/);
    assert.equal((await f.activity()).availableAtomicXmr, '999999999900');
    const runtimePool = new Pool(f.runtime);
    try { await assert.rejects(resolveCustodyIncident(runtimePool, randomUUID()), /Schema-owner/); }
    finally { await runtimePool.end(); }
    const resolved = await resolveCustodyIncident(f.admin, randomUUID()); assert.ok(resolved.resolutionId);
    // Previously detecting instances remain latched even after authorized acknowledgement.
    await f.market.audit(); assert.equal(f.market.diagnostics().integrityHold, true);
    await assert.rejects(f.execute(other, { type: 'mock-credit' }), /reconciliation/);
    const verified = await PostgresCustody.open(f.runtime, options);
    try { assert.equal(verified.diagnostics().mutationHold, false); await verified.audit(); }
    finally { await verified.close(); }
  } finally { await second.close(); }
});

test('targeted mutation checks reject corrupt reservations before posting or cancelling their outbox', async t => {
  const f = await fixture(t); await f.execute(buyer, { type: 'mock-credit' });
  const w = await f.execute(buyer, { type: 'withdraw', destination: 'mock-external', amountAtomicXmr: '500', maxFeeAtomicXmr: '20' }) as CustodyWithdrawal;
  await f.admin.query("UPDATE wk_custody.outbox SET data='{}' WHERE id=$1", [w.id]);
  await assert.rejects(f.execute(buyer, { type: 'cancel-withdrawal', withdrawalId: w.id }), /integrity check/);
  assert.equal((await f.admin.query('SELECT status FROM wk_custody.withdrawals WHERE id=$1', [w.id])).rows[0].status, 'reserved');
  assert.equal((await f.admin.query("SELECT count(*)::int AS n FROM wk_custody.journals WHERE kind='withdrawal-cancellation'")).rows[0].n, 0);
  assert.equal((await f.activity()).availableAtomicXmr, '999999999480');
  assert.equal(f.market.diagnostics().integrityHold, true);
  await assert.rejects(resolveCustodyIncident(f.admin, randomUUID()), /integrity check/);
  assert.equal((await f.admin.query("SELECT count(*)::int AS n FROM wk_custody.custody_meta WHERE key='integrity_incident_v1'")).rows[0].n, 1);
});

test('business audit pages past 128 records and catches later corrupted output without dropping claims', async t => {
  const f = await fixture(t); await f.execute(buyer, { type: 'mock-credit' });
  let last: CustodyWithdrawal | undefined;
  for (let i = 0; i < 130; i++) {
    const w = await f.execute(buyer, { type: 'withdraw', destination: 'mock-external', amountAtomicXmr: '1', maxFeeAtomicXmr: '0' }) as CustodyWithdrawal;
    if (!last || last.id < w.id) last = w;
  }
  assert.equal((await f.market.audit()).journals, 131);
  await f.admin.query("UPDATE wk_custody.outbox SET data='{}' WHERE id=$1", [last!.id]);
  await assert.rejects(f.market.audit(), /withdrawal-outbox-invalid/);
  assert.equal((await f.activity()).withdrawalReservedAtomicXmr, '130');
});

test('imported fixture completion and cancellation states reconcile their exact terminal postings', async t => {
  const f = await fixture(t); await f.admin.query('DELETE FROM wk_custody.custody_meta');
  const dir = await mkdtemp(join(tmpdir(), 'custody-terminal-audit-')); t.after(() => rm(dir, { recursive: true, force: true }));
  const source = await CustodyMarket.open(dir, options), id = source.id;
  const run = (who: typeof buyer, action: CustodyAction) => source.execute(envelope(id, who, action));
  run(buyer, { type: 'mock-credit' });
  const l = run(seller, { type: 'list', cardHash: hash, sequence: 0, priceAtomicXmr: '1000', feeBps: 250 }) as CustodyListing;
  const o = run(buyer, { type: 'reserve', listingId: l.id, priceAtomicXmr: l.priceAtomicXmr, feeBps: 250 }) as CustodyOrder;
  source.settleFixture(o.id, `fixture:${'c'.repeat(64)}`);
  const w = run(buyer, { type: 'withdraw', destination: 'mock-external', amountAtomicXmr: '1000', maxFeeAtomicXmr: '50' }) as CustodyWithdrawal;
  source.prepareWithdrawalFixture(w.id, randomUUID(), `fixture:${'d'.repeat(64)}`, '10'); source.confirmWithdrawalFixture(w.id, `fixture:${'d'.repeat(64)}`);
  const cancelled = run(buyer, { type: 'withdraw', destination: 'mock-external', amountAtomicXmr: '100', maxFeeAtomicXmr: '5' }) as CustodyWithdrawal;
  run(buyer, { type: 'cancel-withdrawal', withdrawalId: cancelled.id });
  const expected = source.audit(); source.close();
  await importSqliteCustody(join(dir, 'testnet', 'custody-sandbox.sqlite'), f.admin);
  const pg = await PostgresCustody.open(f.runtime, options);
  try { assert.deepEqual(await pg.audit(), expected); }
  finally { await pg.close(); }
});