Design a Digital Wallet

BEGIN;

-- Serialization point: a retry conflicts here and we return the stored result.
INSERT INTO transfers (transfer_id, idempotency_key, from_wallet_id, to_wallet_id, amount, state)
VALUES (:tid, :ikey, :from, :to, :amount, 'completed')
ON CONFLICT (idempotency_key) DO NOTHING;
-- If 0 rows inserted: SELECT * FROM transfers WHERE idempotency_key = :ikey; return it. Done.

-- Lock both wallets in deterministic order to avoid deadlock.
SELECT wallet_id, cached_balance FROM wallets
WHERE wallet_id IN (:from, :to)
ORDER BY wallet_id
FOR UPDATE;

-- Insufficient funds aborts everything, including the transfer row.
UPDATE wallets SET cached_balance = cached_balance - :amount
WHERE wallet_id = :from AND cached_balance >= :amount;
-- If 0 rows updated: ROLLBACK and fail with INSUFFICIENT_FUNDS.

UPDATE wallets SET cached_balance = cached_balance + :amount
WHERE wallet_id = :to;

INSERT INTO ledger_entries (transfer_id, wallet_id, direction, amount, balance_after)
VALUES
  (:tid, :from, 'debit',  :amount, (SELECT cached_balance FROM wallets WHERE wallet_id = :from)),
  (:tid, :to,   'credit', :amount, (SELECT cached_balance FROM wallets WHERE wallet_id = :to));

COMMIT;

import { pool } from "./db";

export async function transfer(ikey: string, from: string, to: string, amount: bigint) {
  if (amount <= 0n) throw new Error("INVALID_AMOUNT");
  const client = await pool.connect();
  try {
    await client.query("BEGIN");
    const ins = await client.query(
      "INSERT INTO transfers (idempotency_key, from_wallet_id, to_wallet_id, amount, state) " +
      "VALUES ($1, $2, $3, $4, 'completed') ON CONFLICT (idempotency_key) DO NOTHING RETURNING transfer_id",
      [ikey, from, to, amount]
    );
    if (ins.rowCount === 0) {
      await client.query("ROLLBACK");
      const prev = await pool.query("SELECT * FROM transfers WHERE idempotency_key = $1", [ikey]);
      return { replayed: true, ...prev.rows[0] }; // exact original outcome
    }
    const tid = ins.rows[0].transfer_id;
    const [a, b] = [from, to].sort(); // deterministic lock order
    await client.query("SELECT 1 FROM wallets WHERE wallet_id = ANY($1) ORDER BY wallet_id FOR UPDATE", [[a, b]]);
    const deb = await client.query(
      "UPDATE wallets SET cached_balance = cached_balance - $1 WHERE wallet_id = $2 AND cached_balance >= $1",
      [amount, from]
    );
    if (deb.rowCount === 0) { await client.query("ROLLBACK"); throw new Error("INSUFFICIENT_FUNDS"); }
    await client.query("UPDATE wallets SET cached_balance = cached_balance + $1 WHERE wallet_id = $2", [amount, to]);
    await client.query(
      "INSERT INTO ledger_entries (transfer_id, wallet_id, direction, amount, balance_after) VALUES " +
      "($1, $2, 'debit', $3, (SELECT cached_balance FROM wallets WHERE wallet_id = $2)), " +
      "($1, $4, 'credit', $3, (SELECT cached_balance FROM wallets WHERE wallet_id = $4))",
      [tid, from, amount, to]
    );
    await client.query("COMMIT");
    return { replayed: false, transfer_id: tid };
  } catch (e) {
    await client.query("ROLLBACK").catch(() => {});
    throw e;
  } finally {
    client.release();
  }
}

-- 1. Global conservation law: all debits equal all credits.
SELECT
  COALESCE(SUM(amount) FILTER (WHERE direction = 'debit'), 0)  AS total_debits,
  COALESCE(SUM(amount) FILTER (WHERE direction = 'credit'), 0) AS total_credits
FROM ledger_entries;
-- Assert total_debits = total_credits; anything else is corruption.

-- 2. Every cached balance equals its replayed ledger sum.
SELECT w.wallet_id, w.cached_balance, l.ledger_balance
FROM wallets w
JOIN LATERAL (
  SELECT COALESCE(SUM(CASE direction WHEN 'credit' THEN amount ELSE -amount END), 0)
         AS ledger_balance
  FROM ledger_entries e WHERE e.wallet_id = w.wallet_id
) l ON true
WHERE w.cached_balance <> l.ledger_balance;
-- Assert zero rows; any row is a wallet to freeze and investigate.