Event Sourcing — Event Store, Replay & Audit Trail Guide
In this tutorial, you'll learn about Event Sourcing. We cover key concepts, practical examples, and best practices.
Event sourcing stores every state change as an append-only event in an event store, with the current state derived by replaying events — providing a complete audit trail, temporal query capability, and the ability to rebuild any system state from history.
Why Event Sourcing Matters
Traditional databases store only the current state. If a user changes their address, the old address is gone. Event sourcing preserves every change forever. Financial systems have always done this — every transaction is logged because you can't erase money. Event sourcing generalizes this principle. EventStoreDB processes millions of events per second at companies like PayPal and Jet.com. Doda Browser's tab sync uses event sourcing — every tab open, close, and move is an event, enabling perfect cross-device synchronization.
Plain-Language Explanation
Think of a bank statement. It doesn't show your current balance and nothing else. It shows every deposit and withdrawal, in order, from the account's opening. Your current balance is simply the sum of all those transactions. If the bank made an error, they don't just fix the balance — they add a correcting transaction. The statement is the event log. The balance is the projected state. Event sourcing is the same idea for all application data.
graph LR
subgraph "Write Path"
CMD[Command] --> VAL[Validate]
VAL --> STORE[(Event Store
Append-Only Log)]
end
subgraph "Read Path"
STORE --> PROJ[Projection Engine]
PROJ --> READ[(Read Model
Materialized View)]
end
subgraph "Rebuild"
STORE --> REPLAY[Replay All Events]
REPLAY --> NEW[New Read Model]
end
AUDIT[Audit Query] --> STORE
style STORE fill:#e67e22,color:#fff
style PROJ fill:#3498db,color:#fff
style REPLAY fill:#27ae60,color:#fff
Event Store Implementation
A simple event store with PostgreSQL:
CREATE TABLE events (
id BIGSERIAL PRIMARY KEY,
aggregate_type VARCHAR(128) NOT NULL,
aggregate_id VARCHAR(64) NOT NULL,
event_type VARCHAR(128) NOT NULL,
event_data JSONB NOT NULL,
version INT NOT NULL,
created_at TIMESTAMP DEFAULT NOW(),
UNIQUE(aggregate_type, aggregate_id, version)
);
CREATE INDEX idx_events_aggregate ON events(aggregate_type, aggregate_id);
CREATE INDEX idx_events_created ON events(created_at);
import json, time
from dataclasses import dataclass
@dataclass
class Event:
aggregate_id: str
event_type: str
data: dict
version: int
timestamp: float
class EventStore:
def __init__(self, db):
self.db = db
def append(self, aggregate_type: str, aggregate_id: str, events: list[Event]):
for event in events:
self.db.execute(
"INSERT INTO events (aggregate_type, aggregate_id, event_type, event_data, version, created_at) "
"VALUES (:atype, :aid, :etype, :edata, :ver, :ts)",
{"atype": aggregate_type, "aid": aggregate_id, "etype": event.event_type,
"edata": json.dumps(event.data), "ver": event.version, "ts": event.timestamp},
)
def get_events(self, aggregate_type: str, aggregate_id: str) -> list[Event]:
rows = self.db.query(
"SELECT * FROM events WHERE aggregate_type = :atype AND aggregate_id = :aid ORDER BY version",
{"atype": aggregate_type, "aid": aggregate_id},
)
return [Event(r["aggregate_id"], r["event_type"], json.loads(r["event_data"]), r["version"], r["created_at"]) for r in rows]
Event Replay for State Rebuild
Rebuild aggregate state from its event stream:
class AccountAggregate:
def __init__(self):
self.balance = 0
self.version = 0
def apply(self, event: Event):
if event.event_type == "AccountOpened":
self.balance = event.data["initial_balance"]
elif event.event_type == "MoneyDeposited":
self.balance += event.data["amount"]
elif event.event_type == "MoneyWithdrawn":
self.balance -= event.data["amount"]
self.version = event.version
@classmethod
def replay(cls, events: list[Event]):
account = cls()
for event in events:
account.apply(event)
return account
events = [
Event("acc-1", "AccountOpened", {"initial_balance": 1000}, 1, time.time()),
Event("acc-1", "MoneyDeposited", {"amount": 500}, 2, time.time()),
Event("acc-1", "MoneyWithdrawn", {"amount": 200}, 3, time.time()),
]
account = AccountAggregate.replay(events)
print(f"Balance: ${account.balance}, Version: {account.version}")
Expected output:
Balance: $700, Version: 3
Snapshot Strategy
Replaying millions of events is slow. Take periodic snapshots:
class SnapshotStore:
def save_snapshot(self, aggregate_id: str, state: dict, version: int):
self.db.execute(
"INSERT INTO snapshots (aggregate_id, state, version) "
"VALUES (:aid, :state, :ver) "
"ON CONFLICT (aggregate_id) DO UPDATE SET state = :state, version = :ver",
{"aid": aggregate_id, "state": json.dumps(state), "ver": version},
)
def load_snapshot(self, aggregate_id: str) -> tuple[dict, int]:
row = self.db.query_one(
"SELECT state, version FROM snapshots WHERE aggregate_id = :aid",
{"aid": aggregate_id},
)
if row:
return json.loads(row["state"]), row["version"]
return {}, 0
Common Mistakes
Schema evolution: Events are immutable. When the event schema changes, old events don't change. All projection code must handle all event versions.
Event explosion: Storing every mouse click as an event generates terabytes. Store meaningful domain events, not UI interactions.
No snapshots: Without snapshots, rebuilding an aggregate with 1M events takes minutes. Snapshot every N events (100-1000).
Single event store bottleneck: In high-throughput systems, the event store can become a write bottleneck. Partition by aggregate type.
Ignoring event versioning: Event types evolve. Include a version number in every event and write migration functions for projections.
Practice Questions
What is event sourcing? Storing all state changes as an append-only event log. The current state is derived by replaying events, enabling audit trails and temporal queries.
How does event sourcing differ from CQRS? Event sourcing is about how you store state (as events). CQRS is about how you read state (separate read/write models). They work well together but are independent.
What is a snapshot and why is it needed? A saved copy of an aggregate's state at a specific version. Without snapshots, replaying millions of events for every read is prohibitively slow.
How do you handle event schema changes? New event types can have new fields. Old events must still be readable. Use upcasters — functions that convert old event formats to the current schema on the fly.
When should you NOT use event sourcing? For simple CRUD applications where audit trails aren't needed. The complexity of event versioning and projection management isn't justified.
Mini Project
Build a bank account event sourcing system:
import json, time
events = []
class BankAccount:
def __init__(self, owner: str):
event = {"type": "AccountOpened", "owner": owner, "balance": 0}
events.append(event)
def deposit(self, amount: float):
events.append({"type": "Deposited", "amount": amount})
def withdraw(self, amount: float):
events.append({"type": "Withdrawn", "amount": amount})
@staticmethod
def get_balance() -> float:
balance = 0
for e in events:
if e["type"] == "AccountOpened":
balance = 0
elif e["type"] == "Deposited":
balance += e["amount"]
elif e["type"] == "Withdrawn":
balance -= e["amount"]
return balance
@staticmethod
def get_statement() -> list:
return [e for e in events]
acc = BankAccount("Alice")
acc.deposit(1000)
acc.withdraw(300)
acc.deposit(500)
print(f"Balance: ${BankAccount.get_balance()}")
print(f"Events: {len(events)}")
Expected output:
Balance: $1200
Events: 4
Cross-References
- CQRS Pattern
- Saga Pattern
- Event-Driven Architecture
- Key-Value Store Design
- API Design
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro