wikideck/server/market/monero-deposit-allocation-store.ts

import { createHash, randomUUID } from 'node:crypto';
import type { Pool, PoolClient } from 'pg';
import { z } from 'zod';
import type { MoneroDepositAddressRpc } from './monero-observer.ts';
import { validateXmrSubaddress } from './xmr.ts';

const uuid = z.string().uuid();
const hex64 = z.string().regex(/^[a-f0-9]{64}$/);
const index = z.number().int().min(0).max(0xffff_ffff).safe();
const reserveInputSchema = z.object({
  requestId: uuid,
  custodianId: uuid,
  marketNetwork: z.enum(['wikideck-randomx-mainnet-1', 'wikideck-randomx-testnet-6']),
  marketChainId: hex64,
  beneficiary: hex64,
  walletGenerationId: uuid,
  walletFingerprint: hex64,
  accountIndex: index,
  reservedAt: z.number().int().nonnegative().safe(),
}).strict();
const registerInputSchema = z.object({
  allocationId: uuid,
  walletGenerationId: uuid,
  moneroAddress: z.string().length(95),
  registeredAt: z.number().int().nonnegative().safe(),
}).strict();

export type MoneroDepositAllocation = {
  allocationId: string;
  requestId: string;
  custodianId: string;
  marketNetwork: 'wikideck-randomx-mainnet-1' | 'wikideck-randomx-testnet-6';
  marketChainId: string;
  beneficiary: string;
  walletGenerationId: string;
  walletFingerprint: string;
  moneroNetwork: 'mainnet';
  accountIndex: number;
  addressIndex: number;
  status: 'reserved' | 'registered';
  moneroAddress: string | null;
  reservedAt: number;
  registeredAt: number | null;
};

type AllocationRow = {
  allocation_id: string;
  request_id: string;
  custodian_id: string;
  market_network: 'wikideck-randomx-mainnet-1' | 'wikideck-randomx-testnet-6';
  market_chain_id: string;
  beneficiary: string;
  wallet_generation_id: string;
  wallet_fingerprint: string;
  monero_network: 'mainnet';
  account_index: string;
  address_index: string;
  status: 'reserved' | 'registered';
  monero_address: string | null;
  reserved_at: string;
  registered_at: string | null;
  request_digest: string;
};

function toAllocation(row: AllocationRow): MoneroDepositAllocation {
  return {
    allocationId: row.allocation_id,
    requestId: row.request_id,
    custodianId: row.custodian_id,
    marketNetwork: row.market_network,
    marketChainId: row.market_chain_id,
    beneficiary: row.beneficiary,
    walletGenerationId: row.wallet_generation_id,
    walletFingerprint: row.wallet_fingerprint,
    moneroNetwork: row.monero_network,
    accountIndex: Number(row.account_index),
    addressIndex: Number(row.address_index),
    status: row.status,
    moneroAddress: row.monero_address,
    reservedAt: Number(row.reserved_at),
    registeredAt: row.registered_at === null ? null : Number(row.registered_at),
  };
}

function requestDigest(input: z.infer<typeof reserveInputSchema>): string {
  return createHash('sha256').update(JSON.stringify([
    'wikideck-monero-deposit-allocation-v1', input.custodianId, input.marketNetwork,
    input.marketChainId, input.beneficiary, input.walletGenerationId,
    input.walletFingerprint, input.accountIndex,
  ])).digest('hex');
}

async function finishTransaction(client: PoolClient, commit: boolean): Promise<void> {
  if (commit) await client.query('COMMIT');
  else await client.query('ROLLBACK').catch(() => {});
}

/**
 * Durably reserve a never-reused wallet subaddress index and bind it to a beneficiary before
 * any wallet RPC address-generation side effect. This module is deliberately not wired to
 * startup, HTTP, a wallet RPC, or financial posting. A reserved row is never reclaimed.
 */
export async function reserveMoneroDepositAllocation(
  pool: Pool,
  input: unknown,
): Promise<{ allocation: MoneroDepositAllocation; replayed: boolean }> {
  const parsed = reserveInputSchema.parse(input);
  const digest = requestDigest(parsed);
  const client = await pool.connect();
  try {
    await client.query('BEGIN');
    await client.query("SET LOCAL statement_timeout='10s'; SET LOCAL lock_timeout='3s'; SET LOCAL idle_in_transaction_session_timeout='10s'");
    // Per-wallet/account serialization makes MAX(index)+1 safe across independent processes.
    await client.query('SELECT pg_advisory_xact_lock(hashtextextended($1,0))',
      [`wikideck-monero-deposit-allocation-v1:${parsed.walletGenerationId}:${parsed.accountIndex}`]);

    const byRequest = await client.query<AllocationRow>(`SELECT * FROM wk_custody.monero_deposit_allocations
      WHERE request_id=$1 FOR UPDATE`, [parsed.requestId]);
    if (byRequest.rowCount) {
      if (byRequest.rows[0].request_digest !== digest) throw new Error('Deposit allocation request ID was reused with different terms');
      await finishTransaction(client, true);
      return { allocation: toAllocation(byRequest.rows[0]), replayed: true };
    }

    const byBeneficiary = await client.query<AllocationRow>(`SELECT * FROM wk_custody.monero_deposit_allocations
      WHERE custodian_id=$1 AND market_network=$2 AND market_chain_id=$3 AND beneficiary=$4
        AND wallet_generation_id=$5 FOR UPDATE`, [parsed.custodianId, parsed.marketNetwork, parsed.marketChainId,
      parsed.beneficiary, parsed.walletGenerationId]);
    if (byBeneficiary.rowCount) {
      if (byBeneficiary.rows[0].request_digest !== digest) throw new Error('Beneficiary already has a different deposit allocation for this wallet generation');
      await finishTransaction(client, true);
      return { allocation: toAllocation(byBeneficiary.rows[0]), replayed: true };
    }

    const next = await client.query<{ next_index: string }>(`SELECT (COALESCE(MAX(address_index),-1)+1)::text AS next_index
      FROM wk_custody.monero_deposit_allocations WHERE wallet_generation_id=$1 AND account_index=$2`,
    [parsed.walletGenerationId, parsed.accountIndex]);
    const addressIndex = BigInt(next.rows[0].next_index);
    if (addressIndex > 0xffff_ffffn) throw new Error('Monero subaddress index space is exhausted for this account');

    const inserted = await client.query<AllocationRow>(`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,request_digest,reserved_at)
      VALUES($1,$2,$3,$4,$5,$6,$7,$8,'mainnet',$9,$10,'reserved',$11,$12) RETURNING *`,
    [randomUUID(), parsed.requestId, parsed.custodianId, parsed.marketNetwork, parsed.marketChainId,
      parsed.beneficiary, parsed.walletGenerationId, parsed.walletFingerprint, parsed.accountIndex,
      addressIndex.toString(), digest, parsed.reservedAt]);
    await finishTransaction(client, true);
    return { allocation: toAllocation(inserted.rows[0]), replayed: false };
  } catch (error) {
    await finishTransaction(client, false);
    if (typeof error === 'object' && error !== null && 'code' in error && error.code === '23505') {
      throw new Error('Deposit allocation conflicts with an existing beneficiary, wallet index, or address');
    }
    throw error;
  } finally {
    client.release();
  }
}

/** Register the address returned by a separately managed wallet operation exactly once. */
export async function registerMoneroDepositSubaddress(
  pool: Pool,
  input: unknown,
): Promise<{ allocation: MoneroDepositAllocation; replayed: boolean }> {
  const parsed = registerInputSchema.parse(input);
  const address = await validateXmrSubaddress(parsed.moneroAddress);
  const client = await pool.connect();
  try {
    await client.query('BEGIN');
    await client.query("SET LOCAL statement_timeout='10s'; SET LOCAL lock_timeout='3s'; SET LOCAL idle_in_transaction_session_timeout='10s'");
    const found = await client.query<AllocationRow>(`SELECT * FROM wk_custody.monero_deposit_allocations
      WHERE allocation_id=$1 AND wallet_generation_id=$2 FOR UPDATE`, [parsed.allocationId, parsed.walletGenerationId]);
    if (!found.rowCount) throw new Error('Deposit allocation reservation was not found for this wallet generation');
    const row = found.rows[0];
    if (row.status === 'registered') {
      if (row.monero_address !== address) throw new Error('Registered deposit address cannot be replaced');
      await finishTransaction(client, true);
      return { allocation: toAllocation(row), replayed: true };
    }
    if (parsed.registeredAt < Number(row.reserved_at)) throw new Error('Deposit address registration predates its durable reservation');
    const registered = await client.query<AllocationRow>(`UPDATE wk_custody.monero_deposit_allocations
      SET status='registered',monero_address=$3,registered_at=$4
      WHERE allocation_id=$1 AND wallet_generation_id=$2 RETURNING *`,
    [parsed.allocationId, parsed.walletGenerationId, address, parsed.registeredAt]);
    await finishTransaction(client, true);
    return { allocation: toAllocation(registered.rows[0]), replayed: false };
  } catch (error) {
    await finishTransaction(client, false);
    if (typeof error === 'object' && error !== null && 'code' in error && error.code === '23505') {
      throw new Error('Monero deposit address is already registered to an allocation');
    }
    throw error;
  } finally {
    client.release();
  }
}

/**
 * Create or recover the wallet subaddress for a previously committed reservation. A stable
 * wallet label makes a lost create_address response recoverable via get_address at the exact
 * reserved index. The wallet RPC must be dedicated to this allocator: external address
 * creation can consume indexes and is treated as a hard mismatch, never silently reassigned.
 */
export async function createOrRecoverMoneroDepositSubaddress(
  pool: Pool,
  rpc: MoneroDepositAddressRpc,
  allocationId: string,
  walletGenerationId: string,
  registeredAt: number,
): Promise<{ allocation: MoneroDepositAllocation; replayed: boolean }> {
  const parsedAllocationId = uuid.parse(allocationId);
  const parsedGenerationId = uuid.parse(walletGenerationId);
  if (!Number.isSafeInteger(registeredAt) || registeredAt < 0) throw new Error('Invalid deposit address registration time');

  const client = await pool.connect();
  let address: string | undefined;
  let label: string | undefined;
  let lockKey: string | undefined;
  let locked = false;
  try {
    const initial = await client.query<AllocationRow>(`SELECT * FROM wk_custody.monero_deposit_allocations
      WHERE allocation_id=$1 AND wallet_generation_id=$2`, [parsedAllocationId, parsedGenerationId]);
    if (!initial.rowCount) throw new Error('Deposit allocation reservation was not found for this wallet generation');
    if (initial.rows[0].status === 'registered') {
      return { allocation: toAllocation(initial.rows[0]), replayed: true };
    }
    const row = initial.rows[0];
    lockKey = `wikideck-monero-deposit-address-v1:${row.wallet_generation_id}:${row.account_index}`;
    await client.query('SELECT pg_advisory_lock(hashtextextended($1,0))', [lockKey]);
    locked = true;

    // Re-read after obtaining the per-wallet lock in case another process finished first.
    const current = await client.query<AllocationRow>(`SELECT * FROM wk_custody.monero_deposit_allocations
      WHERE allocation_id=$1 AND wallet_generation_id=$2`, [parsedAllocationId, parsedGenerationId]);
    if (!current.rowCount) throw new Error('Deposit allocation reservation disappeared');
    if (current.rows[0].status === 'registered') {
      return { allocation: toAllocation(current.rows[0]), replayed: true };
    }
    const allocation = current.rows[0];
    const expectedLabel = `wikideck-deposit-v1:${allocation.allocation_id}`;
    const expectedIndex = Number(allocation.address_index);
    const existing = await rpc.getSubaddress(Number(allocation.account_index), expectedIndex);
    if (existing) {
      if (existing.label !== expectedLabel) throw new Error('Reserved Monero subaddress index is occupied by a different wallet label');
      address = await validateXmrSubaddress(existing.address);
      label = existing.label;
    } else {
      const created = await rpc.createSubaddress(Number(allocation.account_index), expectedLabel);
      if (created.addressIndex !== expectedIndex) {
        throw new Error('Monero wallet created an unexpected subaddress index; allocation requires operator reconciliation');
      }
      const verified = await rpc.getSubaddress(Number(allocation.account_index), expectedIndex);
      if (!verified || verified.label !== expectedLabel || verified.address !== created.address) {
        throw new Error('Monero wallet subaddress could not be verified at its reserved index');
      }
      address = await validateXmrSubaddress(verified.address);
      label = verified.label;
    }
  } finally {
    if (locked && lockKey) await client.query('SELECT pg_advisory_unlock(hashtextextended($1,0))', [lockKey]).catch(() => {});
    client.release();
  }
  if (!address || !label) throw new Error('Monero deposit address allocation was not verified');
  return registerMoneroDepositSubaddress(pool, { allocationId: parsedAllocationId,
    walletGenerationId: parsedGenerationId, moneroAddress: address, registeredAt });
}