Verified incoming transfers against Monero Wallet PRC

tests/postgres/custody.test.ts

@@ -15,9 +15,11 @@ 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, grantMoneroObserverRuntime, CUSTODY_SCHEMA_V1, CUSTODY_SCHEMA_V1_DIGEST,
-  CUSTODY_MIGRATION_V2, CUSTODY_SCHEMA_V2_DIGEST } from '../../server/market/postgres-schema.ts';
+  grantMoneroDepositCreditRuntime, CUSTODY_MIGRATION_V2, CUSTODY_SCHEMA_V2_DIGEST } from '../../server/market/postgres-schema.ts';
 import { persistMoneroObservationScan } from '../../server/market/monero-observation-store.ts';
 import { moneroWalletFingerprint } from '../../server/market/monero-observer.ts';
+import { depositCreditApprovalMessage, postReviewedDepositPendingCredit, prepareDepositCreditReview,
+  reconcileDepositCreditReorgHolds, type DepositCreditPolicy } from '../../server/market/monero-deposit-credit.ts';
 import { createOrRecoverMoneroDepositSubaddress, registerMoneroDepositSubaddress,
   reserveMoneroDepositAllocation } from '../../server/market/monero-deposit-allocation-store.ts';
 import { importSqliteCustody } from '../../server/market/custody-import.ts';
@@ -206,7 +208,7 @@ test('custody schema v1 upgrades additively through v3 without dropping existing
   try {
     await migrateCustody(admin); await migrateCustody(admin);
     const versions = await admin.query('SELECT version FROM wk_custody.schema_versions ORDER BY version');
-    assert.deepEqual(versions.rows.map(row => Number(row.version)), [1, 2, 3, 4]);
+    assert.deepEqual(versions.rows.map(row => Number(row.version)), [1, 2, 3, 4, 5]);
     assert.equal((await admin.query("SELECT value FROM wk_custody.custody_meta WHERE key='v1-preserved-marker'")).rows[0].value,
       'existing-test-data');
     for (const table of ['monero_observer_state', 'monero_transfer_observations', 'monero_observation_events', 'monero_observation_scans']) {
@@ -229,7 +231,7 @@ test('custody schema v2 upgrades explicitly to v3 address-allocation registry',
     await client.query('COMMIT');
     await migrateCustody(admin); await migrateCustody(admin);
     const versions = await admin.query('SELECT version FROM wk_custody.schema_versions ORDER BY version');
-    assert.deepEqual(versions.rows.map(row => Number(row.version)), [1, 2, 3, 4]);
+    assert.deepEqual(versions.rows.map(row => Number(row.version)), [1, 2, 3, 4, 5]);
     assert.equal((await admin.query("SELECT value FROM wk_custody.custody_meta WHERE key='v2-marker'")).rows[0].value, 'preserve');
   } catch (error) { await client.query('ROLLBACK').catch(() => {}); throw error; }
   finally { client.release(); await admin.end(); }
@@ -404,6 +406,110 @@ test('Monero scan persistence is idempotent, reorg-visible and never posts custo
   } finally { await custodyRuntime.end(); await runtime.end(); }
 });
 
+test('reviewed Monero outputs post once to pending liability and reorgs create holds without reversal', async t => {
+  const f = await fixture(t);
+  await f.admin.query(`DO $$ BEGIN
+    IF NOT EXISTS(SELECT 1 FROM pg_roles WHERE rolname='monero_observer') THEN CREATE ROLE monero_observer LOGIN; END IF;
+    IF NOT EXISTS(SELECT 1 FROM pg_roles WHERE rolname='monero_deposit_credit') THEN CREATE ROLE monero_deposit_credit LOGIN; END IF;
+  END $$`);
+  await grantMoneroObserverRuntime(f.admin, 'monero_observer');
+  await grantMoneroDepositCreditRuntime(f.admin, 'monero_deposit_credit');
+  const creditPrivileges = await f.admin.query(`SELECT has_table_privilege('monero_deposit_credit',
+    'wk_custody.monero_deposit_credit_events','SELECT') AS can_read_credit,
+    has_table_privilege('monero_deposit_credit','wk_custody.monero_observation_events','SELECT') AS can_read_events`);
+  assert.equal(creditPrivileges.rows[0].can_read_credit, true);
+  assert.equal(creditPrivileges.rows[0].can_read_events, true);
+  const observer = new Pool({ ...config, user: 'monero_observer' });
+  const creditRuntime = new Pool({ ...config, user: 'monero_deposit_credit' });
+  const secondCreditRuntime = new Pool({ ...config, user: 'monero_deposit_credit' });
+  const custodyRuntime = new Pool(f.runtime);
+  const wallet = '8'.repeat(64), txid = '9'.repeat(64), outputPublicKey = 'a'.repeat(64);
+  const observationId = `${txid}:${outputPublicKey}`, now = Date.now(), scanId = randomUUID();
+  const generationId = randomUUID(), allocationId = randomUUID(), requestId = randomUUID();
+  const observation = { observationId, transactionId: txid, amountAtomicXmr: '9007199254740993', height: 90,
+    address: ADDITIONAL_MARKET_FEE.address, confirmations: 11, timestamp: now, unlockTime: '0',
+    accountIndex: 0, addressIndex: 4, outputPublicKey, globalOutputIndex: 123, locked: false, doubleSpendSeen: false };
+  const page = { network: 'mainnet' as const, walletHeight: 100, fromHeight: 0, throughHeight: 100,
+    primaryAddress: ADDITIONAL_MARKET_FEE.address, incoming: [observation] };
+  const policy: DepositCreditPolicy = { approvedReviewerOwners: [host.actor], minimumConfirmations: 10,
+    maximumObservationAgeMs: 120_000 };
+  try {
+    await f.admin.query(`INSERT INTO wk_custody.monero_deposit_allocations(
+      allocation_id,request_id,custodian_id,market_network,market_chain_id,beneficiary,wallet_generation_id,
+      wallet_fingerprint,monero_network,account_index,address_index,status,monero_address,request_digest,reserved_at,registered_at)
+      VALUES($1,$2,$3,$4,$5,$6,$7,$8,'mainnet',0,4,'registered',$9,$10,$11,$12)`,
+    [allocationId, requestId, f.market.id, NETWORK, CHAIN_ID, buyer.actor, generationId, wallet,
+      ADDITIONAL_MARKET_FEE.address, 'b'.repeat(64), now - 1, now]);
+    await persistMoneroObservationScan(observer, { scanId, walletFingerprint: wallet, expectedScanHeight: 0,
+      scannedAt: now, page });
+    await assert.rejects(custodyRuntime.query('SELECT * FROM wk_custody.monero_transfer_observations'));
+    await assert.rejects(observer.query('SELECT * FROM wk_custody.accounts'));
+    await assert.rejects(creditRuntime.query('SELECT balance FROM wk_custody.accounts'));
+    await assert.rejects(creditRuntime.query('UPDATE wk_custody.monero_transfer_observations SET currently_present=false'));
+    await assert.rejects(creditRuntime.query('DELETE FROM wk_custody.journals'));
+    await assert.rejects(creditRuntime.query('SELECT * FROM wk_custody.entries'));
+    await assert.rejects(creditRuntime.query('UPDATE wk_custody.accounts SET balance=0'));
+
+    await assert.rejects(prepareDepositCreditReview(creditRuntime, wallet, observationId,
+      { ...policy, minimumConfirmations: 12 }, now), /not eligible/);
+    const subject = await prepareDepositCreditReview(creditRuntime, wallet, observationId, policy, now);
+    const nonce = randomUUID(), expiresAt = now + 60_000;
+    const approval = { subject, reviewerOwner: host.actor, reviewerPublicKey: host.publicKey, nonce, expiresAt,
+      signature: host.sign(depositCreditApprovalMessage(subject, host.actor, nonce, expiresAt)) };
+    const unapproved = { ...approval, reviewerOwner: buyer.actor, reviewerPublicKey: buyer.publicKey,
+      signature: buyer.sign(depositCreditApprovalMessage(subject, buyer.actor, nonce, expiresAt)) };
+    await assert.rejects(postReviewedDepositPendingCredit(creditRuntime, unapproved, policy, now), /signature is invalid/);
+    const expiredNonce = randomUUID(), expiredAt = now - 1;
+    await assert.rejects(postReviewedDepositPendingCredit(creditRuntime, { ...approval, nonce: expiredNonce,
+      expiresAt: expiredAt, signature: host.sign(depositCreditApprovalMessage(subject, host.actor, expiredNonce, expiredAt)) },
+    policy, now), /expired or outside/);
+
+    const concurrent = await Promise.all([
+      postReviewedDepositPendingCredit(creditRuntime, approval, policy, now),
+      postReviewedDepositPendingCredit(secondCreditRuntime, approval, policy, now),
+    ]);
+    assert.equal(concurrent.filter(result => !result.replayed).length, 1, 'parallel retries must commit one pending credit');
+    const first = concurrent.find(result => !result.replayed)!;
+    assert.equal(first.amountAtomicXmr, observation.amountAtomicXmr);
+    const replay = await postReviewedDepositPendingCredit(creditRuntime, approval, policy, now + 1);
+    assert.equal(replay.replayed, true); assert.equal(replay.journalId, first.journalId);
+    const lateReplay = await postReviewedDepositPendingCredit(creditRuntime, approval, policy, expiresAt + 1);
+    assert.equal(lateReplay.replayed, true, 'a retry of an already committed review returns its receipt after expiry');
+    await assert.rejects(postReviewedDepositPendingCredit(creditRuntime,
+      { ...approval, subject: { ...subject, amountAtomicXmr: '1' } }, policy, now + 2), /signature/);
+
+    const pendingId = `customer:${buyer.actor}:deposit-pending:${first.creditId}`;
+    const posted = await f.admin.query('SELECT id,normal,balance::text AS balance FROM wk_custody.accounts WHERE id=ANY($1::text[])',
+      [[pendingId, 'treasury:monero:pending', `customer:${buyer.actor}:available`]]);
+    assert.deepEqual(posted.rows.map(row => [row.id, row.normal, row.balance]).sort((a,b) => String(a[0]).localeCompare(String(b[0]))), [
+      [pendingId, 'credit', observation.amountAtomicXmr], ['treasury:monero:pending', 'debit', observation.amountAtomicXmr],
+    ].sort((a,b) => String(a[0]).localeCompare(String(b[0]))), 'pending credit must never enter spendable available balance');
+    assert.equal((await f.admin.query("SELECT count(*)::int AS n FROM wk_custody.journals WHERE kind='monero-deposit-pending-credit'")).rows[0].n, 1);
+    assert.equal((await f.market.audit()).assetsAtomicXmr, observation.amountAtomicXmr);
+
+    const missing = await persistMoneroObservationScan(observer, { scanId: randomUUID(), walletFingerprint: wallet,
+      expectedScanHeight: 100, scannedAt: now + 100,
+      page: { ...page, fromHeight: 90, walletHeight: 101, throughHeight: 101, incoming: [] } });
+    assert.equal(missing.notSeen, 1);
+    assert.equal(await reconcileDepositCreditReorgHolds(creditRuntime, wallet, now + 110), 1);
+    assert.equal(await reconcileDepositCreditReorgHolds(creditRuntime, wallet, now + 111), 0, 'hold replay is idempotent');
+    const reappeared = await persistMoneroObservationScan(observer, { scanId: randomUUID(), walletFingerprint: wallet,
+      expectedScanHeight: 101, scannedAt: now + 200,
+      page: { ...page, fromHeight: 90, walletHeight: 102, throughHeight: 102,
+        incoming: [{ ...observation, confirmations: 13 }] } });
+    assert.equal(reappeared.reappeared, 1);
+    assert.equal(await reconcileDepositCreditReorgHolds(creditRuntime, wallet, now + 210), 1);
+    assert.equal((await f.admin.query('SELECT count(*)::int AS n FROM wk_custody.monero_deposit_reorg_holds')).rows[0].n, 2);
+    assert.equal((await f.admin.query('SELECT count(*)::int AS n FROM wk_custody.journals')).rows[0].n, 1,
+      'reorg/reappearance must not reverse or duplicate the pending credit journal');
+    assert.equal((await f.admin.query('SELECT balance::text FROM wk_custody.accounts WHERE id=$1', [pendingId])).rows[0].balance,
+      observation.amountAtomicXmr);
+    await assert.rejects(creditRuntime.query('UPDATE wk_custody.monero_deposit_credit_events SET amount_atomic_xmr=1'));
+    await assert.rejects(creditRuntime.query('DELETE FROM wk_custody.monero_deposit_reorg_holds'));
+    assert.equal((await f.market.audit()).assetsAtomicXmr, observation.amountAtomicXmr);
+  } finally { await custodyRuntime.end(); await secondCreditRuntime.end(); await creditRuntime.end(); await observer.end(); }
+});
+
 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();
@@ -466,7 +572,8 @@ test('exact amounts above JS safe integer survive NUMERIC, JSON signatures and a
   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.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');