CQRS & Event Sourcing

The 30-second version

They're two separate ideas that happen to work well together. You can use either one without the other, and most systems need neither. This topic shows what each one buys you, what it costs, and how to tell when it's worth it.

Two problems from one small bank

Picture a small bank app with a balance column on the accounts table. One morning, two things go wrong:

1. A balance nobody can explain. Ana's account says $7. She swears it was $130 yesterday. The row has one number in it, last written at 03:12. Was it a night job? A bug? An admin? The database can't say, because each UPDATE erased the number before it.
2. A dashboard that slows everyone down. The branch managers' dashboard runs one SQL query that joinsJOINSQL that combines rows from two tables by matching columns — e.g. each order with its user's name: … FROM orders JOIN users ON users.id = orders.user_id. nine tables and adds up a month of transactions. While it runs, withdrawals on the same tables slow to a crawl.

Event sourcing answers the first problem. CQRS answers the second.

Try the first picture. Both sides below are the same account; the left keeps one number, the right keeps the logbook. Click a few actions, including the night job, then press Why?

The logbook costs more: more rows, and a sum to compute. In return it can answer "why?" and "what was it on Tuesday?" — questions the whiteboard has thrown away.

What's in this topic

• §02–§03: CQRS on its own. CommandsCommandA request to change something, named as an order in the present tense: Withdraw, PlaceOrder, ChangeAddress. The system checks its rules first and may reject it. vs queriesQueryA request that only reads and returns data, like “show my balance”. Asking it twice changes nothing., separate read tables, real Go and JavaScript code. No events needed.
• §04–§06: event sourcing. Events as facts, rebuilding state by replaying them, and a real Postgres event store that refuses conflicting writes.
• §07–§08: read models built from events, and the delay between a write and the screens showing it.
• §09–§11: living with it for years: snapshots, changing event formats, deleting personal data, telling other services, and when not to use any of this.

Commands, queries, and two models

Commands and queries

A CommandCommandA request to change something, named as an order in the present tense: Withdraw, PlaceOrder, ChangeAddress. The system checks its rules first and may reject it. asks the system to change something. It's named like an order: Withdraw, OpenAccount, ChangeAddress. The system checks its business rules first, so a command can be refused ("insufficient funds").

A QueryQueryA request that only reads and returns data, like “show my balance”. Asking it twice changes nothing. asks a question: "what's Ana's balance?", "show the branch dashboard". It never changes anything, so asking twice is harmless.

In a web app this split usually lines up with HTTP methods: commands are POST/PUT/DELETE, queries are GET. The idea is old: Command–query separationCommand–query separation (CQS)A rule for single functions, from Bertrand Meyer (1988): a function either changes something (a command) or returns an answer (a query), never both. CQRS applies the same split to whole models. (1988) said each function should be one or the other. CQRS (Command Query Responsibility Segregation, named by Greg Young around 2010) applies the same split to whole models.

Why one model gets uncomfortable

In a typical CRUDCRUDCreate, Read, Update, Delete: an app that mostly stores forms in tables and shows them back, with few business rules in between. app, the same classes and tables serve both sides. That's fine until the two sides want different things:

When one model tries to serve both, you get the bank's morning: a dashboard query that joins nine tables because the data is stored the way writes like it, running on the same rows that withdrawals need.

Two models

CQRS gives each side its own model:

• The Write modelWrite modelThe part of a CQRS system that handles commands: it loads the current state, checks the business rules and records the change. It's shaped for correctness, not for screens. handles commands. It loads what it needs, checks the rules and saves the change. It's shaped for correctness.
• One or more read modelsRead modelA copy of the data shaped for fast reading (a table, a search index, a cache), built from the events or main records. It can be thrown away and rebuilt. answer queries. Each is a copy of the data shaped for one screen, often a single flat table. It's shaped for speed, and it has no rules because it never changes anything on its own.

Something has to keep the read models up to date when the write side changes. You choose how, and that choice sets most of the cost. Step through the three options:

What CQRS is not

• Not event sourcing. You can build read tables from plain UPDATEs, as §03 does. Event sourcing (§04) is a different decision.
• Not a whole-system architecture. Apply it to the one area that hurts (the dashboard, the search page), not to every table.
• Not free. Every extra read model is code that must be kept in sync, and every separate store adds a delay between a write and the screens that show it.

CQRS without event sourcing, in code

This is the "CQRS, one DB" option from §02, built for real. No events, no new infrastructure: just one extra table and the discipline to keep it up to date.

The tables

The write side is normalizedNormalizationDesigning tables so every fact is stored in exactly one place (a product's name lives only in products), and other tables point at it by key.: a customer's name lives only in customers, a branch's name only in branches. The read side copies those names into one flat row per account. Copying data on purpose like this is called DenormalizationDenormalizationDeliberately copying data into several places (e.g. the product name into every order) so reads need fewer lookups — at the price of having to update every copy..

-- WRITE side: normalized, every fact in one place
CREATE TABLE branches  (id int PRIMARY KEY, name text NOT NULL);
CREATE TABLE customers (id int PRIMARY KEY, name text NOT NULL, branch_id int NOT NULL REFERENCES branches);
CREATE TABLE accounts  (id int PRIMARY KEY, customer_id int NOT NULL REFERENCES customers,
                        balance_cents bigint NOT NULL DEFAULT 0 CHECK (balance_cents >= 0));
CREATE TABLE ledger    (id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
                        account_id int NOT NULL REFERENCES accounts,
                        amount_cents bigint NOT NULL, created_at timestamptz NOT NULL DEFAULT now());

-- READ side: one flat row per account, shaped exactly like the dashboard
CREATE TABLE account_summaries (
  account_id    int PRIMARY KEY,
  owner_name    text   NOT NULL,   -- copied from customers
  branch_name   text   NOT NULL,   -- copied from branches
  balance_cents bigint NOT NULL,
  tx_count      int    NOT NULL,
  last_activity timestamptz
);
-- the dashboard filters by branch and sorts by balance: give it an index for exactly that
CREATE INDEX ON account_summaries (branch_name, balance_cents DESC);

Money is stored in cents as whole numbers (bigint), never as floating-point dollars, because 0.1 + 0.2 isn't exactly 0.3 in floating point.

The command side and the query side

// ---- COMMAND side: change state, return (almost) nothing ----
func deposit(ctx context.Context, db *sql.DB, accountID int, cents int64) error {
	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer tx.Rollback() // undoes everything unless we reach Commit

	// 1. The write model: normalized tables (the rules would go here too).
	if _, err := tx.ExecContext(ctx,
		`UPDATE accounts SET balance_cents = balance_cents + $2 WHERE id = $1`, accountID, cents); err != nil {
		return err
	}
	if _, err := tx.ExecContext(ctx,
		`INSERT INTO ledger (account_id, amount_cents) VALUES ($1, $2)`, accountID, cents); err != nil {
		return err
	}
	// 2. The read model: keep the dashboard's flat row up to date, same transaction.  ← the point
	if _, err := tx.ExecContext(ctx, `
		UPDATE account_summaries
		SET balance_cents = balance_cents + $2, tx_count = tx_count + 1, last_activity = now()
		WHERE account_id = $1`, accountID, cents); err != nil {
		return err
	}
	return tx.Commit()
}

// ---- QUERY side: read, never change anything ----
type Summary struct {
	AccountID    int        `json:"accountId"`
	Owner        string     `json:"owner"`
	BalanceCents int64      `json:"balanceCents"`
	TxCount      int        `json:"txCount"`
	LastActivity *time.Time `json:"lastActivity"` // nil when there's no activity yet
}

func branchDashboard(ctx context.Context, db *sql.DB, branch string) ([]Summary, error) {
	rows, err := db.QueryContext(ctx, `
		SELECT account_id, owner_name, balance_cents, tx_count, last_activity
		FROM account_summaries          -- one table, no joins  ← the point
		WHERE branch_name = $1
		ORDER BY balance_cents DESC LIMIT 50`, branch)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	out := []Summary{}
	for rows.Next() {
		var s Summary
		if err := rows.Scan(&s.AccountID, &s.Owner, &s.BalanceCents, &s.TxCount, &s.LastActivity); err != nil {
			return nil, err
		}
		out = append(out, s)
	}
	return out, rows.Err()
}

Wiring them to HTTP with Go 1.22+ routing patterns, which put the method and a {wildcard} right in the pattern:

mux := http.NewServeMux()
mux.HandleFunc("POST /accounts/{id}/deposits", func(w http.ResponseWriter, r *http.Request) {
	id, err := strconv.Atoi(r.PathValue("id")) // the {id} part of the URL
	if err != nil {
		http.Error(w, "bad account id", http.StatusBadRequest)
		return
	}
	var body struct {
		Cents int64 `json:"cents"`
	}
	if err := json.NewDecoder(r.Body).Decode(&body); err != nil || body.Cents <= 0 {
		http.Error(w, "bad amount", http.StatusBadRequest)
		return
	}
	if err := deposit(r.Context(), db, id, body.Cents); err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	w.WriteHeader(http.StatusNoContent) // a command answers "done", not data
})
mux.HandleFunc("GET /branches/{name}/dashboard", func(w http.ResponseWriter, r *http.Request) {
	rows, err := branchDashboard(r.Context(), db, r.PathValue("name"))
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(rows)
})

Run against Postgres 16 on this machine:

$ curl -i -X POST localhost:8471/accounts/42/deposits -d '{"cents":10000}'
HTTP/1.1 204 No Content
$ curl -X POST localhost:8471/accounts/42/deposits -d '{"cents":2500}'
$ curl localhost:8471/branches/Riverside/dashboard      # (output wrapped to fit)
[{"accountId":42,"owner":"Ana","balanceCents":12500,"txCount":2,"lastActivity":"2026-10-30T15:37:00.303667Z"},
 {"accountId":43,"owner":"Bo","balanceCents":0,"txCount":0,"lastActivity":null}]

The dashboard is one SELECT from one table, served by the index above, however many tables the write side grows.

What you just signed up for

• The summary row must exist. An UPDATE that matches no row changes nothing and reports no error. So the OpenAccount command inserts the account's account_summaries row in the same transaction that inserts the account (for the run below, accounts 42 and 43 were created together with their summary rows).
• Every write path must update the read table. Forget it in one handler (an admin "fix balance" script, say) and the dashboard silently drifts. Keep all writes to an account behind one function, and add a nightly check that compares account_summaries against a fresh calculation from the write tables.
• Copied data must be updated where it's copied. If Ana changes her name, owner_name must change in every summary row of hers. That's the price of DenormalizationDenormalizationDeliberately copying data into several places (e.g. the product name into every order) so reads need fewer lookups — at the price of having to update every copy..
• Writes get a bit slower (one more UPDATE each). Reads get much faster. That trade is usually worth it when reads outnumber writes.

Event sourcing: store the facts

Events are facts

An event is a record that something happened, written in the past tense: AccountOpened, MoneyDeposited, AddressChanged. In business terms, it's a Domain eventDomain eventA record that something meaningful happened in the business, named in the past tense — OrderPlaced, PaymentRefunded — that other parts of the system can react to..

Compare it with a command:

Because events are facts, they are never edited and never deleted. A mistake isn't fixed by changing the old event; it's fixed by adding a new one (DepositReversed), exactly like an accountant who never uses an eraser.

Streams and the event store

All the events of one thing form a StreamStream (event stream)All the events of one thing, in order: every event of account 42 is the stream “account-42”. Each stream has its own version counter: 1, 2, 3…: every event of account 42, in order, is the stream account-42. Each event in a stream gets the next version number: 1, 2, 3…

An Event storeEvent storeA database (or a table) that holds events in order, only ever adding new ones at the end. It can read one stream in order, read everything after a position, and refuse an append if a stream has moved on. keeps these streams. It's an Append-only logAppend-only logA list you can only add to at the end. Existing entries are never changed, so anyone can read it from any position, any number of times.: new events go on the end, old ones are never touched. In §06 it'll be one ordinary Postgres table.

Where does the balance come from?

Nowhere: it's not stored. To learn the balance, you load the stream and apply each event in turn to a running state, starting from "empty":

Walking through a list while carrying one running value is called a FoldFold (reduce)Walking through a list while carrying one running value: start with an initial state, combine it with item 1, then the result with item 2, and so on. JavaScript's array.reduce is a fold.. In JavaScript it's array.reduce.

Three small functions

Every event-sourced write model comes down to three pieces:

1. evolve(state, event) → state. Applies one fact. No checks: the event already happened, so we just record its effect.
2. decide(state, command) → events. Checks the business rules against the current state. Returns the new events, or an error if the command is refused. (In our code it's called withdraw.)
3. The fold: state = events.reduce(evolve, initial).

A command is handled as: load the stream → fold → decide → append the new events. Then, separately, projectionsProjection (read model)A view of data built by processing a stream of events — e.g. turning OrderPlaced/OrderShipped events into an “orders per customer” table that's fast to query. read the events and keep read modelsRead modelA copy of the data shaped for fast reading (a table, a search index, a cache), built from the events or main records. It can be thrown away and rebuilt. (the plain tables that screens query) up to date. Each projection remembers how far it has read with a bookmark called a checkpoint. That's CQRS again, with the events as the bridge between the two sides.

Step through one account's life. Watch the rejected command at step 7, and what happens to the projection at the end:

1function evolve(state, event) {          // apply ONE fact
2  const s = { ...state, version: state.version + 1 }; // every event bumps the version
3  switch (event.type) {
4    case "AccountOpened": return { ...s, owner: event.owner };
5    case "Deposited":     return { ...s, balance: s.balance + event.amount };
6    case "Withdrawn":     return { ...s, balance: s.balance - event.amount };
7  }
8}
9 
10function withdraw(state, amount) {       // decide: check the rules
11  if (state.balance < amount) throw new Error("insufficient funds");
12  return [{ type: "Withdrawn", amount }];
13}
14 
15const state = events.reduce(evolve, initial); // fold ALL facts

What you get

• The "why" for free. Every change is recorded with what happened. This is your Audit trailAudit trailA record of who changed what, when and why, that auditors and support staff can trust. In event sourcing the events themselves are the audit trail., and it can't drift from the data, because it is the data.
• Time travel. "What was the balance on March 3rd?" Fold only the events up to March 3rd. Questions like this are temporal queriesTemporal queryA question about the past: “what was the balance on March 3rd?” or “what did this order look like before the edit?”. Easy with events: fold only the events up to that moment..
• New read models from old data. Need a screen nobody planned for? Write a new projection and replayReplay (rebuild)Deleting a read model and running its projection again from the first event. Because the events are the source of truth, the result is a fresh, correct read model. all events into it. The table appears as if it had always existed.
• Natural change feed. Other parts of the system can react to the events (§10).

What it costs

• No more UPDATE … WHERE. You can't fix bad data with one SQL statement; you add correcting events, and you need tools to do that safely.
• Reads go through projections, which run slightly behind the writes (§08).
• Events are forever, so their format must be designed and changed with care (§09).
• Unfamiliar. Most developers have never debugged a projection. §11 says when that's worth it.

Decide and evolve, in code

Here's the bank account from §04 as real code. Money is in cents, and every event bumps the stream version, which §06 will use to catch conflicting writes.

package bank

import "errors"

// Event is one fact that happened to an account. Past tense, never changed.
type Event struct {
	Type   string `json:"-"`                // "AccountOpened", "Deposited", "Withdrawn"
	Owner  string `json:"owner,omitempty"`  // only set on AccountOpened
	Amount int64  `json:"amount,omitempty"` // in cents; 0 on AccountOpened
}

// Account is the current state, rebuilt from events. Never stored directly.
type Account struct {
	Owner   string
	Balance int64 // cents
	Open    bool
	Version int // how many events we've applied = the stream's version
}

// Evolve applies ONE event to the state and returns the new state.
// No checks, no I/O: the event already happened, we just record its effect.
func Evolve(a Account, e Event) Account {
	switch e.Type {
	case "AccountOpened":
		a.Owner, a.Open = e.Owner, true
	case "Deposited":
		a.Balance += e.Amount
	case "Withdrawn":
		a.Balance -= e.Amount
	}
	a.Version++
	return a
}

// Load folds a whole history into the current state.
func Load(history []Event) Account {
	var a Account // zero value: no owner, balance 0, version 0
	for _, e := range history {
		a = Evolve(a, e) // ← the fold: state = evolve(state, event), repeated
	}
	return a
}

var (
	ErrNotOpen           = errors.New("account is not open")
	ErrBadAmount         = errors.New("amount must be positive")
	ErrInsufficientFunds = errors.New("insufficient funds")
)

// Withdraw decides: given the current state and a command, which events happen?
// This is where the business rules (invariants) live.
func Withdraw(a Account, amount int64) ([]Event, error) {
	switch {
	case !a.Open:
		return nil, ErrNotOpen
	case amount <= 0:
		return nil, ErrBadAmount
	case a.Balance < amount:
		return nil, ErrInsufficientFunds // rejected: no event is recorded
	}
	return []Event{{Type: "Withdrawn", Amount: amount}}, nil
}

Evolve takes the account by value (a copy) and returns a new one, so nobody else's copy changes behind their back. The JavaScript version gets the same effect with spread syntax.

Notice what Evolve doesn't do: it never checks the balance. If the log says Withdrawn 90, then 90 was withdrawn, and refusing to apply it would make the computed state disagree with history. Checks belong in Withdraw, before the event exists.

Testing without a database

Because both functions are pure (output depends only on input, no I/O), the tests are tiny. The usual shape is given some past events, when a command arrives, then expect these new events (or this error):

var history = []Event{ // given
	{Type: "AccountOpened", Owner: "Ana"},
	{Type: "Deposited", Amount: 10_000}, // $100.00
	{Type: "Withdrawn", Amount: 3_000},  // $30.00
}

func TestLoadFoldsHistory(t *testing.T) {
	a := Load(history)
	if a.Balance != 7_000 || a.Version != 3 || a.Owner != "Ana" {
		t.Fatalf("got %+v", a)
	}
}

func TestWithdrawRejectsOverdraft(t *testing.T) {
	_, err := Withdraw(Load(history), 9_000) // when: only $70 available
	if !errors.Is(err, ErrInsufficientFunds) { // then: refused
		t.Fatalf("want ErrInsufficientFunds, got %v", err)
	}
}

func TestWithdrawEmitsEvent(t *testing.T) {
	got, err := Withdraw(Load(history), 2_000)
	if err != nil || len(got) != 1 || got[0] != (Event{Type: "Withdrawn", Amount: 2_000}) {
		t.Fatalf("got %v, %v", got, err)
	}
}

$ go test -v ./bank/
=== RUN   TestLoadFoldsHistory
--- PASS: TestLoadFoldsHistory (0.00s)
=== RUN   TestWithdrawRejectsOverdraft
--- PASS: TestWithdrawRejectsOverdraft (0.00s)
=== RUN   TestWithdrawEmitsEvent
--- PASS: TestWithdrawEmitsEvent (0.00s)
PASS
ok      example.com/bank/bank   0.002s

$ node --test
# tests 3
# pass 3
# fail 0

(Go 1.24 and Node 22 on this machine.)

In DDD terms, the thing whose rules decide protects (one account) is an AggregateAggregateA cluster of objects (an Order and its lines) that must stay consistent together, treated as one unit: changed only through its root, loaded and saved whole, in one transaction., and one aggregate usually maps to one stream. See DDD for how to draw those boundaries.

The event store: streams, versions, conflicts

One table is enough

You don't need special software to start. Here's a complete Event storeEvent storeA database (or a table) that holds events in order, only ever adding new ones at the end. It can read one stream in order, read everything after a position, and refuse an append if a stream has moved on. in Postgres:

CREATE TABLE events (
  position   bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY, -- global order: 1, 2, 3…
  tx_id      xid8        NOT NULL DEFAULT pg_current_xact_id(), -- which transaction wrote it (§07)
  stream_id  text        NOT NULL,           -- which account: 'account-42'
  version    int         NOT NULL,           -- 1, 2, 3… within that one stream
  type       text        NOT NULL,           -- 'Deposited'
  data       jsonb       NOT NULL,           -- {"amount": 10000}
  created_at timestamptz NOT NULL DEFAULT now(),
  UNIQUE (stream_id, version)                -- ← optimistic concurrency, enforced by Postgres
);
CREATE INDEX ON events (tx_id, position);    -- for projections (§07)

Two numbers, two jobs:

• version counts events within one stream. Loading an account means WHERE stream_id = 'account-42' ORDER BY version. The UNIQUE rule also creates an index on (stream_id, version), which makes that query fast.
• position orders every event in the table. It's the Global positionGlobal positionA number that orders every event in the whole store, across all streams (1, 2, 3…). Projections use it to remember how far they've read., and projections (§07) use it to remember how far they've read.

data is JSONBJSONBPostgres's binary JSON column type. It stores parsed JSON so the database can search inside it and index it., so each event type can have its own fields without a table per type. Ignore tx_id for now: it records which database transaction wrote the event, and §07 needs it to read events safely.

Why the version check matters

Two clerks withdraw $60 each from an account holding $100, at the same moment. Each one loads the account, checks the rule on what it loaded, and appends. Flip between the two modes:

1async function handleWithdraw(accountId, amount) {
2  const account = await load(accountId);           // fold → state + version
3  const events = withdraw(account, amount);         // rules, on this copy
4  await append(accountId, account.version, events); // "I saw version N"
5  // append inserts version N+1. If N+1 already exists → ConcurrencyError,
6  // and the caller reloads and decides again.
7}

That bug is a lost update: a decision made on data that changed before it was saved (Race conditions §05 shows it with ordinary rows). The fix here is Optimistic lockingOptimistic lockingDon't lock while reading. When writing, check that nobody changed the row since you read it (usually with a version number); if someone did, re-read and retry.: don't lock anything while deciding; instead, send your Expected versionExpected versionThe stream version a writer saw when it loaded the stream. It sends that number with its append; if the stream has grown since, the store refuses the append, so the writer must reload and decide again. with the write, and let the database reject it if someone got there first.

With UNIQUE (stream_id, version) that's free. If you saw version 2, your new event is version 3. If version 3 already exists, the insert breaks the UNIQUE constraintUNIQUE constraintA rule on a database column: the database refuses to store a second row with the same value there — even if two inserts arrive at the same instant. and fails. Here's the real thing on Postgres 16:

INSERT INTO events (stream_id, version, type, data) VALUES
  ('account-42', 1, 'AccountOpened', '{"owner": "Ana"}'),
  ('account-42', 2, 'Deposited',     '{"amount": 10000}'),
  ('account-42', 3, 'Withdrawn',     '{"amount": 3000}');

-- Two clerks both loaded the stream at version 3. Clerk A appends first:
INSERT INTO events (stream_id, version, type, data)
VALUES ('account-42', 4, 'Withdrawn', '{"amount": 5000}');

-- Clerk B, still believing the stream is at version 3, tries the same:
INSERT INTO events (stream_id, version, type, data)
VALUES ('account-42', 4, 'Withdrawn', '{"amount": 6000}');

INSERT 0 3
INSERT 0 1
ERROR:  duplicate key value violates unique constraint "events_stream_id_version_key"
DETAIL:  Key (stream_id, version)=(account-42, 4) already exists.

Postgres reports this as error code 23505 (unique_violation). The application turns that one code into "someone else wrote first".

Those two inserts ran one after the other. If they really overlap (A has inserted version 4 but not committed yet), B's insert waits for A's transaction to finish, and then fails the same way. On this machine B waited about 0.8 s for A's deliberately slow commit, then got the same error. If A had rolled back instead, B's insert would have gone through.

Append, load, and the retry loop

import (
	"github.com/jackc/pgx/v5/pgconn"
	_ "github.com/jackc/pgx/v5/stdlib" // registers the "pgx" driver for database/sql
	// …plus context, database/sql, encoding/json, errors, fmt and our bank package
)

// Opened elsewhere with: db, err := sql.Open("pgx", "postgres://user@host/dbname")

// ErrConcurrency means: someone appended to this stream after you read it.
var ErrConcurrency = errors.New("stream changed since you read it")

// isUniqueViolation reports whether Postgres refused a row because of a UNIQUE rule.
// It only works with the pgx driver: other drivers wrap errors in their own types.
func isUniqueViolation(err error) bool {
	var pgErr *pgconn.PgError // the error type pgx returns for server errors
	return errors.As(err, &pgErr) && pgErr.Code == "23505" // 23505 = unique_violation
}

// Append writes new events to one stream, but only if the stream is still at
// expectedVersion. Each event gets the next version: expected+1, +2, …
func Append(ctx context.Context, db *sql.DB, streamID string, expectedVersion int, events []bank.Event) error {
	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer tx.Rollback() // no-op after Commit

	for i, e := range events {
		data, err := json.Marshal(e)
		if err != nil {
			return err
		}
		_, err = tx.ExecContext(ctx,
			`INSERT INTO events (stream_id, version, type, data) VALUES ($1, $2, $3, $4)`,
			streamID, expectedVersion+i+1, e.Type, data)
		if isUniqueViolation(err) {
			return ErrConcurrency // ← somebody else already wrote this version
		}
		if err != nil {
			return err
		}
	}
	return tx.Commit()
}

// Load reads one stream in version order and folds it into an Account.
func Load(ctx context.Context, db *sql.DB, streamID string) (bank.Account, error) {
	rows, err := db.QueryContext(ctx,
		`SELECT type, data FROM events WHERE stream_id = $1 ORDER BY version`, streamID)
	if err != nil {
		return bank.Account{}, err
	}
	defer rows.Close()

	var a bank.Account
	for rows.Next() {
		var data []byte
		var e bank.Event
		if err := rows.Scan(&e.Type, &data); err != nil {
			return bank.Account{}, err
		}
		if err := json.Unmarshal(data, &e); err != nil {
			return bank.Account{}, fmt.Errorf("decode event: %w", err)
		}
		a = bank.Evolve(a, e)
	}
	return a, rows.Err()
}

// HandleWithdraw: load → decide → append, retrying if we lost a race.
func HandleWithdraw(ctx context.Context, db *sql.DB, accountID string, amount int64) error {
	for attempt := 1; attempt <= 3; attempt++ {
		acct, err := Load(ctx, db, accountID)
		if err != nil {
			return err
		}
		events, err := bank.Withdraw(acct, amount) // rules checked on fresh state
		if err != nil {
			return err // e.g. ErrInsufficientFunds: nothing is written
		}
		err = Append(ctx, db, accountID, acct.Version, events)
		if !errors.Is(err, ErrConcurrency) {
			return err // nil (success) or a real error
		}
		// Lost the race: someone appended first. Loop: reload and decide again.
	}
	return ErrConcurrency
}

Run with the race made deterministic: both clerks load first, then both append (output from this machine):

A appends at expected version 2 → <nil>
B appends at expected version 2 → stream changed since you read it
B retries (reload + decide again) → insufficient funds
final: balance=4000 version=3

Exactly the "expected version" ending from the widget: no lock was held while deciding, and the invariant (balance ≥ 0) survived.

Projections: building read models from events

A ProjectionProjection (read model)A view of data built by processing a stream of events — e.g. turning OrderPlaced/OrderShipped events into an “orders per customer” table that's fast to query. turns the event log into a Read modelRead modelA copy of the data shaped for fast reading (a table, a search index, a cache), built from the events or main records. It can be thrown away and rebuilt.. The code that runs it is often called a projector. Each projection keeps its own checkpoint: a bookmark saying which event it finished last. In the lab below it's simply the Global positionGlobal positionA number that orders every event in the whole store, across all streams (1, 2, 3…). Projections use it to remember how far they've read. of that event; on real Postgres it needs one more number, for a reason this section explains. Many projections can read the same log, each at its own pace, like several readers of one logbook each with their own bookmark.

Play with one

Three things to try, and the lesson in each:

1. Process a few, then append. The read model is catching upCatch-upA projection reading every event after its checkpoint, in batches, until it reaches the newest event. Then it waits for more.: it's always somewhat behind the log, and it closes the gap batch by batch.
2. Crash with "checkpoint saved separately", then process next. The same event is applied twice, and the balance is wrong forever after. The table change was saved, the bookmark wasn't.
3. Rebuild from scratch. The table is thrown away and rebuilt from position 0, and it comes out right. This ReplayReplay (rebuild)Deleting a read model and running its projection again from the first event. Because the events are the source of truth, the result is a fresh, correct read model. is how you fix a projection bug, or add a column: change the code, rebuild.

Which events are safe to read?

Before the code, one trap that catches almost everyone who builds an event store on an ordinary database. position numbers are handed out when an INSERT runs, but a row only becomes visible when its transaction commits, and transactions don't commit in the order they started. Two sessions on Postgres 16:

T2: INSERT … RETURNING position  → 8
projection reads position > 6 now:
 position | stream_id |     type
----------+-----------+---------------
        8 | account-9 | AccountOpened
(1 row)
-- …T1 commits a moment later; position 7 now exists, behind the checkpoint.

The obvious fix is "only read events whose transaction has finished". Every Postgres transaction that writes gets a transaction ID, a number that only goes up, assigned at its first write (see Transaction ID (xid8) / pg_snapshot_xminTransaction ID (xid8) / pg_snapshot_xminPostgres gives each transaction an increasing number, its transaction ID, the first time it writes something. xid8 is the column type for it; pg_current_xact_id() returns the current one. pg_snapshot_xmin(pg_current_snapshot()) is the ID of the oldest transaction still running: every ID below it has finished (committed or rolled back).). The events table from §06 records it in tx_id, and pg_snapshot_xmin(pg_current_snapshot()) gives the ID of the oldest transaction still running. So tx_id < pg_snapshot_xmin(…) keeps only events from transactions that are certainly finished.

That filter alone is not enough. A transaction that did some other write first gets an older ID but can still take a higher position. Step through it, comparing the three ways of reading:

1-- checkpoint: position only
2SELECT * FROM events
3WHERE position > $checkpoint
4ORDER BY position

The same interleaving on real Postgres 16, typed into psql (results abbreviated to the rows returned):

-- A: xid 737, position 3, committed.   B: xid 738, position 2, still open.
-- Checkpoint so far: position 1, from transaction 736.
position only            WHERE position > 1                                → 3
position + filter        … AND tx_id < pg_snapshot_xmin(…)                 → 3
pair + filter            WHERE (tx_id, position) > ('736', 1) AND …filter  → 737 | 3
-- B commits. Next read, from each method's saved checkpoint:
position only            WHERE position > 3                                → (no rows: position 2 is lost)
pair + filter            WHERE (tx_id, position) > ('737', 3) AND …filter  → 738 | 2

The rule that works: read only finished transactions, in (tx_id, position) order, and save both numbers as the checkpoint. Rows are compared as pairs: first by tx_id, and only on a tie by position. A transaction that is still running has an ID at or above the horizon, so everything it will ever write sorts after anything the projector has already passed.

The projector, in code

The checkpoint table stores both numbers:

CREATE TABLE checkpoints (
  name     text   PRIMARY KEY,               -- one row per projection: 'balances', 'publisher'…
  tx_id    xid8   NOT NULL DEFAULT '0',      -- last event's transaction ID…
  position bigint NOT NULL DEFAULT 0         -- …and its position
);
INSERT INTO checkpoints (name) VALUES ('balances');

CREATE TABLE balances (account_id text PRIMARY KEY, owner text NOT NULL, balance_cents bigint NOT NULL DEFAULT 0);

If the read model lives in the same Postgres as the checkpoint, the fix for lesson 2 of the lab is easy too: update the table and the checkpoint in one Database transactionDatabase transactionA group of database changes that happen all together or not at all (COMMIT keeps them, ROLLBACK undoes them). How much other users can see or interfere with the half-done state depends on the isolation level.. Either both are saved or neither is, so after a crash the event is simply processed again from a clean slate.

// stored is one row of the events table, as a projector reads it.
type stored struct {
	TxID     string // xid8 of the writing transaction, kept as text: only SQL compares it
	Position int64
	StreamID string
	Version  int
	Type      string
	Data      []byte
	CreatedAt time.Time // when it was appended (the publisher in §10 uses it)
}

// querier is anything that can run a SELECT: both *sql.DB and *sql.Tx have this method.
type querier interface {
	QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error)
}

// readAfter returns events after the checkpoint (txID, pos), in transaction-ID order,
// and only from transactions that have finished.
func readAfter(ctx context.Context, q querier, txID string, pos int64, limit int) ([]stored, error) {
	rows, err := q.QueryContext(ctx, `
		SELECT tx_id::text, position, stream_id, version, type, data, created_at FROM events
		WHERE (tx_id, position) > ($1::xid8, $2)               -- after the checkpoint
		  AND tx_id < pg_snapshot_xmin(pg_current_snapshot()) -- writer has finished  ← the point
		ORDER BY tx_id, position
		LIMIT $3`, txID, pos, limit)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []stored
	for rows.Next() {
		var e stored
		if err := rows.Scan(&e.TxID, &e.Position, &e.StreamID, &e.Version, &e.Type, &e.Data, &e.CreatedAt); err != nil {
			return nil, err
		}
		out = append(out, e)
	}
	return out, rows.Err()
}

// CatchUp applies every safe event after the checkpoint, one batch per transaction.
// It returns how many events it applied (0 = caught up, for now).
func CatchUp(ctx context.Context, db *sql.DB) (int, error) {
	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		return 0, err
	}
	defer tx.Rollback()

	var cpTx string // FOR UPDATE: only one copy of this projector runs a batch at a time
	var cpPos int64
	if err := tx.QueryRowContext(ctx,
		`SELECT tx_id::text, position FROM checkpoints WHERE name = 'balances' FOR UPDATE`).Scan(&cpTx, &cpPos); err != nil {
		return 0, err
	}
	batch, err := readAfter(ctx, tx, cpTx, cpPos, 500)
	if err != nil {
		return 0, err
	}
	for _, e := range batch {
		if err := apply(ctx, tx, e); err != nil {
			return 0, err
		}
		cpTx, cpPos = e.TxID, e.Position
	}
	// The checkpoint moves in the SAME transaction as the read-model changes.  ← the point
	if _, err := tx.ExecContext(ctx,
		`UPDATE checkpoints SET tx_id = $1::xid8, position = $2 WHERE name = 'balances'`, cpTx, cpPos); err != nil {
		return 0, err
	}
	return len(batch), tx.Commit()
}

func apply(ctx context.Context, tx *sql.Tx, e stored) error {
	var d struct {
		Owner  string `json:"owner"`
		Amount int64  `json:"amount"`
	}
	if err := json.Unmarshal(e.Data, &d); err != nil {
		return err
	}
	var err error
	switch e.Type {
	case "AccountOpened":
		_, err = tx.ExecContext(ctx, `
			INSERT INTO balances (account_id, owner) VALUES ($1, $2)
			ON CONFLICT (account_id) DO NOTHING`, e.StreamID, d.Owner) // seen it before? skip
	case "Deposited":
		_, err = tx.ExecContext(ctx,
			`UPDATE balances SET balance_cents = balance_cents + $2 WHERE account_id = $1`, e.StreamID, d.Amount)
	case "Withdrawn":
		_, err = tx.ExecContext(ctx,
			`UPDATE balances SET balance_cents = balance_cents - $2 WHERE account_id = $1`, e.StreamID, d.Amount)
	} // other event types: this projection doesn't care, skip them
	return err
}

The A/B interleaving from the stepper, run through this exact Go code on Postgres 16 (the JS version prints the same, with different numbers):

TB holds position 4 (open); TA committed position 5
catch-up while TB is open: applied 1 → checkpoint (741, 5) · rows: account-A
catch-up after TB commits: applied 1 → checkpoint (742, 4) · rows: account-A account-B

Notice the checkpoint's position went down (5, then 4) while the pair went up. Never assume positions arrive in increasing order. A normal run over the bank's events ends with balances holding account-42 (Ana) at 7000 and account-43 (Bo) at 0, matching a fold of the events.

A real projector calls CatchUp in a loop: while it returns more than 0, call again; when it returns 0, wait (poll every 100–500 ms, or have the writer send a Postgres NOTIFY so the projector wakes up at once) and try again.

When the read model lives somewhere else

If the read model is in Elasticsearch, Redis or another database, the checkpoint and the change can't share one transaction. Then the projector will sometimes see an event twice (after a crash, it re-reads from the last saved checkpoint). That's At-least-once deliveryAt-least-once deliveryA delivery promise where a message is never lost but may arrive more than once (for example, re-sent after a crash). Receivers must therefore handle duplicates safely. delivery, and the cure is to make each update IdempotentIdempotentDoing it twice has the same effect as doing it once. “Set balance to 50” is idempotent; “add 50 to balance” is not — so only the first is safe to retry blindly.: applying it twice must equal applying it once.

The trick: store, in each read-model row, the stream version it has applied, and skip events it has already seen. Shown in SQL for clarity; the same idea works in any store that can do a conditional write:

CREATE TABLE balances (
  account_id    text   PRIMARY KEY,
  owner         text   NOT NULL,
  balance_cents bigint NOT NULL DEFAULT 0,
  version       int    NOT NULL DEFAULT 0   -- last stream version applied to this row
);

-- AccountOpened (version 1): a second copy hits the existing row and does nothing
INSERT INTO balances (account_id, owner, version) VALUES ('account-42', 'Ana', 1)
ON CONFLICT (account_id) DO NOTHING;

-- Deposited 10000 (version 2): only if this row hasn't seen version 2 yet
UPDATE balances SET balance_cents = balance_cents + 10000, version = 2
WHERE account_id = 'account-42' AND version < 2;

Each statement run twice in a row on Postgres 16: the row ends at balance 10000, version 2, not 20000. This works because one stream's events always arrive in version order (the writer rule above). More dedup patterns: Idempotency §08.

Rebuilding without downtime

Rebuilding the live table empties the dashboard until the replay finishes. For big read models, do a blue/green rebuild instead: create balances_v2, run the new projection into it with its own checkpoint row, and when it has caught up, switch the query code (or a database view) to balances_v2 and drop the old one. Replaying millions of events can take hours, so measure it before you need it in an incident.

The lag: eventual consistency in the UI

Where the gap comes from

A command appends an event and returns. The projection picks the event up a moment later: a few milliseconds when all is well, seconds when it's busy, minutes if it's stuck. Until then the read model shows the old state. That's Eventual consistencyEventual consistencyA promise that if writes stop, every copy or derived view of the data (replicas, caches, read models, other services) will end up agreeing — but for a while, different readers may see different (stale) versions.: the read side will agree with the write side, just not instantly.

For most screens nobody notices. The screen that does notice is the one the user is looking at right after clicking a button. Ana withdraws $30, the page reloads, and it still says $100. Did it work? She clicks again.

Try the strategies

• Just reload: fine when the lag is tiny, confusing when it isn't. Drag the lag slider above about 70 ms and the reload shows the old balance.
• Wait for the version: the command returns the position of its event; the next query passes it along and the server waits until the projection has reached it. This gives Read-your-writesRead-your-writesA guarantee that after you change something, your own later reads show the change. Other users may still see the old value for a moment.: you always see your own change, at the cost of a short wait.
• Optimistic UI: the browser shows the expected result immediately and marks it pending (Optimistic UIOptimistic UIThe screen shows the result of an action right away, assuming it will succeed, and marks it as pending until the server confirms. If the server says no, the screen rolls back.). Snappy, but the client has to guess the outcome, and must roll back if the server refuses.

"Wait for the version", in code

The command handler returns where its event landed, and the query handler takes ?after=<position> and waits, up to a limit, until the projection has applied that event.

First, Append from §06 must report the position. Three lines change: ExecContext can't read a returned row, so the insert becomes a QueryRowContext(…).Scan:

var position int64 // ← new: the last event's position, returned to the caller
err = tx.QueryRowContext(ctx, `
	INSERT INTO events (stream_id, version, type, data) VALUES ($1, $2, $3, $4)
	RETURNING position`, streamID, expectedVersion+i+1, e.Type, data).Scan(&position)
// …and the function now returns (int64, error): `return position, tx.Commit()`

The withdrawal handler then answers with JSON like {"position": 19} instead of an empty 204. (Strict CQRS says commands return nothing; a position is metadata, and returning it is common practice.)

"Has the projection applied position 19?" can't be answered by comparing positions, because §07's projector reads in (tx_id, position) order and positions can arrive out of order. So the check compares in the same order the projector uses: look up the event's pair and compare it with the checkpoint's pair.

// waitFor blocks until the projection has applied the event at `position`, or ctx ends.
func waitFor(ctx context.Context, db *sql.DB, position int64) error {
	if position == 0 {
		return nil // the caller didn't write anything: no need to wait
	}
	tick := time.NewTicker(50 * time.Millisecond)
	defer tick.Stop()
	for {
		var done bool
		err := db.QueryRowContext(ctx, `
			SELECT (c.tx_id, c.position) >= (e.tx_id, e.position)  -- same order the projector reads in
			FROM checkpoints c, events e
			WHERE c.name = 'balances' AND e.position = $1`, position).Scan(&done)
		if err != nil {
			return err // includes sql.ErrNoRows: no such event
		}
		if done {
			return nil // the read model includes the caller's write  ← the point
		}
		select {
		case <-ctx.Done():
			return ctx.Err() // timed out (or the client went away)
		case <-tick.C: // check again in 50 ms
		}
	}
}

func balanceHandler(db *sql.DB) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		after, _ := strconv.ParseInt(r.URL.Query().Get("after"), 10, 64) // 0 if missing
		ctx, cancel := context.WithTimeout(r.Context(), 2*time.Second)
		defer cancel()

		stale := false
		if err := waitFor(ctx, db, after); errors.Is(err, context.DeadlineExceeded) {
			stale = true // waited 2 s and still behind: answer anyway, but say so
		} else if err != nil {
			http.Error(w, err.Error(), http.StatusInternalServerError)
			return
		}
		var cents int64
		err := db.QueryRowContext(r.Context(),
			`SELECT balance_cents FROM balances WHERE account_id = $1`, r.PathValue("id")).Scan(&cents)
		if err != nil {
			http.Error(w, "not found", http.StatusNotFound)
			return
		}
		w.Header().Set("Content-Type", "application/json")
		json.NewEncoder(w).Encode(map[string]any{"balanceCents": cents, "stale": stale})
	}
}

// mux.HandleFunc("GET /accounts/{id}/balance", balanceHandler(db))

Measured on this machine, with the Go handler and §07's projector:

$ curl 'localhost:8472/accounts/account-42/balance?after=15'   # already applied
{"balanceCents":7000,"stale":false}                             ← immediate

# a new Deposited 500 lands at position 18; the projector is stopped
$ time curl 'localhost:8472/accounts/account-42/balance?after=18'
{"balanceCents":7000,"stale":true}
real    0m2.008s                                                ← waited the full 2 s, then answered

# another deposit at position 19; the projector runs 0.4 s later
$ time curl 'localhost:8472/accounts/account-42/balance?after=19'
{"balanceCents":7501,"stale":false}
real    0m0.462s                                                ← answered as soon as it caught up

Polling the checkpoint every 50 ms is simple and fine for modest traffic. At scale, have the projector broadcast "checkpoint is now N" (Postgres LISTEN/NOTIFY, Redis pub/subPublish/subscribe (pub/sub)Senders (publishers) emit messages without knowing who receives them; receivers (subscribers) sign up for the kinds of message they care about. The two sides never reference each other directly.) so waiting requests wake up at once instead of polling.

Other ways to close the gap

• Read your own data from the write side. For "the thing I just changed", load the stream and fold it (§06's Load), which is always current. Use the read model for lists, search and dashboards.
• Return the new state from the command. Strict CQRS says commands return nothing, but returning {position, newBalance} is harmless and saves a round trip. Treat "commands return nothing" as a guideline, not a law.
• Push updates. Send a message over a WebSocketWebSocketA long-lived two-way connection between browser and server, used for chat and live updates. It starts as a normal HTTP request, then the server takes over (“hijacks”) the raw connection. or Server-sent eventsServer-sent events (SSE)A simple one-way stream from server to browser over one long-lived HTTP response. Each message is a line like data: hello, sent (and flushed) as soon as it's ready. when the projection has applied the user's event, and update the screen then.
• Design the screen for it. "Your withdrawal is being processed" is honest and is what banks actually show.

Living with events for years

Long streams and snapshots

Every command loads its stream and folds it (§06). For an account with 50 events that's nothing. For a stream with 100,000 events, it's real work on every single command. How much, on this machine (Postgres 16 on localhost, Go with pgx, so no network in the way):

The fold is cheap; reading and decoding is the cost. Over a real network it's slower still.

A SnapshotSnapshot (event sourcing)A saved copy of one stream's state as of some version, e.g. “account-42 at version 10,000: balance $512”. Loading then reads the snapshot plus only the newer events. is the folded state saved at some version, like "account-big at version 100,000: balance $999.99". Loading becomes "read the snapshot, then fold only the events after it":

// LoadWithSnapshot starts from the saved snapshot (if any) and folds only newer events.
func LoadWithSnapshot(ctx context.Context, db *sql.DB, streamID string) (bank.Account, error) {
	var a bank.Account // zero value = "no snapshot": start from version 0
	var state []byte
	err := db.QueryRowContext(ctx,
		`SELECT state FROM snapshots WHERE stream_id = $1 AND schema = 1`, streamID).Scan(&state)
	switch {
	case errors.Is(err, sql.ErrNoRows): // no snapshot yet (or an old-format one): fold from the start
	case err != nil:
		return a, err
	default:
		if err := json.Unmarshal(state, &a); err != nil {
			return a, err
		}
	}

	rows, err := db.QueryContext(ctx, `
		SELECT type, data FROM events
		WHERE stream_id = $1 AND version > $2   -- ← only what came after the snapshot
		ORDER BY version`, streamID, a.Version)
	// … then the same fold loop as Load in §06
}

Saving uses an UpsertUpsertInsert a row, or update it if a row with the same key already exists, in one statement. In Postgres: INSERT … ON CONFLICT … DO UPDATE.: one statement that inserts the snapshot row, or replaces it if the stream already has one.

Rules that keep snapshots from hurting you:

• A snapshot is a cache, never the truth. You can delete every snapshot and the system still works, just slower. Never write business data only into a snapshot.
• Version the snapshot's shape. When you add a field to Account, old snapshots don't have it. The schema = 1 filter above simply ignores old-format snapshots, so the next load folds from scratch and a new snapshot replaces the old one.
• Don't add them until you've measured. Most streams are short. If one stream type grows without limit, the better question is why, which brings us to closing the books.

When the event format changes

The code will change; the stored events won't. Two years in, you'll want Deposited to look different from the one you wrote on day one. Order of preference:

1. Add, don't change. Adding an optional field (channel: "atm") needs no special handling if your readers are lenient: ignore unknown fields, use a default when a field is missing. That style is called a Weak schemaWeak schema (tolerant reader)Reading events leniently: ignore fields you don't know, fill missing optional fields with defaults. Small additions then need no version bump or upcaster.. Go's encoding/json and plain JS objects already behave this way.
2. Version and upcast. When a field's meaning changes, store a schema version with each event and convert old versions to the current shape while reading. That's UpcastingUpcastingConverting an old-format event into today's format while reading it, so the code only handles the newest shape. The stored event itself is never rewritten.. evolve then only ever sees the newest shape.
3. New event type. If it's really a different fact, give it a new name (DepositReceived) and handle both in evolve.
4. Rewrite the store (copy every event through a transformer into a new store, then switch over). A last resort: it's a big migration, and it breaks the promise that events never change, which auditors and other consumers may depend on.

Here's an upcaster for a real-looking mistake: version 1 stored dollars as a decimal number with no currency.

One trap first: converting 1.005 dollars with Math.round(1.005 * 100) gives 100 cents, not 101, because 1.005 can't be stored exactly in floating point (it's really 1.00499999…). The upcaster exists to fix a float problem, so it mustn't add another one: it does the ×100 on the digits as written.

// DepositedV2 is the shape the code understands today.
type DepositedV2 struct {
	Amount   int64  `json:"amount"`   // cents
	Currency string `json:"currency"` // "USD"
}

// upcastDeposited turns ANY stored version of Deposited into today's shape.
// Old rows in the database are never rewritten; we translate them on read.
func upcastDeposited(schema int, raw []byte) (DepositedV2, error) {
	switch schema {
	case 1: // 2023: {"amount": 100.5}, dollars with a decimal point, no currency
		var v1 struct {
			Amount json.Number `json:"amount"` // keep the digits as written: "1.005", not a float
		}
		if err := json.Unmarshal(raw, &v1); err != nil {
			return DepositedV2{}, err
		}
		cents, err := dollarsToCents(v1.Amount.String())
		if err != nil {
			return DepositedV2{}, err
		}
		return DepositedV2{Amount: cents, Currency: "USD"}, nil // every v1 account was in USD
	case 2: // today's shape: decode as is
		var v2 DepositedV2
		err := json.Unmarshal(raw, &v2)
		return v2, err
	}
	return DepositedV2{}, fmt.Errorf("unknown Deposited schema %d", schema)
}

// dollarsToCents does the ×100 in exact decimal arithmetic, then rounds half away from zero.
func dollarsToCents(s string) (int64, error) {
	r, ok := new(big.Rat).SetString(s) // "1.005" → exactly 1005/1000
	if !ok {
		return 0, fmt.Errorf("bad amount %q", s)
	}
	r.Mul(r, big.NewRat(100, 1))                      // exactly 100.5
	return strconv.ParseInt(r.FloatString(0), 10, 64) // "101"
}

$ go run ./upcast                                   $ node upcast.js
{"amount": 100.5} → {Amount:10050 Currency:USD}     100.5 → { amount: 10050, currency: 'USD' }
{"amount": 1.005} → {Amount:101 Currency:USD}       1.005 → { amount: 101, currency: 'USD' }
{"amount": 2.675} → {Amount:268 Currency:USD}       2.675 → { amount: 268, currency: 'USD' }
v2 EUR            → {Amount:10050 Currency:EUR}     v2 EUR → { amount: 10050, currency: 'EUR' }
                                                    naive Math.round(1.005 * 100) = 100

The schema number lives next to the event: a schema int column in the events table, or a field in a metadata JSON column. The JS version shows the chain style: when v3 arrives you add one function 2: (d) => …, and v1 events flow through both steps.

"Delete my data": append-only meets the law

Under the Right to erasureRight to erasure (GDPR)The EU rule that people may ask a company to delete their personal data. A log you can never change makes this hard, so event-sourced systems plan for it up front., a customer can ask you to delete their Personal dataPersonal data (PII)Personally identifiable information: anything that points to a real person, such as a name, email, address, phone number or IP address.. But your log is append-only, it's copied into backups, and other services have read it. You can't reach into all of that and delete one name.

The standard answer is Crypto-shreddingCrypto-shreddingStoring personal data encrypted with a per-person key and “deleting” it by destroying the key. The bytes stay in the append-only log, but nobody can read them any more.: never put personal data into events in plain text. Encrypt it with a key per person, keep the keys in a separate (deletable) table, and to "forget" someone, delete their key. The encrypted bytes remain, everywhere, and are now useless.

func encryptFor(customer string, plaintext []byte) ([]byte, error) {
	key, ok := keys[customer] // keys: a table of one random 32-byte key per customer
	if !ok {
		key = make([]byte, 32) // AES-256
		rand.Read(key)
		keys[customer] = key
	}
	block, err := aes.NewCipher(key)
	if err != nil {
		return nil, err
	}
	gcm, err := cipher.NewGCM(block)
	if err != nil {
		return nil, err
	}
	nonce := make([]byte, gcm.NonceSize())
	rand.Read(nonce)
	return gcm.Seal(nonce, nonce, plaintext, nil), nil // nonce is stored in front
}

func decryptFor(customer string, sealed []byte) ([]byte, error) {
	key, ok := keys[customer]
	if !ok {
		return nil, errors.New("key deleted: data is unreadable") // ← shredded
	}
	block, err := aes.NewCipher(key)
	if err != nil {
		return nil, err
	}
	gcm, err := cipher.NewGCM(block)
	if err != nil {
		return nil, err
	}
	n := gcm.NonceSize()
	return gcm.Open(nil, sealed[:n], sealed[n:], nil)
}

{"name":"Ana Lima","email":"ana@example.com"} <nil>     ← decrypted with the key
key deleted: data is unreadable                          ← after delete(keys, "cust-7")

Both use AESAESAdvanced Encryption Standard: the standard symmetric cipher, built into the hardware of most CPUs so it runs at gigabytes per second. TLS uses it in GCM mode (AES-GCM), which also adds a tamper-proof tag. in GCM mode, which also detects tampering. The nonce (or IV) is a random value used once per encryption; it isn't secret, so it's stored in front of the ciphertext.

Telling other services: outbox, duplicates, sagas

The trap: writing twice

Without event sourcing, a service that wants to tell others "money was withdrawn" is tempted to do two things in a row: commit to its database, then send a message to KafkaKafkaApache Kafka: a durable, ordered log of events, split into partitions, that many programs can publish to and read from at their own pace.. If the process dies between the two, the database says "withdrawn" and nobody else ever hears about it. Swap the order and you can announce a withdrawal that never committed.

The usual fix is the Outbox patternOutbox patternWrite an outgoing event into an outbox table in the same database transaction as the change itself; a separate process reads that table and publishes the events. Either both happen or neither does.: write the message into an outbox table in the same transaction as the change, and let a separate process publish rows from that table. (Queues §10 and Sagas §08 build it.)

With event sourcing you get this for free: the events table is the outbox. The decision and the event are the same row, so they can't disagree.

Internal events vs the public contract

Don't publish your stored events raw. They're shaped for your evolve function, and §09 showed they'll change. If other teams read them directly, every internal refactor becomes a cross-team negotiation.

Instead, translate them into an Integration eventIntegration eventA domain event re-shaped into a stable, versioned message for other contexts. Internal events can change freely; integration events are a published contract.: a small, versioned, documented message that you promise to keep stable. Several internal events may map to one public one, and some internal events (an internal SnapshotTaken, say) aren't published at all.

A publisher in Go

It has the same shape as the projector from §07: read the safe events after its own checkpoint row (publisher), do something with each event, move the checkpoint. It reuses §07's readAfter and stored (including CreatedAt), passing the database instead of a transaction.

// Publisher is whatever sends a message to your broker (Kafka, RabbitMQ, NATS…).
type Publisher interface {
	Publish(ctx context.Context, key string, msg []byte) error
}

// AccountTransactionV1 is the PUBLIC contract. Internal events can change; this can't, easily.
type AccountTransactionV1 struct {
	EventID     string    `json:"eventId"`     // stable and unique: consumers dedupe on it
	AccountID   string    `json:"accountId"`
	AmountCents int64     `json:"amountCents"` // + deposit, − withdrawal
	OccurredAt  time.Time `json:"occurredAt"`
}

// PublishBatch sends the safe events after the checkpoint, THEN saves the checkpoint.
// A crash in between re-sends the batch: at-least-once delivery.
func PublishBatch(ctx context.Context, db *sql.DB, pub Publisher) (int, error) {
	var cpTx string
	var cpPos int64
	if err := db.QueryRowContext(ctx,
		`SELECT tx_id::text, position FROM checkpoints WHERE name = 'publisher'`).Scan(&cpTx, &cpPos); err != nil {
		return 0, err
	}
	batch, err := readAfter(ctx, db, cpTx, cpPos, 100) // §07's safe read
	if err != nil {
		return 0, err
	}
	for _, e := range batch {
		cpTx, cpPos = e.TxID, e.Position
		var d struct {
			Amount int64 `json:"amount"`
		}
		if err := json.Unmarshal(e.Data, &d); err != nil {
			return 0, err
		}
		sign := int64(1)
		switch e.Type {
		case "Withdrawn":
			sign = -1
		case "Deposited":
		default:
			continue // internal-only event: not part of the public contract
		}
		msg, err := json.Marshal(AccountTransactionV1{
			EventID:     fmt.Sprintf("%s:%d", e.StreamID, e.Version), // ← same event, same ID, every time
			AccountID:   e.StreamID,
			AmountCents: sign * d.Amount,
			OccurredAt:  e.CreatedAt,
		})
		if err != nil {
			return 0, err
		}
		if err := pub.Publish(ctx, e.StreamID, msg); err != nil { // key = account → per-account order
			return 0, err // checkpoint not moved: this batch will be retried
		}
	}
	_, err = db.ExecContext(ctx,
		`UPDATE checkpoints SET tx_id = $1::xid8, position = $2 WHERE name = 'publisher'`, cpTx, cpPos)
	return len(batch), err
}

Run on this machine with a Publisher that just prints:

key=account-42 {"eventId":"account-42:2","accountId":"account-42","amountCents":10000,"occurredAt":"2026-10-30T16:11:51.911252Z"}
key=account-42 {"eventId":"account-42:3","accountId":"account-42","amountCents":-3000,"occurredAt":"2026-10-30T16:11:51.91521Z"}

Three choices in that code matter more than the rest:

1. Publish, then save the checkpoint. If the process dies after publishing but before saving, the next run publishes the same events again. Nothing is lost, but there will be duplicates: At-least-once deliveryAt-least-once deliveryA delivery promise where a message is never lost but may arrive more than once (for example, re-sent after a crash). Receivers must therefore handle duplicates safely. delivery. (The reverse order would lose messages instead.)
2. A stable event ID. account-42:3 is the same every time that event is published, so consumers can recognise a repeat. Random IDs generated at publish time would make duplicates look new.
3. Key by stream. Kafka keeps messages with the same key in the same PartitionPartition (Kafka)One ordered slice of a topic. Messages with the same key always land in the same partition, so they stay in order relative to each other., in order, so each account's transactions arrive in order. Different accounts may interleave, which is fine.

Consumers must expect duplicates

Every consumer of these messages has to be IdempotentIdempotentDoing it twice has the same effect as doing it once. “Set balance to 50” is idempotent; “add 50 to balance” is not — so only the first is safe to retry blindly.: processing the same message twice must have the effect of processing it once. The usual way is DeduplicationDeduplicationRecognising that a message or request was already handled (usually by an ID you remember) and skipping it the second time.: record each eventId you've handled (in the same transaction as your own changes) and skip ones you've seen. Idempotency §08 builds exactly this, in Go and JS.

Sagas: when one command isn't enough

Some business actions span several streams or services: "transfer $50 from account 42 to account 43" touches two accounts, and each account is its own stream with its own version check. There's no single append that covers both.

The event-driven answer is a SagaSagaA long business process split into steps, each with its own transaction, plus an “undo” step (a compensation) for each one. If step 3 fails, the compensations for steps 2 and 1 run instead of a rollback.: TransferRequested → withdraw from 42 → Withdrawn → deposit to 43 → Deposited → TransferCompleted. If the deposit fails, a compensating step puts the money back in 42. The component that listens to events and sends the next command is often called a process manager. It's a full topic of its own: Sagas, especially §06 on orchestration vs choreography.

When to use it, when not, and what it costs

Where it pays for itself

• Money and ledgers. Accounting has been event-sourced for centuries: a ledger is a list of entries, and balances are sums. Payments, wallets, billing, inventory movements.
• Audit and compliance. When "who changed this, when, and why" is a legal requirement, an Audit trailAudit trailA record of who changed what, when and why, that auditors and support staff can trust. In event sourcing the events themselves are the audit trail. that is the data can't fall out of sync with it.
• Questions about the past. Insurance ("what did the policy cover on the day of the accident?"), healthcare records, pricing disputes. Temporal queriesTemporal queryA question about the past: “what was the balance on March 3rd?” or “what did this order look like before the edit?”. Easy with events: fold only the events up to that moment. are a fold with a date filter.
• Many read shapes. Search, dashboards, per-customer views and exports over the same data, each wanting its own store. That's CQRS's home turf, with or without events.
• Debugging by replay. Copy a misbehaving stream into a test, fold it, and step through exactly what happened. No "can't reproduce".

Where it doesn't

• Forms over tables. User profiles, settings, a product catalogue edited by three people, admin panels: CRUDCRUDCreate, Read, Update, Delete: an app that mostly stores forms in tables and shows them back, with few business rules in between. is the right shape, and an updated_at column or a simple history table answers the occasional "who changed this?".
• A small team on a deadline. The patterns themselves are simple; the operations (rebuilds, lag, versioning, tooling) are not.
• Strong consistency everywhere. If every screen must show every write instantly, separate read models fight you (§08).
• Lots of deletion. If records must routinely vanish, an append-only log is working against you (§09).

Try it on a part of a system you know:

The real costs, listed honestly

Adopt it in slices

You don't choose this for a company; you choose it for one area.

1. Start with CQRS-lite. One flat read table updated in the same transaction (§03) solves most "slow dashboard" problems with almost no risk.
2. Event-source one Bounded contextBounded contextA part of the system inside which every word has exactly one meaning and one model. “Product” inside Catalog means a description with photos; inside Shipping it means a box with a weight. Each context has its own code and data. (the ledger, the order lifecycle), keep everything else CRUD, and connect them with published integration events (§10).
3. Own the tooling early: a "show me this stream" admin page, a rebuild command, a lag dashboard.

Running it: what to watch

• Projection lag = newest position in the store − each projection's checkpoint. Graph it per projection and alert when it grows for minutes; it's the first sign a projector is stuck on a bad event. (How to build such metrics and alerts: Observability.)
• Poison events. Like a Poison messagePoison messageA message that fails every time it's processed (bad data, a bug for that one input). Retrying it can never succeed and, without a limit, blocks or burns the consumer. in a queue: one event a projector can't handle (a bug, an unknown shape) stops it at that checkpoint forever. Decide up front: stop and alert (safest for money), or log, skip and mark the read model as suspect.
• Rebuild time. Measure how long a full replay of your biggest projection takes, before an incident forces you to find out.
• Stream size. Track the longest streams. One that grows without bound is a design smell (§09: closing the books).

Tooling, with caveats

• Postgres, by hand, as in this topic. A table, a unique constraint, a checkpoint table. Great to start and good for a long time; you own the §07 safe-read rule ((tx_id, position) checkpoints) and the tooling. Libraries wrap this pattern, e.g. Marten (.NET, on Postgres).

• EventStoreDB (rebranded in 2024–25 as KurrentDB): a database built for event sourcing. Streams, expected-version appends, ordered subscriptions for projections, and catch-up are built in, which removes most of §06–§07's hand-written code. The cost is one more database to run and learn.

• Frameworks such as Axon (Java) bundle an event store with command handling and projections. Productive inside their conventions, constraining outside them.

• Kafka as the event store: be careful. KafkaKafkaApache Kafka: a durable, ordered log of events, split into partitions, that many programs can publish to and read from at their own pace. is a superb log for moving events between services (§10), but it's a poor source of truth for entities:

  • It has no expected-version append per key: two writers can both append to "account 42" and Kafka will accept both. The §06 check has to live somewhere else.
  • Loading one entity means scanning a partition for its key; there's no "read stream account-42" index.
  • The default RetentionRetentionHow long a log keeps messages before deleting the oldest — e.g. 7 days or 500 GB per partition — whether or not anyone has read them. deletes old messages (7 days unless you change it), and Log compactionLog compactionA Kafka cleanup mode that keeps at least the newest message for each key and deletes older ones, so the topic holds the latest state per key instead of full history. keeps only the latest message per key: both destroy history unless configured otherwise.

  Teams that make it work usually pair Kafka with a separate store for per-entity state and checks. See Queues §07 for retention and compaction.

• Under the hood, append-heavy storage is a theme across databases: an event log is the application-level cousin of the write-ahead log and of LSM trees, which also turn every write into an append and build readable structures from it later.

Pitfalls gallery

const { balance_cents } = await readBalances(accountId); // projection: maybe stale
if (balance_cents < cents) throw new Error("insufficient funds");
await appendAnyway(pool, accountId, [{ type: "Withdrawn", amount: cents }]); // no version

const account = await load(pool, accountId);       // fold the stream: current
const events = withdraw(account, cents);            // rule runs here
await append(pool, accountId, account.version, events); // and is protected here

case "Withdrawn":
	if a.Balance < e.Amount {
		return a // "invalid", silently skipped
	}
	sendReceipt(a.Owner, e.Amount) // runs again on every load!
	a.Balance -= e.Amount

case "Withdrawn":
	a.Balance -= e.Amount // just apply the fact
// Rules live in Withdraw (decide).
// Emails live in a subscriber that reads
// events once, with its own checkpoint.

Check yourself

Predict the answer before you click. Each one is covered above: Q1 in §05, Q2 in §04, Q3 in §06, Q4 in §07, Q5 in §09, Q6 in §11.

AccountOpened {owner: "Cy"}
Deposited     {amount: 500}
Withdrawn     {amount: 200}
Deposited     {amount: 50}

Cheat sheet