LLD Series: Digital Wallet (Low-Level Design)

 +-----------------+  1     *  +------------------+
 |     Wallet      |<----------|   LedgerEntry    |  (append-only; 1 DEBIT + 1 CREDIT per txn)
 | id, ownerId     |           | txnId, type,     |
 | type USER/SYSTEM|           | amount, balAfter |
 | balance + lock  |           +------------------+
 +--------^--------+                    ^
          | uses                         | appends
 +--------+--------------------------------------------+
 |                 WalletService                       |---> Ledger
 | topUp() / transfer()  -> execute(idempotencyKey,..) |
 | idempotency: Map<key, Future<Txn>>                  |
 +----+-------------------+----------------------+-----+
      | rules (Strategy)  | listeners (Observer) | produces
      v                   v                      v
 <<iface>> LimitRule   <<iface>> TransactionListener   WalletTransaction
   PerTransactionLimit    SmsNotificationListener      id, key, type, from,to,
   DailySpendLimit        (Email/Push/Audit...)        amount, status, reason
   MaxBalanceLimit

// Enums.java
enum WalletType { USER, SYSTEM }              // SYSTEM wallets (bank settlement, fees) may go negative
enum WalletStatus { ACTIVE, FROZEN }
enum EntryType { DEBIT, CREDIT }
enum TxnType { TOP_UP, TRANSFER }
enum TxnStatus { SUCCESS, FAILED }

// Money.java
import java.math.BigDecimal;

public final class Money {
    private Money() {}
    /** "250.50" -> 25050 paise */
    public static long of(String major) { return new BigDecimal(major).movePointRight(2).longValueExact(); }
    public static String fmt(long minor) { return "Rs " + BigDecimal.valueOf(minor, 2).toPlainString(); }
}

// WalletException.java
public class WalletException extends RuntimeException {
    public WalletException(String m) { super(m); }
}

// IdempotencyConflictException.java
/** Same idempotency key re-used with a different request payload: a client bug, reject loudly. */
public class IdempotencyConflictException extends WalletException {
    public IdempotencyConflictException(String key) { super("Idempotency key reused with different request: " + key); }
}

// Wallet.java
import java.time.LocalDate;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Wallet owns its balance and its lock. The balance is only mutated while the lock is held
 * (the service acquires locks; mutators here are package-private to prevent misuse).
 */
public final class Wallet {
    private final String id, ownerId;
    private final WalletType type;
    private final ReentrantLock lock = new ReentrantLock();
    private long balance;                  // guarded by lock
    private WalletStatus status = WalletStatus.ACTIVE;   // guarded by lock
    private LocalDate spendDate;           // guarded by lock
    private long spentOnDate;              // guarded by lock

    public Wallet(String id, String ownerId, WalletType type) { this.id = id; this.ownerId = ownerId; this.type = type; }

    public String id() { return id; }
    public String ownerId() { return ownerId; }
    public WalletType type() { return type; }
    ReentrantLock lock() { return lock; }

    // --- caller MUST hold lock() for the ones below ---
    long balanceLocked() { return balance; }
    WalletStatus statusLocked() { return status; }
    void applyDebit(long amt) { balance -= amt; }
    void applyCredit(long amt) { balance += amt; }
    void freeze() { status = WalletStatus.FROZEN; }
    long spentOn(LocalDate d) { return d.equals(spendDate) ? spentOnDate : 0; }
    void recordSpend(LocalDate d, long amt) {
        if (!d.equals(spendDate)) { spendDate = d; spentOnDate = 0; }
        spentOnDate += amt;
    }

    /** Safe read for UI/tests. */
    public long balance() { lock.lock(); try { return balance; } finally { lock.unlock(); } }
}

// LedgerEntry.java
import java.time.Instant;

/** Immutable, append-only. A transfer always produces exactly one DEBIT and one CREDIT of equal amount. */
public final class LedgerEntry {
    public final long entryId; public final String txnId, walletId; public final EntryType type;
    public final long amount, balanceAfter; public final Instant at;

    public LedgerEntry(long entryId, String txnId, String walletId, EntryType type, long amount, long balanceAfter, Instant at) {
        this.entryId = entryId; this.txnId = txnId; this.walletId = walletId; this.type = type;
        this.amount = amount; this.balanceAfter = balanceAfter; this.at = at;
    }
    @Override public String toString() {
        return String.format("#%d %s %-6s %s %s balanceAfter=%s", entryId, txnId, type, walletId, Money.fmt(amount), Money.fmt(balanceAfter));
    }
}

// Ledger.java
import java.time.Clock;
import java.time.Instant;
import java.util.*;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;

public final class Ledger {
    private final List<LedgerEntry> entries = new CopyOnWriteArrayList<>();
    private final AtomicLong seq = new AtomicLong();
    private final Clock clock;
    public Ledger(Clock clock) { this.clock = clock; }

    /** Called while holding both wallet locks, so entries of one txn are written together. */
    void append(String txnId, String walletId, EntryType type, long amount, long balanceAfter) {
        entries.add(new LedgerEntry(seq.incrementAndGet(), txnId, walletId, type, amount, balanceAfter, Instant.now(clock)));
    }
    public List<LedgerEntry> forWallet(String walletId) {
        return entries.stream().filter(e -> e.walletId.equals(walletId)).collect(Collectors.toList());
    }
    /** Invariant: total debits == total credits, and per txn debit == credit. */
    public boolean isBalanced() {
        Map<String, Long> perTxn = new HashMap<>();
        for (LedgerEntry e : entries) perTxn.merge(e.txnId, e.type == EntryType.DEBIT ? e.amount : -e.amount, Long::sum);
        return perTxn.values().stream().allMatch(v -> v == 0);
    }
    public int size() { return entries.size(); }
}

// WalletTransaction.java
import java.time.Instant;

public final class WalletTransaction {
    public final String id, idempotencyKey, fromWalletId, toWalletId, failureReason;
    public final TxnType type; public final long amount; public final TxnStatus status; public final Instant at;

    WalletTransaction(String id, String key, TxnType type, String from, String to, long amount,
                      TxnStatus status, String failureReason, Instant at) {
        this.id = id; this.idempotencyKey = key; this.type = type; this.fromWalletId = from; this.toWalletId = to;
        this.amount = amount; this.status = status; this.failureReason = failureReason; this.at = at;
    }
    @Override public String toString() {
        return id + " " + type + " " + fromWalletId + "->" + toWalletId + " " + Money.fmt(amount) + " " + status
                + (failureReason == null ? "" : " [" + failureReason + "]");
    }
}

// LimitRule.java
import java.time.Clock;

/** Strategy / rule chain element. Evaluated under the wallet locks, so reads are consistent. */
public interface LimitRule {
    void check(Wallet from, Wallet to, long amount, Clock clock);   // throws WalletException to veto
}

// PerTransactionLimit.java
import java.time.Clock;

public final class PerTransactionLimit implements LimitRule {
    private final long max;
    public PerTransactionLimit(long max) { this.max = max; }
    @Override public void check(Wallet from, Wallet to, long amount, Clock c) {
        if (from.type() == WalletType.SYSTEM) return;       // limits protect customers' spending, not bank top-ups
        if (amount > max) throw new WalletException("Per-transaction limit " + Money.fmt(max) + " exceeded");
    }
}

// DailySpendLimit.java
import java.time.Clock;
import java.time.LocalDate;

public final class DailySpendLimit implements LimitRule {
    private final long max;
    public DailySpendLimit(long max) { this.max = max; }
    @Override public void check(Wallet from, Wallet to, long amount, Clock c) {
        if (from.type() == WalletType.SYSTEM) return;
        long already = from.spentOn(LocalDate.now(c));
        if (already + amount > max) throw new WalletException("Daily limit " + Money.fmt(max) + " exceeded (spent " + Money.fmt(already) + ")");
    }
}

// MaxBalanceLimit.java
import java.time.Clock;

/** Receiver-side rule: e.g. a minimum-KYC wallet may not hold more than X. */
public final class MaxBalanceLimit implements LimitRule {
    private final long max;
    public MaxBalanceLimit(long max) { this.max = max; }
    @Override public void check(Wallet from, Wallet to, long amount, Clock c) {
        if (to.type() == WalletType.USER && to.balanceLocked() + amount > max)
            throw new WalletException("Receiver would exceed max balance " + Money.fmt(max));
    }
}

// TransactionListener.java
/** Observer. Invoked AFTER locks are released and AFTER the txn is durable; failures are isolated. */
public interface TransactionListener {
    void onTransaction(WalletTransaction txn);
}

// SmsNotificationListener.java
public final class SmsNotificationListener implements TransactionListener {
    @Override public void onTransaction(WalletTransaction t) {
        if (t.status == TxnStatus.SUCCESS) System.out.println("    [SMS] " + t.type + " of " + Money.fmt(t.amount) + " " + t.fromWalletId + " -> " + t.toWalletId + " ok (" + t.id + ")");
        else System.out.println("    [SMS] " + t.type + " of " + Money.fmt(t.amount) + " failed: " + t.failureReason);
    }
}

// WalletService.java
import java.time.Clock;
import java.time.Instant;
import java.time.LocalDate;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

/**
 * Concurrency design:
 *  - lock per wallet (ReentrantLock), taken in ascending wallet-id order => no deadlock for A->B vs B->A.
 *  - idempotency: putIfAbsent of a CompletableFuture, so exactly one thread executes, duplicates wait for / reuse its result.
 *  - listeners are called after locks are released (never run foreign code under a lock).
 */
public final class WalletService {
    private static final class IdemEntry {
        final String fingerprint; final CompletableFuture<WalletTransaction> result = new CompletableFuture<>();
        IdemEntry(String fp) { this.fingerprint = fp; }
    }

    private final Map<String, Wallet> wallets = new ConcurrentHashMap<>();
    private final Map<String, IdemEntry> idempotency = new ConcurrentHashMap<>();
    private final List<TransactionListener> listeners = new CopyOnWriteArrayList<>();
    private final List<LimitRule> rules;
    private final Ledger ledger;
    private final Clock clock;
    private final AtomicLong txnSeq = new AtomicLong();
    private final String systemWalletId;

    public WalletService(List<LimitRule> rules, Ledger ledger, Clock clock) {
        this.rules = List.copyOf(rules); this.ledger = ledger; this.clock = clock;
        this.systemWalletId = "SYS-SETTLEMENT";
        wallets.put(systemWalletId, new Wallet(systemWalletId, "system", WalletType.SYSTEM));
    }

    public void addListener(TransactionListener l) { listeners.add(l); }

    public Wallet createWallet(String walletId, String ownerId) {
        Wallet w = new Wallet(walletId, ownerId, WalletType.USER);
        if (wallets.putIfAbsent(walletId, w) != null) throw new WalletException("Wallet exists: " + walletId);
        return w;
    }
    public Wallet wallet(String id) {
        Wallet w = wallets.get(id);
        if (w == null) throw new WalletException("Unknown wallet " + id);
        return w;
    }

    /** Money enters the system from the settlement account (e.g. bank/UPI collect succeeded). */
    public WalletTransaction topUp(String idempotencyKey, String walletId, long amount) {
        return execute(idempotencyKey, TxnType.TOP_UP, systemWalletId, walletId, amount);
    }

    public WalletTransaction transfer(String idempotencyKey, String fromWalletId, String toWalletId, long amount) {
        return execute(idempotencyKey, TxnType.TRANSFER, fromWalletId, toWalletId, amount);
    }

    private WalletTransaction execute(String key, TxnType type, String fromId, String toId, long amount) {
        if (key == null || key.isBlank()) throw new WalletException("Idempotency key required");
        if (amount <= 0) throw new WalletException("Amount must be positive");
        if (fromId.equals(toId)) throw new WalletException("Cannot transfer to the same wallet");
        Wallet from = wallet(fromId), to = wallet(toId);

        IdemEntry mine = new IdemEntry(type + "|" + fromId + "|" + toId + "|" + amount);
        IdemEntry existing = idempotency.putIfAbsent(key, mine);
        if (existing != null) {                                   // duplicate request
            if (!existing.fingerprint.equals(mine.fingerprint)) throw new IdempotencyConflictException(key);
            return existing.result.join();                        // same outcome, no second debit
        }

        WalletTransaction txn;
        try {
            txn = process(key, type, from, to, amount);
        } catch (RuntimeException e) {                            // never leave a duplicate waiting forever
            idempotency.remove(key);
            mine.result.completeExceptionally(e);
            throw e;
        }
        mine.result.complete(txn);
        for (TransactionListener l : listeners) {
            try { l.onTransaction(txn); } catch (RuntimeException ignored) { /* a bad listener must not break payments */ }
        }
        return txn;
    }

    private WalletTransaction process(String key, TxnType type, Wallet from, Wallet to, long amount) {
        String txnId = "TXN-" + txnSeq.incrementAndGet();
        Wallet first = from.id().compareTo(to.id()) < 0 ? from : to;      // global lock order
        Wallet second = first == from ? to : from;
        first.lock().lock();
        second.lock().lock();
        try {
            try {
                if (from.statusLocked() != WalletStatus.ACTIVE || to.statusLocked() != WalletStatus.ACTIVE)
                    throw new WalletException("Wallet is frozen");
                if (from.type() == WalletType.USER && from.balanceLocked() < amount)
                    throw new WalletException("Insufficient balance");
                for (LimitRule r : rules) r.check(from, to, amount, clock);
            } catch (WalletException rejected) {
                return new WalletTransaction(txnId, key, type, from.id(), to.id(), amount, TxnStatus.FAILED, rejected.getMessage(), Instant.now(clock));
            }
            from.applyDebit(amount);
            to.applyCredit(amount);
            if (from.type() == WalletType.USER) from.recordSpend(LocalDate.now(clock), amount);
            ledger.append(txnId, from.id(), EntryType.DEBIT, amount, from.balanceLocked());
            ledger.append(txnId, to.id(), EntryType.CREDIT, amount, to.balanceLocked());
            return new WalletTransaction(txnId, key, type, from.id(), to.id(), amount, TxnStatus.SUCCESS, null, Instant.now(clock));
        } finally {
            second.lock().unlock();
            first.lock().unlock();
        }
    }
}

// Main.java
import java.time.Clock;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class Main {
    public static void main(String[] args) throws Exception {
        Clock clock = Clock.systemUTC();
        Ledger ledger = new Ledger(clock);
        WalletService svc = new WalletService(List.of(
                new PerTransactionLimit(Money.of("25000")),
                new DailySpendLimit(Money.of("40000")),
                new MaxBalanceLimit(Money.of("200000"))), ledger, clock);
        svc.addListener(new SmsNotificationListener());
        svc.createWallet("W-ALICE", "alice");
        svc.createWallet("W-BOB", "bob");

        System.out.println("== basic flow ==");
        svc.topUp("topup-1", "W-ALICE", Money.of("50000"));
        System.out.println(svc.transfer("pay-1", "W-ALICE", "W-BOB", Money.of("1200.50")));
        System.out.println("alice=" + Money.fmt(svc.wallet("W-ALICE").balance()) + " bob=" + Money.fmt(svc.wallet("W-BOB").balance()));

        System.out.println("== idempotent retry (same key) ==");
        WalletTransaction a = svc.transfer("pay-1", "W-ALICE", "W-BOB", Money.of("1200.50"));
        System.out.println("returned same txn: " + a.id + "; alice=" + Money.fmt(svc.wallet("W-ALICE").balance()));
        try { svc.transfer("pay-1", "W-ALICE", "W-BOB", Money.of("999")); }
        catch (IdempotencyConflictException e) { System.out.println("conflict: " + e.getMessage()); }

        System.out.println("== rule failures (recorded as FAILED, no ledger entries) ==");
        System.out.println(svc.transfer("pay-2", "W-BOB", "W-ALICE", Money.of("5000")));      // insufficient
        System.out.println(svc.transfer("pay-3", "W-ALICE", "W-BOB", Money.of("30000")));     // per-txn limit
        System.out.println(svc.transfer("pay-4", "W-ALICE", "W-BOB", Money.of("25000")));     // ok: spent 26200.50
        System.out.println(svc.transfer("pay-5", "W-ALICE", "W-BOB", Money.of("20000")));     // daily limit

        System.out.println("== ledger for alice ==");
        ledger.forWallet("W-ALICE").forEach(e -> System.out.println("  " + e));

        System.out.println("== stress: same key from 50 threads ==");
        WalletService s2 = new WalletService(List.of(), new Ledger(clock), clock);
        s2.createWallet("A", "a"); s2.createWallet("B", "b");
        s2.topUp("t", "A", Money.of("1000"));
        ExecutorService pool = Executors.newFixedThreadPool(16);
        Set<String> ids = ConcurrentHashMap.newKeySet();
        List<Future<?>> fs = new ArrayList<>();
        for (int i = 0; i < 50; i++) fs.add(pool.submit(() -> ids.add(s2.transfer("dup-key", "A", "B", Money.of("100")).id)));
        for (Future<?> f : fs) f.get();
        System.out.println("distinct txn ids=" + ids.size() + " A=" + Money.fmt(s2.wallet("A").balance()) + " B=" + Money.fmt(s2.wallet("B").balance()));

        System.out.println("== stress: random opposite-direction transfers across 16 threads (deadlock + conservation check) ==");
        Ledger l3 = new Ledger(clock);
        WalletService s3 = new WalletService(List.of(), l3, clock);
        s3.createWallet("A", "a"); s3.createWallet("B", "b"); s3.createWallet("C", "c");
        for (String w : List.of("A", "B", "C")) s3.topUp("init-" + w, w, Money.of("10000"));
        String[] ws = {"A", "B", "C"};
        AtomicInteger ok = new AtomicInteger(), failed = new AtomicInteger();
        List<Future<?>> fs3 = new ArrayList<>();
        for (int t = 0; t < 16; t++) {
            final int seed = t;
            fs3.add(pool.submit(() -> {
                Random r = new Random(seed);
                for (int i = 0; i < 1250; i++) {
                    String x = ws[r.nextInt(3)], y = ws[r.nextInt(3)];
                    if (x.equals(y)) continue;
                    WalletTransaction tx = s3.transfer("k-" + seed + "-" + i, x, y, Money.of(String.valueOf(1 + r.nextInt(900))));
                    (tx.status == TxnStatus.SUCCESS ? ok : failed).incrementAndGet();
                }
            }));
        }
        for (Future<?> f : fs3) f.get();
        pool.shutdown();
        long total = 0, min = Long.MAX_VALUE;
        for (String w : ws) { long b = s3.wallet(w).balance(); total += b; min = Math.min(min, b); }
        System.out.println("total=" + Money.fmt(total) + " (expected Rs 30000.00) noNegativeBalance=" + (min >= 0)
                + " ledgerBalanced=" + l3.isBalanced() + " (some transfers legitimately FAILED: insufficient funds)");
    }
}

== basic flow ==
    [SMS] TOP_UP of Rs 50000.00 SYS-SETTLEMENT -> W-ALICE ok (TXN-1)
    [SMS] TRANSFER of Rs 1200.50 W-ALICE -> W-BOB ok (TXN-2)
TXN-2 TRANSFER W-ALICE->W-BOB Rs 1200.50 SUCCESS
alice=Rs 48799.50 bob=Rs 1200.50
== idempotent retry (same key) ==
returned same txn: TXN-2; alice=Rs 48799.50
conflict: Idempotency key reused with different request: pay-1
== rule failures (recorded as FAILED, no ledger entries) ==
    [SMS] TRANSFER of Rs 5000.00 failed: Insufficient balance
TXN-3 TRANSFER W-BOB->W-ALICE Rs 5000.00 FAILED [Insufficient balance]
    [SMS] TRANSFER of Rs 30000.00 failed: Per-transaction limit Rs 25000.00 exceeded
TXN-4 TRANSFER W-ALICE->W-BOB Rs 30000.00 FAILED [Per-transaction limit Rs 25000.00 exceeded]
    [SMS] TRANSFER of Rs 25000.00 W-ALICE -> W-BOB ok (TXN-5)
TXN-5 TRANSFER W-ALICE->W-BOB Rs 25000.00 SUCCESS
    [SMS] TRANSFER of Rs 20000.00 failed: Daily limit Rs 40000.00 exceeded (spent Rs 26200.50)
TXN-6 TRANSFER W-ALICE->W-BOB Rs 20000.00 FAILED [Daily limit Rs 40000.00 exceeded (spent Rs 26200.50)]
== ledger for alice ==
  #2 TXN-1 CREDIT W-ALICE Rs 50000.00 balanceAfter=Rs 50000.00
  #3 TXN-2 DEBIT  W-ALICE Rs 1200.50 balanceAfter=Rs 48799.50
  #5 TXN-5 DEBIT  W-ALICE Rs 25000.00 balanceAfter=Rs 23799.50
== stress: same key from 50 threads ==
distinct txn ids=1 A=Rs 900.00 B=Rs 100.00
== stress: random opposite-direction transfers across 16 threads (deadlock + conservation check) ==
total=Rs 30000.00 (expected Rs 30000.00) noNegativeBalance=true ledgerBalanced=true (some transfers legitimately FAILED: insufficient funds)

transfer(key="pay-1", A -> B, 1200.50)
Client        WalletService              IdempotencyMap     Wallet A / B (locks)    Ledger   Listeners
 |--transfer-->|                              |                    |                   |          |
 |             |-validate args (amt>0, A!=B)  |                    |                   |          |
 |             |--putIfAbsent(key, entry)---->|                    |                   |          |
 |             |<-- null (I am the owner)-----|                    |                   |          |
 |             |-lock(min(A,B).id) then lock(max)------------------>|                  |          |
 |             |-check status, balance, LimitRules (consistent view under lock)         |          |
 |             |-debit A, credit B, recordSpend                    |                   |          |
 |             |--append DEBIT A, CREDIT B (same txnId)------------------------------>|          |
 |             |-unlock both                                       |                   |          |
 |             |-entry.future.complete(txn)   |                    |                   |          |
 |             |--------------------------------------------- onTransaction(txn) ---------------->| SMS
 |<--txn-------|

Duplicate (same key, same payload):  putIfAbsent returns existing -> future.join() -> same txn, NO debit
Duplicate with different payload:    fingerprint mismatch -> IdempotencyConflictException