Event Sourcing: A Complete Guide with Examples in 11 Programming Languages

Event Sourcing is an architectural pattern in which the state of an application is stored not as a snapshot of the current values, but as an ordered, append-only log of every event that has ever occurred. Instead of saving { status: "shipped", address: "123 Main St" } to a database row and overwriting it when things change, you save OrderPlaced, AddressUpdated, PaymentConfirmed, OrderShipped — each event an immutable fact about what happened. The current state is derived on demand by replaying the event log from the beginning.

This shifts the fundamental question from "What is the current state?" to "What happened, and in what order?" That shift has profound consequences: you gain a complete, tamper-evident history of every change; you can reconstruct the state of any entity at any point in time; you can build new read models by replaying old events; and you eliminate the impedance mismatch between a domain model's transitions and a relational store that only remembers the latest state.

In this comprehensive guide, we'll explore Event Sourcing with real-world examples in Python, TypeScript, Java, JavaScript, C#, PHP, Go, Rust, Dart, Swift, and Kotlin — complete with language-specific best practices, common pitfalls, and the judgment to apply it where it genuinely fits.

Table of Contents

• What is Event Sourcing?
• Why Use Event Sourcing?
• Event Sourcing Comparison
• Event Sourcing Explained
• Architecture Diagram
• Beginner-Friendly Example
• Production-Ready Example
• Real-World Use Cases
• Language-Specific Mistakes and Anti-Patterns
• Frequently Asked Questions
• Key Takeaways

What is Event Sourcing?

Most applications store state by mutation: a user updates their email address and the database row is overwritten. The previous email is gone. When was it changed? Why? By whom? There is no record — unless the developer happened to add an audit log table as an afterthought.

Event Sourcing takes a different approach: never overwrite state; append events instead. When a user updates their email, an EmailChanged event is stored: { userId: "u-123", oldEmail: "a@x.com", newEmail: "b@x.com", changedAt: "2024-01-15T10:30:00Z", changedBy: "admin-1" }. The current email is the value from the most recent EmailChanged event. The previous email is still there — it was never erased.

The event store is the source of truth. It is an append-only log, ordered by sequence number or timestamp. Every state change is an event appended to the stream. The current state of any entity is derived by loading its events and replaying them in order — a process called event replay or reconstitution.

This is not a new idea. Financial ledgers have worked this way for centuries: transactions are appended; the balance is computed from the sum of all transactions. Source control (Git) works this way: commits are events; the current file state is derived from replaying the commit history. The append-only log is a battle-tested foundation for correctness.

Event Sourcing became prominent in software architecture through Greg Young and Martin Fowler's work on Domain-Driven Design and CQRS. It pairs naturally with CQRS (which it complements but does not require) and with Domain Events (which become the atoms of the event store).

Why Use Event Sourcing?

Event Sourcing is justified when one or more of these requirements are present:

1. Complete audit trail: Every state change is an immutable, ordered fact. Who changed what, when, and from what previous value — all available without a separate audit log table. Regulatory compliance (GDPR, SOX, HIPAA, PCI-DSS) often mandates this.

2. Temporal queries: "What was the state of this order on March 15th?" is a trivial query against an event log — replay events up to that timestamp. Against a snapshot store, you need either a history table or a time-travel feature.

3. Debugging and forensics: When something goes wrong, replay the event stream to see exactly how the system arrived at the broken state. Step through events one by one, inspecting state after each application.

4. Event-driven integration: Other services can subscribe to your event stream and build their own projections. Your event log becomes a publish/subscribe integration bus. Adding a new consumer does not require changes to the producing service.

5. Projection rebuilding: Read models are derived from events. Adding a new read model — a new dashboard, a new report, a new API view — means subscribing to the event stream and building the projection from scratch. No schema migration required.

6. Business value of history: In domains where history is the product (banking transactions, medical records, legal documents, accounting entries), Event Sourcing aligns the storage model with the domain model. The history is not a side effect — it is the thing itself.

Event Sourcing Comparison

Event Sourcing vs. CRUD: CRUD stores the latest snapshot and discards history. Event Sourcing stores all history and derives the latest snapshot. CRUD is simpler for simple domains; Event Sourcing is more powerful when history matters.

Event Sourcing vs. Audit Log: An audit log is an afterthought — it captures some fields, sometimes, for compliance. Event Sourcing makes the event log the primary source of truth — every state change is captured, completely and consistently, because that is the only way state changes are recorded.

Event Sourcing vs. CQRS: These are independent patterns. Event Sourcing concerns where you store state (event log vs. snapshot). CQRS concerns how you separate reads from writes (two models vs. one). They combine powerfully: Event Sourcing provides the event stream; CQRS projections consume that stream to build read models. But you can use either without the other.

Event Sourcing vs. Temporal Tables: Temporal tables (SQL Server, PostgreSQL) store row history in a relational structure. This is simpler than Event Sourcing and sufficient for simple auditing. Event Sourcing goes further: events carry business meaning, can be replayed to build any projection, and integrate natively with event-driven architectures.

Event Sourcing Explained

Core Components

Domain Event: An immutable record of something that happened in the domain. Named in the past tense: OrderPlaced, PaymentFailed, ItemAddedToCart, UserEmailChanged. Each event carries all the data relevant to what happened — enough to reconstruct the state change without any additional context. Events are facts; they cannot be deleted or modified.

Event Stream: The ordered sequence of events for a single aggregate (entity). An order's event stream might be: [OrderPlaced, ItemAdded, ItemRemoved, PaymentConfirmed, OrderShipped]. Each event has a sequence number (position in the stream) and a timestamp.

Event Store: The append-only database that persists event streams. The write path is append(stream_id, expected_version, events). The read path is load(stream_id, from_version). No updates; no deletes. Common implementations: EventStoreDB, Apache Kafka (as a log), PostgreSQL with an events table, DynamoDB.

Aggregate: A domain entity whose state is derived from its event stream. The aggregate has no mutable state fields — it has an apply(event) method that updates its internal state from each event, and a handle(command) method that produces new events. Loading an aggregate means loading its event stream and replaying all events through apply().

Reconstitution (Replay): The process of rebuilding an aggregate's current state by loading all its events from the event store and applying them in sequence. For aggregates with many events, this can be expensive — mitigated by snapshots.

Snapshot: An optimization: a serialized copy of an aggregate's state at a specific version. When loading an aggregate, load the latest snapshot (if any) plus only the events after the snapshot version — instead of replaying from event 1. Snapshots are a performance optimization only; the event log remains the source of truth.

Projection / Read Model: A derived view of the event stream, built for a specific query purpose. A projection subscribes to an event stream, processes each event, and updates a queryable store (a relational table, a document, a cache). Because projections are derived, they can be dropped and rebuilt from the event log at any time.

Event Handler / Projector: The function that processes events to build or update a projection. A CustomerOrderHistoryProjector subscribes to OrderPlaced, OrderPaid, OrderShipped, OrderCancelled events and maintains a denormalized customer order history table.

The Write Path

1. A Command arrives (e.g., ShipOrder(orderId, trackingNumber)).
2. The Command Handler loads the aggregate's event stream from the event store.
3. The aggregate is reconstituted by replaying all events through apply().
4. The command handler calls a domain method on the aggregate (e.g., order.ship(trackingNumber)).
5. The domain method validates business rules and produces new domain events (e.g., OrderShipped).
6. The new events are appended to the event stream in the event store (with optimistic concurrency check on the expected version).
7. The events are published to interested consumers — projectors, other services, notification handlers.

The Read Path

1. A Query arrives (e.g., GetOrderHistory(customerId)).
2. The Query Handler reads from a projection — a pre-built, query-optimized view maintained by event handlers.
3. The projection returns immediately — no event replay needed for reads.

Optimistic Concurrency

The event store enforces optimistic concurrency: when appending events, you specify the expected current version of the stream. If another process has appended events since you loaded the stream, your append fails with a concurrency exception. You must reload the stream, reconstitute the aggregate with the new events, and retry if the command is still valid.

This prevents lost updates — a critical correctness guarantee in an append-only system.

Event Versioning

Events are immutable, but their structure must evolve as the domain evolves. Strategies:

• Upcasting: when loading old events, transform them to the current schema before applying. The event store holds the original; the aggregate sees the current version.
• New event types: add OrderPlacedV2 alongside OrderPlaced; handle both in apply(). Gradually migrate.
• Weak schema: use flexible serialization (JSON with optional fields) so new fields in old events default to sensible values.

Architecture Diagram

Beginner-Friendly Example

The clearest Event Sourcing introduction: a bank account where every deposit, withdrawal, and opening is recorded as an event. The balance is never stored directly — it is always derived by replaying the event history.

from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime, timezone
from decimal import Decimal
from typing import Optional
from uuid import uuid4


# ════════════════════════════════════════════════════════════════
# Domain Events — immutable facts about what happened
# ════════════════════════════════════════════════════════════════

@dataclass(frozen=True)
class DomainEvent:
    event_id:    str      = field(default_factory=lambda: str(uuid4())[:8])
    occurred_at: datetime = field(default_factory=lambda: datetime.now(tz=timezone.utc))
    version:     int      = 0   # sequence number within the stream


@dataclass(frozen=True)
class AccountOpenedEvent(DomainEvent):
    account_id:    str     = ""
    owner_name:    str     = ""
    initial_deposit: str   = "0.00"   # Decimal as string for frozen dataclass


@dataclass(frozen=True)
class MoneyDepositedEvent(DomainEvent):
    account_id: str = ""
    amount:     str = "0.00"
    description: str = ""


@dataclass(frozen=True)
class MoneyWithdrawnEvent(DomainEvent):
    account_id: str = ""
    amount:     str = "0.00"
    description: str = ""


@dataclass(frozen=True)
class AccountClosedEvent(DomainEvent):
    account_id: str = ""
    reason:     str = ""


# ════════════════════════════════════════════════════════════════
# Aggregate — state derived entirely from events
# ════════════════════════════════════════════════════════════════

class BankAccount:
    """
    Event-Sourced aggregate.
    - State is NEVER stored directly; always derived from events.
    - Domain methods produce events — they do NOT mutate state directly.
    - apply() processes each event to update internal state.
    - Reconstitution = loading events + replaying through apply().
    """

    def __init__(self) -> None:
        # Internal state — updated ONLY through apply()
        self.account_id: str             = ""
        self.owner_name: str             = ""
        self.balance:    Decimal         = Decimal("0.00")
        self.is_open:    bool            = False
        self.version:    int             = 0   # current stream version

        # Unpublished events produced this session
        self._pending_events: list[DomainEvent] = []

    # ── Reconstitution ────────────────────────────────────────
    @classmethod
    def from_events(cls, events: list[DomainEvent]) -> "BankAccount":
        """Rebuild current state by replaying all events in order."""
        account = cls()
        for event in events:
            account._apply(event)
            account.version = event.version
        return account

    def _apply(self, event: DomainEvent) -> None:
        """Route each event to its handler; update internal state."""
        match event:
            case AccountOpenedEvent():
                self.account_id = event.account_id
                self.owner_name = event.owner_name
                self.balance    = Decimal(event.initial_deposit)
                self.is_open    = True

            case MoneyDepositedEvent():
                self.balance += Decimal(event.amount)

            case MoneyWithdrawnEvent():
                self.balance -= Decimal(event.amount)

            case AccountClosedEvent():
                self.is_open = False

    # ── Domain methods — produce events, enforce invariants ───
    @classmethod
    def open(cls, owner_name: str, initial_deposit: Decimal) -> "BankAccount":
        if initial_deposit < Decimal("0"):
            raise ValueError("Initial deposit cannot be negative")
        account = cls()
        event = AccountOpenedEvent(
            account_id      = str(uuid4())[:8],
            owner_name      = owner_name,
            initial_deposit = str(initial_deposit),
            version         = 1,
        )
        account._apply(event)
        account._pending_events.append(event)
        account.version = 1
        return account

    def deposit(self, amount: Decimal, description: str = "") -> None:
        if not self.is_open:
            raise ValueError("Cannot deposit to a closed account")
        if amount <= Decimal("0"):
            raise ValueError("Deposit amount must be positive")
        event = MoneyDepositedEvent(
            account_id  = self.account_id,
            amount      = str(amount),
            description = description,
            version     = self.version + 1,
        )
        self._apply(event)
        self._pending_events.append(event)
        self.version += 1

    def withdraw(self, amount: Decimal, description: str = "") -> None:
        if not self.is_open:
            raise ValueError("Cannot withdraw from a closed account")
        if amount <= Decimal("0"):
            raise ValueError("Withdrawal amount must be positive")
        if amount > self.balance:
            raise ValueError(f"Insufficient funds: balance={self.balance}, requested={amount}")
        event = MoneyWithdrawnEvent(
            account_id  = self.account_id,
            amount      = str(amount),
            description = description,
            version     = self.version + 1,
        )
        self._apply(event)
        self._pending_events.append(event)
        self.version += 1

    def close(self, reason: str) -> None:
        if not self.is_open:
            raise ValueError("Account is already closed")
        if self.balance != Decimal("0"):
            raise ValueError(f"Cannot close account with balance {self.balance}")
        event = AccountClosedEvent(
            account_id = self.account_id,
            reason     = reason,
            version    = self.version + 1,
        )
        self._apply(event)
        self._pending_events.append(event)
        self.version += 1

    def collect_events(self) -> list[DomainEvent]:
        """Drain and return pending events for persistence."""
        events = list(self._pending_events)
        self._pending_events.clear()
        return events

    def __repr__(self) -> str:
        return (f"BankAccount(id={self.account_id}, owner={self.owner_name!r}, "
                f"balance={self.balance}, open={self.is_open}, v={self.version})")


# ════════════════════════════════════════════════════════════════
# In-memory Event Store
# ════════════════════════════════════════════════════════════════

class InMemoryEventStore:
    """
    Simplified event store — append-only, ordered by version.
    Real implementations: EventStoreDB, PostgreSQL events table, Kafka.
    """

    def __init__(self) -> None:
        self._streams: dict[str, list[DomainEvent]] = {}

    def append(self, stream_id: str, events: list[DomainEvent],
               expected_version: int) -> None:
        """
        Append events to a stream with optimistic concurrency check.
        expected_version = current version caller loaded from.
        Fails if another writer has appended since then.
        """
        current = self._streams.get(stream_id, [])
        current_version = len(current)

        if current_version != expected_version:
            raise ConcurrencyError(
                f"Concurrency conflict on stream '{stream_id}': "
                f"expected v{expected_version}, actual v{current_version}"
            )

        self._streams.setdefault(stream_id, []).extend(events)
        print(f"  [EventStore] Appended {len(events)} event(s) to stream '{stream_id}'")

    def load(self, stream_id: str, from_version: int = 0) -> list[DomainEvent]:
        """Load all events for a stream from a given version."""
        return [e for e in self._streams.get(stream_id, [])
                if e.version > from_version]

    def load_all(self, stream_id: str) -> list[DomainEvent]:
        return self._streams.get(stream_id, [])

    def stream_ids(self) -> list[str]:
        return list(self._streams.keys())


class ConcurrencyError(Exception):
    pass


# ════════════════════════════════════════════════════════════════
# Command Handlers
# ════════════════════════════════════════════════════════════════

class BankAccountCommandHandler:
    def __init__(self, event_store: InMemoryEventStore) -> None:
        self._store = event_store

    def _load_account(self, account_id: str) -> BankAccount:
        events = self._store.load_all(account_id)
        if not events:
            raise KeyError(f"Account not found: {account_id}")
        return BankAccount.from_events(events)

    def open_account(self, owner_name: str,
                     initial_deposit: Decimal) -> str:
        account = BankAccount.open(owner_name, initial_deposit)
        events  = account.collect_events()
        self._store.append(account.account_id, events, expected_version=0)
        return account.account_id

    def deposit(self, account_id: str, amount: Decimal,
                description: str = "") -> None:
        account = self._load_account(account_id)
        account.deposit(amount, description)
        events = account.collect_events()
        self._store.append(account_id, events,
                           expected_version=account.version - len(events))

    def withdraw(self, account_id: str, amount: Decimal,
                 description: str = "") -> None:
        account = self._load_account(account_id)
        account.withdraw(amount, description)
        events = account.collect_events()
        self._store.append(account_id, events,
                           expected_version=account.version - len(events))

    def close_account(self, account_id: str, reason: str) -> None:
        account = self._load_account(account_id)
        account.close(reason)
        events = account.collect_events()
        self._store.append(account_id, events,
                           expected_version=account.version - len(events))


# ════════════════════════════════════════════════════════════════
# Projection — account balance view (rebuilt from events)
# ════════════════════════════════════════════════════════════════

class AccountBalanceProjection:
    """
    Read model: current balance and transaction count.
    Built by replaying all events — can be dropped and rebuilt at any time.
    """

    def __init__(self) -> None:
        self._balances: dict[str, dict] = {}

    def handle(self, event: DomainEvent) -> None:
        match event:
            case AccountOpenedEvent():
                self._balances[event.account_id] = {
                    "account_id":        event.account_id,
                    "owner":             event.owner_name,
                    "balance":           Decimal(event.initial_deposit),
                    "transaction_count": 0,
                    "is_open":           True,
                }
            case MoneyDepositedEvent():
                if view := self._balances.get(event.account_id):
                    view["balance"]           += Decimal(event.amount)
                    view["transaction_count"] += 1
            case MoneyWithdrawnEvent():
                if view := self._balances.get(event.account_id):
                    view["balance"]           -= Decimal(event.amount)
                    view["transaction_count"] += 1
            case AccountClosedEvent():
                if view := self._balances.get(event.account_id):
                    view["is_open"] = False

    def get(self, account_id: str) -> Optional[dict]:
        return self._balances.get(account_id)

    def rebuild(self, event_store: InMemoryEventStore,
                account_id: str) -> None:
        """Rebuild this projection from scratch for a given account."""
        self._balances.pop(account_id, None)
        for event in event_store.load_all(account_id):
            self.handle(event)
        print(f"  [Projection] Rebuilt balance view for {account_id}")


# ════════════════════════════════════════════════════════════════
# Demonstration
# ════════════════════════════════════════════════════════════════

if __name__ == "__main__":
    store      = InMemoryEventStore()
    handler    = BankAccountCommandHandler(store)
    projection = AccountBalanceProjection()

    print("=" * 55)
    print("1. Open account and make transactions")
    print("=" * 55)

    acct_id = handler.open_account("Alice Johnson", Decimal("500.00"))
    # Sync projection from the events just appended
    for e in store.load_all(acct_id):
        projection.handle(e)

    handler.deposit(acct_id, Decimal("250.00"), "Paycheck")
    for e in store.load_all(acct_id)[-1:]:
        projection.handle(e)

    handler.withdraw(acct_id, Decimal("100.00"), "Rent")
    for e in store.load_all(acct_id)[-1:]:
        projection.handle(e)

    handler.deposit(acct_id, Decimal("75.50"), "Freelance work")
    for e in store.load_all(acct_id)[-1:]:
        projection.handle(e)

    print("\n" + "=" * 55)
    print("2. Query: Current balance from projection")
    print("=" * 55)
    view = projection.get(acct_id)
    print(f"  Owner:        {view['owner']}")
    print(f"  Balance:      ${view['balance']}")
    print(f"  Transactions: {view['transaction_count']}")
    print(f"  Open:         {view['is_open']}")

    print("\n" + "=" * 55)
    print("3. Time-travel: replay events up to version 2")
    print("=" * 55)
    all_events   = store.load_all(acct_id)
    past_events  = [e for e in all_events if e.version <= 2]
    past_account = BankAccount.from_events(past_events)
    print(f"  State at v2: balance=${past_account.balance}, "
          f"open={past_account.is_open}, version={past_account.version}")

    print("\n" + "=" * 55)
    print("4. Full event log — the source of truth")
    print("=" * 55)
    for event in store.load_all(acct_id):
        print(f"  v{event.version:02d} [{type(event).__name__:<25}] "
              f"at {event.occurred_at.strftime('%H:%M:%S')}")

    print("\n" + "=" * 55)
    print("5. Domain rule: cannot overdraw")
    print("=" * 55)
    try:
        handler.withdraw(acct_id, Decimal("1000.00"), "Too much")
    except ValueError as e:
        print(f"  Expected error: {e}")

    print("\n" + "=" * 55)
    print("6. Close account (balance must be 0)")
    print("=" * 55)
    try:
        handler.close_account(acct_id, "Switching banks")
    except ValueError as e:
        print(f"  Expected: {e}")

    # Withdraw remaining balance first
    balance = projection.get(acct_id)["balance"]
    handler.withdraw(acct_id, balance, "Final withdrawal")
    handler.close_account(acct_id, "Switching banks")

    # Rebuild projection from scratch — proves event log is the source of truth
    print("\n" + "=" * 55)
    print("7. Rebuild projection from event log (drop and replay)")
    print("=" * 55)
    projection.rebuild(store, acct_id)
    final_view = projection.get(acct_id)
    print(f"  Rebuilt: balance=${final_view['balance']}, open={final_view['is_open']}, "
          f"tx_count={final_view['transaction_count']}")
    print(f"  Total events in stream: {len(store.load_all(acct_id))}")

Production-Ready Example

The most instructive Event Sourcing challenge in production: an e-commerce order aggregate that records every state change as an event, supports full lifecycle (placed → items modified → paid → shipped → delivered), enforces domain invariants, and exposes three independent projections rebuilt from the event log.

🐍 Python (Production)

from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime, timezone
from decimal import Decimal
from enum import Enum
from typing import Optional, Any
from uuid import uuid4


# ════════════════════════════════════════════════════════════════
# Domain Events — every state change as an immutable fact
# ════════════════════════════════════════════════════════════════

@dataclass(frozen=True)
class OrderEvent:
    event_id:   str      = field(default_factory=lambda: str(uuid4())[:8])
    occurred_at: datetime = field(default_factory=lambda: datetime.now(tz=timezone.utc))
    version:     int      = 0


@dataclass(frozen=True)
class OrderPlacedEvent(OrderEvent):
    order_id:    str   = ""
    customer_id: str   = ""
    items:       tuple = ()      # tuple of {"sku", "qty", "unit_price"}


@dataclass(frozen=True)
class ItemAddedEvent(OrderEvent):
    order_id:   str = ""
    sku:        str = ""
    qty:        int = 0
    unit_price: str = "0.00"


@dataclass(frozen=True)
class ItemRemovedEvent(OrderEvent):
    order_id: str = ""
    sku:      str = ""


@dataclass(frozen=True)
class ShippingAddressSetEvent(OrderEvent):
    order_id: str = ""
    address:  str = ""


@dataclass(frozen=True)
class PaymentConfirmedEvent(OrderEvent):
    order_id:   str = ""
    payment_id: str = ""
    amount:     str = "0.00"


@dataclass(frozen=True)
class OrderShippedEvent(OrderEvent):
    order_id:        str = ""
    tracking_number: str = ""
    carrier:         str = ""


@dataclass(frozen=True)
class OrderDeliveredEvent(OrderEvent):
    order_id:     str = ""
    delivered_at: str = ""


@dataclass(frozen=True)
class OrderCancelledEvent(OrderEvent):
    order_id: str = ""
    reason:   str = ""


# ════════════════════════════════════════════════════════════════
# Order status enum
# ════════════════════════════════════════════════════════════════

class OrderStatus(str, Enum):
    DRAFT     = "draft"
    CONFIRMED = "confirmed"
    PAID      = "paid"
    SHIPPED   = "shipped"
    DELIVERED = "delivered"
    CANCELLED = "cancelled"


# ════════════════════════════════════════════════════════════════
# Order Aggregate — state derived from events
# ════════════════════════════════════════════════════════════════

@dataclass
class OrderItem:
    sku:        str
    qty:        int
    unit_price: Decimal


class Order:
    """
    Event-Sourced Order Aggregate.

    Rules:
    - All state changes produce events; no direct mutation.
    - apply() is the only way state is updated.
    - Business rules are enforced in domain methods, not in apply().
    - apply() must be side-effect-free and deterministic.
    """

    def __init__(self) -> None:
        # State — never set directly outside of apply()
        self.order_id:        str              = ""
        self.customer_id:     str              = ""
        self.items:           dict[str, OrderItem] = {}
        self.status:          OrderStatus      = OrderStatus.DRAFT
        self.shipping_address: Optional[str]   = None
        self.payment_id:      Optional[str]    = None
        self.tracking_number: Optional[str]    = None
        self.version:         int              = 0
        self._pending:        list[OrderEvent] = []

    @classmethod
    def from_events(cls, events: list[OrderEvent]) -> "Order":
        order = cls()
        for event in events:
            order._apply(event)
            order.version = event.version
        return order

    def _apply(self, event: OrderEvent) -> None:
        """Pure state update — no validation, no side effects."""
        match event:
            case OrderPlacedEvent():
                self.order_id    = event.order_id
                self.customer_id = event.customer_id
                self.status      = OrderStatus.DRAFT
                self.items       = {
                    i["sku"]: OrderItem(i["sku"], i["qty"], Decimal(i["unit_price"]))
                    for i in event.items
                }

            case ItemAddedEvent():
                if event.sku in self.items:
                    existing = self.items[event.sku]
                    self.items[event.sku] = OrderItem(
                        event.sku, existing.qty + event.qty, existing.unit_price
                    )
                else:
                    self.items[event.sku] = OrderItem(
                        event.sku, event.qty, Decimal(event.unit_price)
                    )

            case ItemRemovedEvent():
                self.items.pop(event.sku, None)

            case ShippingAddressSetEvent():
                self.shipping_address = event.address

            case PaymentConfirmedEvent():
                self.payment_id = event.payment_id
                self.status     = OrderStatus.PAID

            case OrderShippedEvent():
                self.tracking_number = event.tracking_number
                self.status          = OrderStatus.SHIPPED

            case OrderDeliveredEvent():
                self.status = OrderStatus.DELIVERED

            case OrderCancelledEvent():
                self.status = OrderStatus.CANCELLED

    def _emit(self, event: OrderEvent) -> None:
        self._apply(event)
        self._pending.append(event)
        self.version = event.version

    # ── Domain Methods — enforce invariants, produce events ───

    @classmethod
    def place(cls, customer_id: str,
              items: list[dict]) -> "Order":
        if not items:
            raise ValueError("Order must have at least one item")
        order = cls()
        event = OrderPlacedEvent(
            order_id    = str(uuid4())[:8],
            customer_id = customer_id,
            items       = tuple(items),
            version     = 1,
        )
        order._emit(event)
        return order

    def add_item(self, sku: str, qty: int,
                 unit_price: Decimal) -> None:
        if self.status not in (OrderStatus.DRAFT,):
            raise ValueError(f"Cannot add items in status '{self.status}'")
        if qty <= 0:
            raise ValueError("Quantity must be positive")
        self._emit(ItemAddedEvent(
            order_id=self.order_id, sku=sku, qty=qty,
            unit_price=str(unit_price), version=self.version + 1,
        ))

    def remove_item(self, sku: str) -> None:
        if self.status != OrderStatus.DRAFT:
            raise ValueError(f"Cannot remove items in status '{self.status}'")
        if sku not in self.items:
            raise ValueError(f"Item not found: {sku}")
        self._emit(ItemRemovedEvent(
            order_id=self.order_id, sku=sku,
            version=self.version + 1,
        ))

    def set_shipping_address(self, address: str) -> None:
        if self.status not in (OrderStatus.DRAFT, OrderStatus.CONFIRMED):
            raise ValueError(f"Cannot set address in status '{self.status}'")
        if not address.strip():
            raise ValueError("Address cannot be empty")
        self._emit(ShippingAddressSetEvent(
            order_id=self.order_id, address=address.strip(),
            version=self.version + 1,
        ))

    def confirm_payment(self, payment_id: str,
                        amount: Decimal) -> None:
        if self.status != OrderStatus.DRAFT:
            raise ValueError(f"Cannot pay in status '{self.status}'")
        if not self.shipping_address:
            raise ValueError("Shipping address must be set before payment")
        if amount < self.total:
            raise ValueError(f"Payment {amount} insufficient for total {self.total}")
        self._emit(PaymentConfirmedEvent(
            order_id=self.order_id, payment_id=payment_id,
            amount=str(amount), version=self.version + 1,
        ))

    def ship(self, tracking_number: str, carrier: str) -> None:
        if self.status != OrderStatus.PAID:
            raise ValueError(f"Cannot ship in status '{self.status}'")
        self._emit(OrderShippedEvent(
            order_id=self.order_id, tracking_number=tracking_number,
            carrier=carrier, version=self.version + 1,
        ))

    def deliver(self) -> None:
        if self.status != OrderStatus.SHIPPED:
            raise ValueError(f"Cannot deliver in status '{self.status}'")
        self._emit(OrderDeliveredEvent(
            order_id=self.order_id,
            delivered_at=datetime.now(tz=timezone.utc).isoformat(),
            version=self.version + 1,
        ))

    def cancel(self, reason: str) -> None:
        if self.status in (OrderStatus.SHIPPED, OrderStatus.DELIVERED):
            raise ValueError(f"Cannot cancel a {self.status} order")
        if self.status == OrderStatus.CANCELLED:
            raise ValueError("Order is already cancelled")
        self._emit(OrderCancelledEvent(
            order_id=self.order_id, reason=reason,
            version=self.version + 1,
        ))

    @property
    def total(self) -> Decimal:
        return sum(item.qty * item.unit_price for item in self.items.values())

    def collect_events(self) -> list[OrderEvent]:
        events = list(self._pending)
        self._pending.clear()
        return events

    def __repr__(self) -> str:
        return (f"Order(id={self.order_id}, status={self.status.value}, "
                f"total={self.total}, items={len(self.items)}, v={self.version})")


# ════════════════════════════════════════════════════════════════
# Event Store with snapshot support
# ════════════════════════════════════════════════════════════════

@dataclass
class Snapshot:
    stream_id: str
    version:   int
    state:     dict      # serialized aggregate state


class ProductionEventStore:
    """
    Append-only event store with snapshot optimization.
    Real implementations: EventStoreDB, PostgreSQL, DynamoDB.
    """

    def __init__(self, snapshot_threshold: int = 10) -> None:
        self._streams:   dict[str, list[OrderEvent]] = {}
        self._snapshots: dict[str, Snapshot]         = {}
        self._snapshot_threshold = snapshot_threshold

    def append(self, stream_id: str, events: list[OrderEvent],
               expected_version: int) -> None:
        current = self._streams.get(stream_id, [])
        if len(current) != expected_version:
            raise ConcurrencyError(
                f"Conflict on '{stream_id}': expected v{expected_version}, "
                f"got v{len(current)}"
            )
        self._streams.setdefault(stream_id, []).extend(events)

        # Auto-snapshot if stream is long
        total_events = len(self._streams[stream_id])
        if total_events % self._snapshot_threshold == 0:
            self._take_snapshot(stream_id, total_events)

        print(f"  [EventStore] +{len(events)} event(s) → '{stream_id}' "
              f"(total: {total_events})")

    def _take_snapshot(self, stream_id: str, version: int) -> None:
        """Create a snapshot of the current aggregate state."""
        events = self._streams.get(stream_id, [])
        order  = Order.from_events(events)
        self._snapshots[stream_id] = Snapshot(
            stream_id=stream_id,
            version=version,
            state={
                "order_id":         order.order_id,
                "customer_id":      order.customer_id,
                "status":           order.status.value,
                "items":            {
                    sku: {"sku": i.sku, "qty": i.qty, "unit_price": str(i.unit_price)}
                    for sku, i in order.items.items()
                },
                "shipping_address": order.shipping_address,
                "payment_id":       order.payment_id,
                "tracking_number":  order.tracking_number,
                "version":          order.version,
            }
        )
        print(f"  [Snapshot] Took snapshot at v{version} for '{stream_id}'")

    def load(self, stream_id: str) -> tuple[Optional[Snapshot], list[OrderEvent]]:
        """
        Returns (snapshot_or_None, events_after_snapshot).
        Caller uses snapshot for initial state, then applies remaining events.
        """
        snapshot      = self._snapshots.get(stream_id)
        all_events    = self._streams.get(stream_id, [])
        if snapshot:
            after_snapshot = [e for e in all_events
                              if e.version > snapshot.version]
            print(f"  [EventStore] Loaded snapshot v{snapshot.version} "
                  f"+ {len(after_snapshot)} event(s) for '{stream_id}'")
            return snapshot, after_snapshot
        print(f"  [EventStore] Loaded {len(all_events)} event(s) "
              f"(no snapshot) for '{stream_id}'")
        return None, all_events

    def load_all_events(self, stream_id: str) -> list[OrderEvent]:
        return self._streams.get(stream_id, [])

    def load_all_streams(self) -> dict[str, list[OrderEvent]]:
        return dict(self._streams)


class ConcurrencyError(Exception):
    pass


# ════════════════════════════════════════════════════════════════
# Projections — three independent read models
# ════════════════════════════════════════════════════════════════

class OrderDetailProjection:
    """Full order detail for the order detail page."""

    def __init__(self) -> None:
        self._views: dict[str, dict] = {}

    def handle(self, event: OrderEvent) -> None:
        match event:
            case OrderPlacedEvent():
                self._views[event.order_id] = {
                    "order_id":        event.order_id,
                    "customer_id":     event.customer_id,
                    "status":          "draft",
                    "items":           {i["sku"]: i for i in event.items},
                    "shipping_address": None,
                    "payment_id":      None,
                    "tracking_number": None,
                    "total":           str(sum(
                        Decimal(i["unit_price"]) * i["qty"] for i in event.items
                    )),
                    "placed_at":       event.occurred_at.isoformat(),
                }
            case ItemAddedEvent():
                if v := self._views.get(event.order_id):
                    if event.sku in v["items"]:
                        v["items"][event.sku]["qty"] += event.qty
                    else:
                        v["items"][event.sku] = {"sku": event.sku, "qty": event.qty, "unit_price": event.unit_price}
                    v["total"] = str(sum(
                        Decimal(i["unit_price"]) * i["qty"] for i in v["items"].values()
                    ))
            case ItemRemovedEvent():
                if v := self._views.get(event.order_id):
                    v["items"].pop(event.sku, None)
                    v["total"] = str(sum(
                        Decimal(i["unit_price"]) * i["qty"] for i in v["items"].values()
                    ))
            case ShippingAddressSetEvent():
                if v := self._views.get(event.order_id):
                    v["shipping_address"] = event.address
            case PaymentConfirmedEvent():
                if v := self._views.get(event.order_id):
                    v["status"]     = "paid"
                    v["payment_id"] = event.payment_id
            case OrderShippedEvent():
                if v := self._views.get(event.order_id):
                    v["status"]          = "shipped"
                    v["tracking_number"] = event.tracking_number
            case OrderDeliveredEvent():
                if v := self._views.get(event.order_id):
                    v["status"] = "delivered"
            case OrderCancelledEvent():
                if v := self._views.get(event.order_id):
                    v["status"] = "cancelled"

    def get(self, order_id: str) -> Optional[dict]:
        return self._views.get(order_id)

    def rebuild(self, event_store: ProductionEventStore) -> None:
        """Rebuild entire projection from all events — idempotent."""
        self._views.clear()
        for events in event_store.load_all_streams().values():
            for event in events:
                self.handle(event)
        print(f"  [Projection] OrderDetail rebuilt: {len(self._views)} orders")


class CustomerOrderHistoryProjection:
    """Order summaries per customer — for the customer's order history page."""

    def __init__(self) -> None:
        self._by_customer: dict[str, list[dict]] = {}

    def handle(self, event: OrderEvent) -> None:
        match event:
            case OrderPlacedEvent():
                total = sum(Decimal(i["unit_price"]) * i["qty"] for i in event.items)
                self._by_customer.setdefault(event.customer_id, []).append({
                    "order_id":  event.order_id,
                    "status":    "draft",
                    "total":     str(total),
                    "item_count": len(event.items),
                    "placed_at": event.occurred_at.isoformat(),
                })
            case PaymentConfirmedEvent() | OrderShippedEvent() | OrderDeliveredEvent() | OrderCancelledEvent():
                status_map = {
                    "PaymentConfirmedEvent": "paid",
                    "OrderShippedEvent":     "shipped",
                    "OrderDeliveredEvent":   "delivered",
                    "OrderCancelledEvent":   "cancelled",
                }
                new_status = status_map.get(type(event).__name__, "unknown")
                for orders in self._by_customer.values():
                    for o in orders:
                        if o["order_id"] == event.order_id:
                            o["status"] = new_status

    def get(self, customer_id: str) -> list[dict]:
        return sorted(
            self._by_customer.get(customer_id, []),
            key=lambda o: o["placed_at"], reverse=True
        )


class FulfillmentProjection:
    """Orders ready to ship — for the warehouse fulfillment queue."""

    def __init__(self) -> None:
        self._queue: dict[str, dict] = {}

    def handle(self, event: OrderEvent) -> None:
        match event:
            case PaymentConfirmedEvent():
                # We need the items — this projection needs OrderPlaced context
                # In production, this would join from a shared event context or separate read model
                self._queue[event.order_id] = {
                    "order_id":   event.order_id,
                    "payment_id": event.payment_id,
                    "paid_at":    event.occurred_at.isoformat(),
                    "status":     "ready_to_ship",
                }
            case OrderShippedEvent():
                self._queue.pop(event.order_id, None)
            case OrderCancelledEvent():
                self._queue.pop(event.order_id, None)

    def queue(self) -> list[dict]:
        return sorted(self._queue.values(), key=lambda o: o["paid_at"])

    def rebuild(self, event_store: ProductionEventStore) -> None:
        self._queue.clear()
        for events in event_store.load_all_streams().values():
            for event in events:
                self.handle(event)
        print(f"  [Projection] Fulfillment rebuilt: {len(self._queue)} orders in queue")


# ════════════════════════════════════════════════════════════════
# Command handler with snapshot-aware loading
# ════════════════════════════════════════════════════════════════

class OrderCommandHandler:
    def __init__(self, event_store: ProductionEventStore,
                 *projections) -> None:
        self._store       = event_store
        self._projections = projections

    def _load(self, order_id: str) -> Order:
        snapshot, events = self._store.load(order_id)
        if snapshot:
            # Restore from snapshot + replay remaining events
            order = Order()
            state = snapshot.state
            order.order_id        = state["order_id"]
            order.customer_id     = state["customer_id"]
            order.status          = OrderStatus(state["status"])
            order.items           = {
                sku: OrderItem(i["sku"], i["qty"], Decimal(i["unit_price"]))
                for sku, i in state["items"].items()
            }
            order.shipping_address = state["shipping_address"]
            order.payment_id       = state["payment_id"]
            order.tracking_number  = state["tracking_number"]
            order.version          = state["version"]
            # Replay events since snapshot
            for event in events:
                order._apply(event)
                order.version = event.version
            return order
        return Order.from_events(events)

    def _save(self, order: Order) -> None:
        events = order.collect_events()
        self._store.append(
            order.order_id, events,
            expected_version=order.version - len(events)
        )
        # Update all projections
        for event in events:
            for projection in self._projections:
                projection.handle(event)

    def place_order(self, customer_id: str,
                    items: list[dict]) -> str:
        order = Order.place(customer_id, items)
        self._save(order)
        print(f"  [Command] Order placed: {order.order_id}")
        return order.order_id

    def add_item(self, order_id: str, sku: str, qty: int,
                 unit_price: Decimal) -> None:
        order = self._load(order_id)
        order.add_item(sku, qty, unit_price)
        self._save(order)
        print(f"  [Command] Item added to {order_id}: {sku} x{qty}")

    def set_address(self, order_id: str, address: str) -> None:
        order = self._load(order_id)
        order.set_shipping_address(address)
        self._save(order)

    def confirm_payment(self, order_id: str, payment_id: str,
                        amount: Decimal) -> None:
        order = self._load(order_id)
        order.confirm_payment(payment_id, amount)
        self._save(order)
        print(f"  [Command] Payment confirmed: {order_id}")

    def ship_order(self, order_id: str, tracking: str,
                   carrier: str) -> None:
        order = self._load(order_id)
        order.ship(tracking, carrier)
        self._save(order)
        print(f"  [Command] Order shipped: {order_id}")

    def deliver_order(self, order_id: str) -> None:
        order = self._load(order_id)
        order.deliver()
        self._save(order)

    def cancel_order(self, order_id: str, reason: str) -> None:
        order = self._load(order_id)
        order.cancel(reason)
        self._save(order)


# ════════════════════════════════════════════════════════════════
# Demonstration
# ════════════════════════════════════════════════════════════════

if __name__ == "__main__":
    event_store = ProductionEventStore(snapshot_threshold=5)
    detail_proj = OrderDetailProjection()
    history_proj = CustomerOrderHistoryProjection()
    fulfill_proj = FulfillmentProjection()

    handler = OrderCommandHandler(
        event_store, detail_proj, history_proj, fulfill_proj
    )

    print("=" * 60)
    print("1. Place order")
    print("=" * 60)
    order_id = handler.place_order(
        customer_id="CUST-001",
        items=[
            {"sku": "LAPTOP-01",  "qty": 1, "unit_price": "1299.99"},
            {"sku": "MOUSE-01",   "qty": 2, "unit_price":   "29.99"},
        ]
    )

    print("\n" + "=" * 60)
    print("2. Add an item, set address")
    print("=" * 60)
    handler.add_item(order_id, "KEYBOARD-01", 1, Decimal("89.99"))
    handler.set_address(order_id, "42 Elm Street, Springfield, IL 62701")

    print("\n" + "=" * 60)
    print("3. Confirm payment")
    print("=" * 60)
    order = Order.from_events(event_store.load_all_events(order_id))
    total = order.total
    handler.confirm_payment(order_id, "PAY-abc123", total)

    print("\n" + "=" * 60)
    print("4. Query: Order detail (from projection)")
    print("=" * 60)
    detail = detail_proj.get(order_id)
    if detail:
        print(f"  Order:    {detail['order_id']}")
        print(f"  Status:   {detail['status']}")
        print(f"  Total:    ${detail['total']}")
        print(f"  Items:    {len(detail['items'])}")
        print(f"  Address:  {detail['shipping_address']}")

    print("\n" + "=" * 60)
    print("5. Ship order")
    print("=" * 60)
    handler.ship_order(order_id, "1Z999AA10123456784", "UPS")

    print("\n" + "=" * 60)
    print("6. Query: Fulfillment queue (should be empty)")
    print("=" * 60)
    queue = fulfill_proj.queue()
    print(f"  Orders in fulfillment queue: {len(queue)} (expected 0 — shipped)")

    print("\n" + "=" * 60)
    print("7. Deliver order")
    print("=" * 60)
    handler.deliver_order(order_id)

    print("\n" + "=" * 60)
    print("8. Full event log")
    print("=" * 60)
    for event in event_store.load_all_events(order_id):
        print(f"  v{event.version:02d} [{type(event).__name__:<28}] "
              f"at {event.occurred_at.strftime('%H:%M:%S')}")

    print("\n" + "=" * 60)
    print("9. Time-travel: state after event v3")
    print("=" * 60)
    past_events = [e for e in event_store.load_all_events(order_id)
                   if e.version <= 3]
    past_order  = Order.from_events(past_events)
    print(f"  State at v3: status={past_order.status.value}, "
          f"items={len(past_order.items)}, total={past_order.total}")

    print("\n" + "=" * 60)
    print("10. Rebuild all projections from event log (drop and replay)")
    print("=" * 60)
    detail_proj.rebuild(event_store)

    rebuilt = detail_proj.get(order_id)
    if rebuilt:
        print(f"  Rebuilt status: {rebuilt['status']} (expected: delivered)")

    print("\n" + "=" * 60)
    print("11. Domain rule: cannot cancel delivered order")
    print("=" * 60)
    try:
        handler.cancel_order(order_id, "Changed mind")
    except ValueError as e:
        print(f"  Expected error: {e}")

    print("\n" + "=" * 60)
    print("12. Customer order history")
    print("=" * 60)
    history = history_proj.get("CUST-001")
    for o in history:
        print(f"  {o['order_id']}: {o['status']} — ${o['total']}")

Real-World Use Cases

Event Sourcing appears wherever the history of changes is as important as the current state.

Financial systems: Every ledger entry, transfer, payment, and fee is an event. The balance is never stored — it is computed from the sum of events. GAAP accounting is inherently event-sourced: a double-entry bookkeeping system is an event log. Event Sourcing makes the software model match the accounting model exactly.

Audit-required domains: Healthcare (HIPAA), financial services (SOX, PCI-DSS), legal systems, and government applications require tamper-evident records of every change. Event Sourcing provides this structurally — there is no "overwrite" operation in the event log. Regulatory audit trails are not added as an afterthought; they are the system.

Collaborative editing: Google Docs, Figma, and similar tools are event-sourced at their core. Each keystroke, cursor move, or shape drag is an event. Conflict resolution, undo/redo, version history, and real-time collaboration are all natural consequences of storing operations rather than state.

E-commerce order lifecycle: An order passes through many states (draft, confirmed, paid, split, partially shipped, returned, refunded) with complex rules at each transition. Event Sourcing captures every state change as an event, enabling full lifecycle visibility, customer service tooling, and business analytics from the same event stream.

Supply chain and logistics: Inventory adjustments, shipment tracking, warehouse transfers, and customs events are all naturally events. The current inventory position is derived from all adjustment events. Historical position at any point in time is available without a separate history table.

Gaming: Player actions, game state changes, matchmaking events, achievement unlocks, and leaderboard updates are all events. Replaying events enables server-side anti-cheat verification, bug reproduction, and game replay features.

DevOps and infrastructure: Deployment events, configuration changes, scaling actions, and incident records are naturally event-sourced. Kubernetes etcd is essentially an event log. Infrastructure-as-code tools record every change as a versioned event.

Language-Specific Mistakes and Anti-Patterns

# BAD: Event is a regular dataclass — can be mutated after creation
@dataclass
class MoneyDepositedEvent:
    amount: Decimal
    account_id: str

event = MoneyDepositedEvent(Decimal("100"), "ACC-001")
event.amount = Decimal("999")   # ← BAD: event mutated after the fact
# An event is a fact about what happened — it cannot change.

# GOOD: frozen=True makes the dataclass immutable
@dataclass(frozen=True)
class MoneyDepositedEvent:
    amount:     Decimal
    account_id: str

event = MoneyDepositedEvent(Decimal("100"), "ACC-001")
# event.amount = Decimal("999")  ← FrozenInstanceError at runtime

# BAD: apply() has business logic — violates the apply/validate separation
def _apply(self, event: DomainEvent) -> None:
    match event:
        case MoneyWithdrawnEvent():
            # Business rule in apply() — WRONG
            if Decimal(event.amount) > self.balance:
                raise ValueError("Insufficient funds")   # ← never throw in apply()!
            self.balance -= Decimal(event.amount)

# apply() must be pure state reconstruction — it must NEVER fail.
# It runs during replay from the event store; events in the store are already
# valid. If apply() throws during replay, the aggregate cannot be reconstituted.

# GOOD: Business rules in domain methods; apply() is pure state update
def withdraw(self, amount: Decimal) -> None:
    # Business rule enforced HERE, before the event is created
    if amount > self.balance:
        raise ValueError("Insufficient funds")
    self._emit(MoneyWithdrawnEvent(amount=str(amount), ...))

def _apply(self, event: DomainEvent) -> None:
    match event:
        case MoneyWithdrawnEvent():
            self.balance -= Decimal(event.amount)   # pure state update; never throws

Frequently Asked Questions

Q1: What is the difference between Event Sourcing and storing a simple audit log?

An audit log is an add-on: it captures some fields, sometimes, after the fact, in a separate table. It is incomplete by design (you only log what you thought to log), inconsistent (different developers log different things), and not the source of truth (the real state is still in the main table).

Event Sourcing is architectural: the event log is the source of truth. Every state change is captured as an event — completely and consistently — because that is the only way state is stored. Nothing is overwritten; nothing is missing. The current state is a derived view of the complete event history. This is a fundamentally different guarantee.

Q2: Does Event Sourcing require CQRS?

No. They are independent patterns that complement each other well. You can use Event Sourcing without CQRS (the same model reads and writes, state is derived from events on every read) and you can use CQRS without Event Sourcing (commands use a traditional mutable state store; reads use projections built from state snapshots rather than event streams).

They pair naturally because Event Sourcing produces an event stream that CQRS projections can consume. The write side is Event Sourced; the read side is a collection of CQRS projections built from the event stream. Most full Event Sourcing implementations also use CQRS.

Q3: How do I query an event-sourced system efficiently?

You don't query the event log for reads — that would require replaying events for every query. Instead, you maintain projections (also called read models): denormalized views derived from the event stream and stored in a query-optimized store (a relational table, a document store, a search index). The projections are updated whenever new events are appended — synchronously in simple implementations, asynchronously via an event bus in production.

The key insight: projections are derived and disposable. They can be dropped and rebuilt from the event log at any time. You can add new projections that answer new queries without touching the event store or the aggregate.

Q4: What happens when event schemas need to change?

Events are immutable — you can't alter stored events. Evolve event schemas through these strategies:

Upcasting: when loading events, apply a transformation to bring old events to the current schema before they reach apply(). The event store holds the original; the aggregate sees the current version. This is the most common approach.

New event types: instead of changing OrderPlaced, add OrderPlacedV2 with the new fields. Handle both versions in apply(). Gradually retire the old event type once all streams have migrated.

Weak schema / optional fields: design events with JSON-flexible serialization where new fields default to sensible values if absent. Old events simply lack the new fields; apply() handles the absence gracefully.

Never change an event's meaning: you can add fields or rename events, but you should never change what an event means. PaymentConfirmed must always mean a payment was confirmed. Changing its meaning would corrupt the replay of all historical events.

Q5: How do I handle concurrency in an event-sourced system?

Through optimistic concurrency control: when appending events, you specify the expected current version of the stream. If another process has appended events since you loaded the stream, the append fails with a concurrency exception. You must reload the stream, reconstitute the aggregate with the new events, and retry if the command is still valid.

This is simpler than pessimistic locking (no deadlocks), and correct for most domain scenarios. For high-contention scenarios (thousands of commands per second on the same aggregate), consider partitioning the aggregate into smaller pieces or using a different consistency model.

Q6: What is a snapshot and when do I need one?

A snapshot is a serialized copy of an aggregate's state at a specific version. Instead of replaying all events from v1 every time, you load the latest snapshot + only the events since the snapshot. This is a performance optimization only — snapshots are not the source of truth; the event log remains authoritative.

You need snapshots when aggregates accumulate hundreds or thousands of events and replay performance becomes unacceptable. A practical threshold: if replay consistently takes more than 50ms, consider snapshots. Start without them; add them when profiling proves they are needed.

Snapshot policy: take a snapshot every N events (e.g., every 50 or 100 events). Store snapshots separately from the event log. Always fall back to full replay if no snapshot exists.

Q7: How do I delete data in an event-sourced system? (GDPR compliance)

This is the most challenging aspect of Event Sourcing. GDPR requires the ability to delete personal data on request. Options:

Crypto-shredding: encrypt personal data in events using a per-customer encryption key stored separately. Deleting the key makes the data unreadable — the event log is technically intact but the PII is inaccessible. This satisfies GDPR's "right to be forgotten" without deleting events.

Forget events: append a CustomerDataForgottenEvent that signals that personal data has been erased. Projections handle this event by nullifying personal fields. The event log still has the business facts (transactions occurred) but the personal identifiers are gone.

Separate PII storage: store personal data in a separate GDPR-compliant store with a reference key. Events contain the reference key, not the data. Deleting the PII record satisfies GDPR; events are unaffected.

Accepted limitations: for strict Event Sourcing purists, physical deletion is not possible. Crypto-shredding is the cleanest solution that preserves Event Sourcing guarantees while achieving GDPR compliance.

Q8: Is Event Sourcing appropriate for every aggregate?

No. Event Sourcing adds significant complexity and should be applied selectively. Apply it to aggregates where:

• The history of changes has business value (orders, transactions, approvals)
• Audit trails are required
• Multiple projections need to be built or rebuilt independently
• The aggregate has complex state transitions that benefit from traceability
• Integration with other systems via the event stream is valuable

Do not apply Event Sourcing to simple lookup tables (countries, currencies, static reference data), high-volume low-value data (click events, telemetry), or aggregates with trivial state that never needs audit. A selective approach — Event Sourcing for core domain aggregates, simple CRUD for supporting domains — is the practical choice in most systems.

Q9: How does Event Sourcing interact with distributed transactions?

Event Sourcing avoids the need for distributed transactions in many cases. Each command targets one aggregate; one aggregate = one transaction = one event store append. Cross-aggregate operations are handled through the Saga pattern: a saga listens to domain events and issues compensating commands. If step 2 fails, the saga issues undo commands. This is eventual consistency rather than two-phase commit — it requires designing operations to be compensatable, but eliminates the distributed transaction problem.

For strict consistency across aggregates (rare in well-designed domain models), consider whether the aggregates should really be one aggregate, or whether the consistency requirement is a modelling mistake.

Q10: What are the best Event Store implementations?

EventStoreDB (Greg Young's purpose-built event store): the reference implementation. Purpose-built for Event Sourcing, supports projections natively, has excellent .NET and Go client libraries.

Apache Kafka: a distributed log. Not purpose-built for Event Sourcing but works well as one. Excellent for high throughput and cross-service event streaming. Consumer groups for projections. Retention policies require care (events should be retained indefinitely).

PostgreSQL with an events table: pragmatic and simple. An events table with (stream_id, version, event_type, data, occurred_at) supports optimistic concurrency with a unique constraint on (stream_id, version). No additional infrastructure. Works well for most applications.

AWS DynamoDB: each item is (pk=stream_id, sk=version). Conditional writes enforce optimistic concurrency. DynamoDB Streams can trigger projection updates via Lambda. Scales horizontally without infrastructure management.

Azure Cosmos DB / MongoDB: document stores where each event is a document, partitioned by stream_id. Optimistic concurrency via _etag. Change streams / change feed for projection updates.

Key Takeaways

🎯 Core Concept: Event Sourcing stores the state of an application as an append-only, ordered log of immutable domain events — not as a snapshot of current values. Every state change is a fact appended to the stream: AccountOpened, MoneyDeposited, MoneyWithdrawn. The current state is derived on demand by replaying these events in order through the aggregate's apply() method. The event log is the single source of truth; projections (read models) are derived views that can be dropped and rebuilt at any time by replaying the log. This gives you a complete, tamper-evident history of every change ever made — for free, as a structural property of how the system works.

🔑 Key Benefits:

• Complete audit trail: every state change is an immutable, ordered fact — no separate audit log required, no possibility of gaps or inconsistency
• Time-travel queries: "What was the state of this order on March 15th?" — replay events up to that timestamp; impossible with snapshot storage
• Projection rebuilding: adding a new read model means subscribing to the event stream and building a new projection from scratch — no schema migration, no data transformation job
• Debugging and forensics: reproduce any bug by replaying the event stream that produced it; step through events one by one to find where things went wrong
• Event-driven integration: the event log is a native pub/sub integration bus — other services subscribe to your events without any special infrastructure
• Domain model alignment: in domains where history is the business (banking, legal, healthcare), Event Sourcing makes the storage model match the domain model

🌐 Language-Specific Highlights:

• Python: @dataclass(frozen=True) for immutable events; match/case for clean apply() dispatch; generator-based replay for memory-efficient projection rebuilding; structural pattern matching exhaustiveness requires case _: raise to catch unhandled events
• TypeScript: discriminated union types (type BankAccountEvent = AccountOpenedEvent | MoneyDepositedEvent | ...) for type-safe event handling; Object.freeze() for runtime immutability; TypeScript never type for exhaustiveness checking in switch statements
• Java: sealed interface + record for the event hierarchy — closed, immutable, compile-time exhaustive; switch expressions with sealed types enforce that every event is handled; Java records provide structural equality needed for event deduplication
• JavaScript: private class fields (#state) protect aggregate internals; Object.freeze() on all events; array spread for immutable event log updates; functional reconstitution with reduce() as an alternative to class-based aggregates
• C#: abstract record hierarchy for events — immutable value semantics, structural equality, destructuring; switch expression with BankAccountEvent exhaustiveness; ImmutableList<T> for the event log; records are perfect for events because they are values, not objects
• PHP: readonly class (PHP 8.2+) for immutable events; match expression for clean event dispatch; JSON serialization (not serialize()) for durable, language-agnostic event storage; explicit version sequence validation during reconstitution
• Go: interface-based events with a common getBase() method for metadata access; type switch for apply() dispatch; sync.RWMutex for concurrent stream access; function-based projectors are idiomatic Go over class-based ones
• Rust: enum for the event hierarchy — sealed by definition, exhaustive match enforced by the compiler; #[derive(Debug, Clone)] on all event variants; std::mem::take(&mut self.pending) for efficient event collection; Box<dyn Error> for command result errors
• Dart: abstract event classes with const constructors; pattern matching via if (event is SpecificEvent) chains (Dart 3 adds proper pattern matching); List.unmodifiable() for the pending events collection; DateTime from the event (not DateTime.now()) in all projection handlers
• Swift: struct (value type) for all events — copied on assignment, naturally immutable; protocol-based event hierarchy with any BankAccountEventSwift; switch over protocol conformance; private var pending: [any BankAccountEvent] in the aggregate class
• Kotlin: sealed class + data class for events — closed hierarchy, immutable fields (val), structural equality, copy() for derived variants; exhaustive when expression for apply() — compiler error if any event subclass is unhandled; Instant for timestamps; runCatching for command error handling

📋 The Immutable Event Rules:

1. Events are facts about what happened — they cannot be changed or deleted
2. apply() is pure state reconstruction — it must never throw, never have side effects, and always produce the same state from the same events
3. Business rules belong in domain methods (validated before events are created), never in apply()
4. Event fields must be self-contained — enough data to reconstruct the state change without external context
5. Events use past tense, domain language: OrderPlaced, PaymentFailed, UserEmailChanged — not SetStatus or UpdateEmail

⚠️ Anti-Patterns to Avoid:

• Mutable events (any var field, non-frozen dataclass, class reference types for events) — events are immutable facts
• Business logic in apply() — apply() is pure state update; it must never throw
• Storing current state instead of events ("saving the aggregate" instead of "appending events")
• Querying the event log directly for reads — use projections; never scan events for query results
• Non-deterministic projection handlers (using DateTime.now(), random values, external API calls in handlers) — projections must produce the same result on every rebuild
• Missing optimistic concurrency check on event append — without it, concurrent writes produce corrupted state
• PHP serialize() for event storage — use JSON; events must be readable by future versions and potentially other languages
• Go event version gaps silently accepted during reconstitution — validate version continuity

🛠️ Best Practices Across Languages:

1. Make events self-describing: include all data needed to understand what happened, including occurred_at, event_id, version, and enough domain context for a human to read the log
2. Design apply() as a pure function: given the same sequence of events, apply() must always produce the same state — no randomness, no external calls, no throwing
3. Enforce optimistic concurrency: always pass expected_version when appending events; treat conflicts as normal flow, reload, and retry
4. Start without snapshots: add snapshot optimization only when profiling proves replay performance is unacceptable
5. Rebuild projections from scratch: design every projection to be fully rebuildable by replaying all events — this is the escape hatch for any projection bug
6. Use a type discriminator in serialized events: store the event type as a string alongside the serialized data; never rely on language-specific serialization mechanisms
7. Version your events from the start: event versioning is hard to retrofit; name events with explicit versions (OrderPlacedV2) or design your upcasting infrastructure early

💡 Event Sourcing + Related Patterns:

• Event Sourcing + CQRS: the most common combination — Event Sourcing provides the event stream; CQRS projections consume the stream to build read models. Each side is independently optimized
• Event Sourcing + Domain-Driven Design: aggregates produce domain events that are the atoms of the event store; bounded contexts communicate via their event streams; ubiquitous language is preserved in event names
• Event Sourcing + Saga: sagas listen to domain events and issue commands to coordinate long-running processes across multiple aggregates — the event-driven complement to the transactional command
• Event Sourcing + Snapshot: snapshots reduce replay overhead for long-lived aggregates; always a performance optimization, never a replacement for the event log
• Event Sourcing + Event-Driven Architecture: the event log is a native integration bus; other services subscribe to domain events and build their own projections or trigger their own workflows

This guide pairs directly with:

• CQRS — the most natural companion to Event Sourcing; read-side projections consume the event stream
• Domain-Driven Design — aggregates, domain events, and bounded contexts are the building blocks of event-sourced systems
• Event-Driven Architecture — the infrastructure layer that transports events between services
• Saga Pattern — long-running process orchestration built on domain events
• Repository Pattern — the write-side repository is replaced by the event store in event-sourced systems

Each guide includes examples in all 11 programming languages with language-specific best practices.