Event sourcing and CQRS

In a normal (state-based) design, a row holds the current state and updates overwrite it: balance = 750. In event sourcing, you store the facts that happened, in order: AccountOpened, MoneyAdded 1000, MoneySpent 250. The current state is the result of replaying the events. Events are never updated or deleted; mistakes are fixed with new compensating events (MoneyRefunded), just like an accountant's ledger.

Events are grouped into streams, usually one per aggregate (one wallet, one order, one cart). To handle a command, you load the aggregate's events (or a recent snapshot plus newer events), rebuild its state, check business rules, and append new events. Concurrency is controlled per stream with an expected version: 'append version 18 only if the latest is 17'.

CQRS (Command Query Responsibility Segregation) means using different models for writes and reads. Commands go to a write model that enforces rules. Queries go to read models (projections) shaped for each screen: a balance table in Postgres, a search index in Elasticsearch, a leaderboard in Redis. Projections are built by consumers that read the event stream, so they are eventually consistent, typically milliseconds to seconds behind.

The two are separate ideas. You can use CQRS without event sourcing (a normal database plus CDC into read models), and event sourcing almost always needs some CQRS, because replaying events for every query is too slow. A Kafka topic is a good transport for events but a weak primary event store: it does not give per-aggregate optimistic concurrency or efficient loading of one aggregate's events.

1. 1Design events as past-tense business facts with the data consumers need: OrderPlaced {order_id, items, total}, PaymentCaptured {payment_id, amount}. Avoid 'RowUpdated'-style events.
2. 2Store them in an event store: EventStoreDB, a Postgres table with a unique (stream_id, version), or DynamoDB with a conditional write on version. Appends are atomic per stream.
3. 3Handle a command: load the stream (or latest snapshot + later events), fold events into state, validate, and append new events with the expected version. On a version conflict, reload and retry.
4. 4Snapshot long streams: every N events (for example 100), store the folded state, so loading reads one snapshot plus at most 99 events.
5. 5Publish events to consumers: subscribe to the store's feed, or use an outbox/CDC into Kafka keyed by stream id.
6. 6Build projections: each consumer applies events to its own read model, idempotently, tracking the last position it applied. A new screen means a new projection, built by replaying history from the start.
7. 7Handle read-after-write: the read model lags. Return the new state from the command, read the write model for the user's own latest change, or wait until the projection reaches the returned version.
8. 8Evolve schemas: events are stored forever, so add a schema version and upcast old events to the new shape when reading; never rewrite history in place.
9. 9Handle personal data: keep personal data out of events where possible, or encrypt it per user and delete the key to 'forget' the user (crypto-shredding).

-- Event store: one append-only table; (stream_id, version) is unique.
CREATE TABLE events (
  stream_id   text        NOT NULL,          -- e.g. 'wallet-42'
  version     int         NOT NULL,          -- 1, 2, 3 ... per stream
  event_type  text        NOT NULL,          -- 'MoneyAdded', 'MoneySpent'
  payload     jsonb       NOT NULL,
  schema_ver  int         NOT NULL DEFAULT 1,
  created_at  timestamptz NOT NULL DEFAULT now(),
  PRIMARY KEY (stream_id, version)
);

-- Append with optimistic concurrency: the caller loaded version 17,
-- decided the command is valid, and appends version 18.
-- If another writer already appended 18, the primary key rejects this
-- insert, and the caller reloads and retries the command.
INSERT INTO events (stream_id, version, event_type, payload)
VALUES ('wallet-42', 18, 'MoneySpent', '{"amount": 250, "order_id": "ord_981"}');