Event Sourcing Fits: When Should You Use It (and When You Shouldn’t)?

// --- Domain events (facts). Keep them immutable. ---
import java.time.Instant;
import java.util.UUID;

public sealed interface WalletEvent permits MoneyDeposited, MoneyWithdrawn {
    UUID walletId();
    long amount();          // stored as minor unit, e.g., VND
    Instant occurredAt();
    String type();
}

public record MoneyDeposited(UUID walletId, long amount, Instant occurredAt) implements WalletEvent {
    public String type() { return "MoneyDeposited"; }
}

public record MoneyWithdrawn(UUID walletId, long amount, Instant occurredAt) implements WalletEvent {
    public String type() { return "MoneyWithdrawn"; }
}

// --- Aggregate: rebuilds state from events and enforces invariants on commands. ---
import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;

public final class WalletAggregate {
    private final UUID walletId;
    private long balance;
    private long version; // event stream version for optimistic concurrency

    private final List<walletevent> newEvents = new ArrayList<>();

    public WalletAggregate(UUID walletId) {
        this.walletId = walletId;
    }

    public static WalletAggregate rehydrate(UUID walletId, List<walletevent> history, long startingVersion) {
        WalletAggregate agg = new WalletAggregate(walletId);
        agg.version = startingVersion;
        for (WalletEvent e : history) {
            agg.apply(e);
            agg.version++; // each historical event increments version
        }
        return agg;
    }

    public void deposit(long amount) {
        if (amount <= 0) throw new IllegalArgumentException("amount must be > 0");
        newEvents.add(new MoneyDeposited(walletId, amount, Instant.now()));
    }

    public void withdraw(long amount) {
        if (amount <= 0) throw new IllegalArgumentException("amount must be > 0");
        long projected = balance - amount;
        if (projected < 0) throw new IllegalStateException("insufficient funds");
        newEvents.add(new MoneyWithdrawn(walletId, amount, Instant.now()));
    }

    public void apply(WalletEvent e) {
        if (e instanceof MoneyDeposited d) balance += d.amount();
        else if (e instanceof MoneyWithdrawn w) balance -= w.amount();
    }

    public List<walletevent> popNewEventsAndApply() {
        for (WalletEvent e : newEvents) apply(e);
        List<walletevent> out = List.copyOf(newEvents);
        newEvents.clear();
        return out;
    }

    public long balance() { return balance; }
    public long expectedVersion() { return version; }
}

// --- Event store: append-only storage with optimistic concurrency. ---
// This example uses JDBC, but the pattern works with Postgres, MySQL, etc.
import javax.sql.DataSource;
import java.sql.;
import java.time.Instant;
import java.util.;

public interface EventStore<e> {
    List<e> load(UUID streamId);
    long loadCurrentVersion(UUID streamId);
    void append(UUID streamId, long expectedVersion, List<e> events);
}

public final class JdbcWalletEventStore implements EventStore<walletevent> {
    private final DataSource ds;

    public JdbcWalletEventStore(DataSource ds) {
        this.ds = ds;
    }

    @Override
    public List<walletevent> load(UUID walletId) {
        String sql = """
            SELECT event_type, amount, occurred_at
            FROM wallet_events
            WHERE wallet_id = ?
            ORDER BY seq ASC
            """;
        List<walletevent> out = new ArrayList<>();
        try (Connection c = ds.getConnection();
             PreparedStatement ps = c.prepareStatement(sql)) {
            ps.setObject(1, walletId);
            try (ResultSet rs = ps.executeQuery()) {
                while (rs.next()) {
                    String type = rs.getString("event_type");
                    long amount = rs.getLong("amount");
                    Instant at = rs.getTimestamp("occurred_at").toInstant();
                    out.add(deserialize(walletId, type, amount, at));
                }
            }
            return out;
        } catch (SQLException ex) {
            throw new RuntimeException(ex);
        }
    }

    @Override
    public long loadCurrentVersion(UUID walletId) {
        String sql = "SELECT COALESCE(MAX(seq), 0) AS v FROM wallet_events WHERE wallet_id = ?";
        try (Connection c = ds.getConnection();
             PreparedStatement ps = c.prepareStatement(sql)) {
            ps.setObject(1, walletId);
            try (ResultSet rs = ps.executeQuery()) {
                rs.next();
                return rs.getLong("v");
            }
        } catch (SQLException ex) {
            throw new RuntimeException(ex);
        }
    }

    @Override
    public void append(UUID walletId, long expectedVersion, List<walletevent> events) {
        if (events.isEmpty()) return;

        // Important: concurrency check + append must be atomic.
        // One simple approach is to lock the stream row or rely on a unique constraint on (wallet_id, seq).
        String insert = """
            INSERT INTO wallet_events(wallet_id, seq, event_type, amount, occurred_at)
            VALUES (?, ?, ?, ?, ?)
            """;

        try (Connection c = ds.getConnection()) {
            c.setAutoCommit(false);

            long current = loadCurrentVersionTx(c, walletId);
            if (current != expectedVersion) {
                c.rollback();
                throw new IllegalStateException("concurrency conflict: expected=" + expectedVersion + " actual=" + current);
            }

            long nextSeq = current;
            try (PreparedStatement ps = c.prepareStatement(insert)) {
                for (WalletEvent e : events) {
                    nextSeq++;
                    ps.setObject(1, walletId);
                    ps.setLong(2, nextSeq);
                    ps.setString(3, e.type());
                    ps.setLong(4, e.amount());
                    ps.setTimestamp(5, Timestamp.from(e.occurredAt()));
                    ps.addBatch();
                }
                ps.executeBatch();
            }

            c.commit();
        } catch (SQLException ex) {
            throw new RuntimeException(ex);
        }
    }

    private long loadCurrentVersionTx(Connection c, UUID walletId) throws SQLException {
        String sql = "SELECT COALESCE(MAX(seq), 0) AS v FROM wallet_events WHERE wallet_id = ? FOR UPDATE";
        try (PreparedStatement ps = c.prepareStatement(sql)) {
            ps.setObject(1, walletId);
            try (ResultSet rs = ps.executeQuery()) {
                rs.next();
                return rs.getLong("v");
            }
        }
    }

    private WalletEvent deserialize(UUID walletId, String type, long amount, Instant at) {
        return switch (type) {
            case "MoneyDeposited" -> new MoneyDeposited(walletId, amount, at);
            case "MoneyWithdrawn" -> new MoneyWithdrawn(walletId, amount, at);
            default -> throw new IllegalArgumentException("unknown event type: " + type);
        };
    }
}

// --- Application service: command handling that loads history, applies command, and appends new events. ---
import java.util.List;
import java.util.UUID;

public final class WalletService {
    private final EventStore<walletevent> store;
    private final WalletBalanceProjection projection; // read model updater

    public WalletService(EventStore<walletevent> store, WalletBalanceProjection projection) {
        this.store = store;
        this.projection = projection;
    }

    public long deposit(UUID walletId, long amount) {
        long currentVersion = store.loadCurrentVersion(walletId);
        List<walletevent> history = store.load(walletId);

        WalletAggregate agg = WalletAggregate.rehydrate(walletId, history, 0);
        agg.deposit(amount);

        List<walletevent> newEvents = agg.popNewEventsAndApply();
        store.append(walletId, currentVersion, newEvents);

        // In production you often publish to a bus and update projections async,
        // but here we show sync to keep the example focused.
        projection.apply(newEvents);

        return agg.balance();
    }

    public long withdraw(UUID walletId, long amount) {
        long currentVersion = store.loadCurrentVersion(walletId);
        List<walletevent> history = store.load(walletId);

        WalletAggregate agg = WalletAggregate.rehydrate(walletId, history, 0);
        agg.withdraw(amount);

        List<walletevent> newEvents = agg.popNewEventsAndApply();
        store.append(walletId, currentVersion, newEvents);
        projection.apply(newEvents);

        return agg.balance();
    }
}
</walletevent></walletevent></walletevent></walletevent></walletevent></walletevent>

// --- Projection: fast read model table. ---
import javax.sql.DataSource;
import java.sql.*;
import java.util.List;
import java.util.UUID;

public final class WalletBalanceProjection {
    private final DataSource ds;

    public WalletBalanceProjection(DataSource ds) {
        this.ds = ds;
    }

    public void apply(List<walletevent> events) {
        for (WalletEvent e : events) {
            if (e instanceof MoneyDeposited d) upsert(d.walletId(), +d.amount());
            else if (e instanceof MoneyWithdrawn w) upsert(w.walletId(), -w.amount());
        }
    }

    private void upsert(UUID walletId, long delta) {
        String sql = """
            INSERT INTO wallet_balance(wallet_id, balance)
            VALUES (?, ?)
            ON CONFLICT (wallet_id) DO UPDATE SET balance = wallet_balance.balance + EXCLUDED.balance
            """;
        try (Connection c = ds.getConnection();
             PreparedStatement ps = c.prepareStatement(sql)) {
            ps.setObject(1, walletId);
            ps.setLong(2, delta);
            ps.executeUpdate();
        } catch (SQLException ex) {
            throw new RuntimeException(ex);
        }
    }
}
</walletevent>

Read more at : Event Sourcing Fits: When Should You Use It (and When You Shouldn’t)?