Event Sourcing and CQRS skill for Codex

Done well, the event stream is the source of truth, every state change is a named business fact appended with a concurrency check, read models can be dropped and rebuilt from the stream at any time, and the team knew why it paid this complexity cost before starting.

Start with “should we?”

Signal	Points to
Audit history is a legal or business requirement (ledgers, claims, trading)	Event sourcing fits
Business asks “what did we know at time T?” or wants to replay with new rules	Event sourcing fits
Domain naturally thinks in events (bookings, payments, shipments)	Event sourcing fits
Read and write shapes differ hugely, reads dominate	CQRS alone (read models from a normal DB or CDC)
CRUD with an audit log requirement	Normal tables + audit/history table or CDC, not ES
Team new to DDD and messaging, deadline in weeks	Do not start with ES
Need ad-hoc queries over current state everywhere	ES makes this harder; needs projections for each

CQRS does not require event sourcing, and event sourcing does not require separate databases. Apply ES to one core aggregate type where it pays, not the whole system.

Store choice

Option	Pick when
PostgreSQL table you own	Moderate volume, team knows Postgres, want one database
Marten (.NET on Postgres)	.NET stack; gives streams, projections, async daemon
KurrentDB (formerly EventStoreDB)	Many streams, subscriptions and projections as first-class features
Axon Framework + Axon Server	JVM, want framework-managed aggregates and sagas
Kafka as the event store	Avoid: no per-stream optimistic concurrency, awkward per-aggregate reads

A Postgres event store

CREATE TABLE events (
  global_position bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
  stream_id       text        NOT NULL,           -- e.g. 'account-7f3c'
  stream_version  integer     NOT NULL,           -- 1, 2, 3 ... per stream
  event_id        uuid        NOT NULL UNIQUE,
  event_type      text        NOT NULL,           -- 'FundsDeposited'
  schema_version  smallint    NOT NULL DEFAULT 1,
  data            jsonb       NOT NULL,
  metadata        jsonb       NOT NULL DEFAULT '{}', -- causation_id, correlation_id, user, trace
  recorded_at     timestamptz NOT NULL DEFAULT now(),
  UNIQUE (stream_id, stream_version)
);

Append with optimistic concurrency: the unique constraint is the check.

def append(conn, stream_id: str, expected_version: int, new_events: list[Event]) -> int:
    """expected_version = version the command handler loaded (0 for a new stream)."""
    with conn.transaction():
        for i, e in enumerate(new_events, start=1):
            try:
                conn.execute(
                    "INSERT INTO events (stream_id, stream_version, event_id, event_type, data, metadata) "
                    "VALUES (%s, %s, %s, %s, %s, %s)",
                    (stream_id, expected_version + i, e.id, e.type, e.json(), e.meta_json()),
                )
            except UniqueViolation as exc:
                raise ConcurrencyConflict(stream_id, expected_version) from exc
    return expected_version + len(new_events)

On ConcurrencyConflict, reload the stream and re-run the command (retry 2-3 times) or return 409 to the caller.

Aggregates from events

class Account:
    def __init__(self):
        self.version = 0
        self.balance = Decimal("0")
        self.closed = False
        self.pending: list[Event] = []

    @classmethod
    def load(cls, events):
        acc = cls()
        for e in events:
            acc._apply(e)
            acc.version += 1
        return acc

    def withdraw(self, amount: Decimal) -> None:          # command: decide, then record
        if self.closed:
            raise DomainError("Account closed")
        if amount > self.balance:
            raise DomainError("Insufficient funds")
        self._record(FundsWithdrawn(amount=amount))

    def _record(self, e):
        self._apply(e)
        self.pending.append(e)

    def _apply(self, e):                                  # evolve: no validation, no I/O
        match e:
            case FundsDeposited(amount=a): self.balance += a
            case FundsWithdrawn(amount=a): self.balance -= a
            case AccountClosed(): self.closed = True

Rules: _apply never throws and never does I/O (old events must always replay); decisions happen in command methods; events are past-tense facts with the data needed to apply them, not commands or diffs of the whole object.

Projections (the CQRS read side)

• Each projection keeps a checkpoint (last processed global position) in the same transaction as its writes, so a crash replays at most the uncommitted batch.
• Projections are idempotent: upsert by key, or compare the stream version stored on the read row and skip older events.
• Rebuild by creating a new table/version (account_summary_v2), replaying from position 0, then switching reads. Never rebuild in place under live traffic.
• Expose read-model lag (head position minus checkpoint) as a metric and alert on it.
• UI after a command: return the new version from the write and either read from the write model for that screen, or poll the read model until it reaches that version.

Gap gotcha with sequence-based positions: in Postgres, a transaction that obtained a lower identity value can commit after one with a higher value. A projection reading WHERE global_position > checkpoint can skip it forever. Mitigate by reading only up to a safe high-water mark (positions older than a short delay, or below the oldest in-flight transaction via pg_snapshot_xmin(pg_current_snapshot()) tracked per row with xid8), or serialise appends. Marten and KurrentDB handle this for you; a home-made store must.

Snapshots

Add only when measurement shows load time matters (commonly streams over a few hundred events). Store {stream_id, version, state, snapshot_schema}; load snapshot then events after its version. Snapshots are a cache: delete and regenerate them whenever the aggregate’s state shape changes. Long streams often signal a modelling issue: consider closing a stream per period (“account-7f3c-2026”) with a carried-forward opening event.

Changing events over time

Stored events are immutable. Evolve with:

Change	Technique
Add a field	New optional field with a default when absent
Rename/restructure	Upcaster: function v1 -> v2 applied on read, keyed by event_type + schema_version
Meaning changed	New event type (FundsWithdrawnV2 or a better name); keep handling the old
Bad data written by a bug	Append a compensating/correcting event; do not edit history

Rewriting the store (copy-transform to a new store) is a last resort for large migrations; plan it like a database migration with verification and a rollback.

Personal data and erasure

Events are forever, erasure requests are not optional. Keep PII out of events where you can (reference a profile record by ID). Where unavoidable, encrypt PII fields with a per-subject key and delete the key on erasure (crypto-shredding). Have a lawyer or privacy officer confirm the approach meets your obligations.

Workflow

- [ ] 1. Justify ES for the specific aggregate (table above); record an ADR
- [ ] 2. Event-storm the aggregate: commands, events, invariants
- [ ] 3. Choose the store; create schema with (stream_id, stream_version) uniqueness
- [ ] 4. Implement decide/apply aggregate; append with expected version
- [ ] 5. Build projections with checkpoints, idempotent writes, lag metric
- [ ] 6. Add upcasting hook and schema_version from day one
- [ ] 7. Write rebuild and replay tooling before production
- [ ] 8. Plan PII handling and erasure

Gotchas

• Publishing the event store’s internal events directly as integration events couples every consumer to your aggregate internals. Map to separate integration events.
• Reading other aggregates’ state inside _apply makes replay depend on the present.
• Projections that call external APIs cannot be rebuilt safely. Keep side effects in separate reactors with their own idempotency.
• Never delete or update rows in the events table to “fix” something without explicit user confirmation and a written plan; it destroys the audit trail ES exists to provide.
• Set-based invariants (unique usernames) cannot be checked inside one stream; use a reservation table with a unique constraint or a dedicated stream per unique value.

Verification

• Given/when/then tests: given past events, when a command, then expected new events or error.
• Concurrency test: two writers with the same expected version; exactly one succeeds.
• Rebuild test: drop a projection, replay, and compare to the live one (row counts and checksums).
• Replay of the full production stream (on a copy) succeeds with current code and upcasters.

Output contract

**Scope:** event sourcing for <aggregate> only; rest of system stays state-based (ADR-00xx)
**Store:** <choice>, schema <path>, concurrency via unique (stream_id, stream_version)
**Events:** <list with schema_version>
**Projections:** <name -> table, checkpoint, rebuild command, lag metric>
**Snapshots:** none | every <n> events (measured load time <x> ms before)
**PII strategy:** <reference-only | crypto-shredding>, reviewed by <who>