CQRS and projections

CQRS splits the code that changes things from the code that reads them. Commands go through an aggregate and append events. Queries read projections: tables shaped for the question being asked, built from those events.

The split matters because an event stream is a poor thing to query. “Which user has this email?” would mean replaying every stream. A users table with a unique index on email answers it in one lookup, and the index also enforces the rule.

Projections run in the append transaction

A projection is a function the repository calls on every save, with the events just appended and the same transaction:

internal/platform/es/repository.go

// Projection updates read models from newly stored events, in the
// transaction that appends them. A failed projection rolls the append back,
// so read models never lag their streams, and a unique index in a read model
// can enforce a rule across aggregates, such as one account per email.
type Projection func(ctx context.Context, tx uow.Tx, events []Event) error

// RepositoryConfig configures a Repository.
type RepositoryConfig[A Aggregate] struct {
  // StreamType names the aggregate's kind, stored with every event: "user".
  StreamType string
  // New returns an empty aggregate to replay events into.
  New      func() A
  Store    Store
  Registry *Registry
  // Project, if set, runs on every Save.
  Project Projection
  // Now stamps events. It defaults to time.Now.
  Now func() time.Time
}

That gives two guarantees:

• Read your writes. When a request returns, the projection already shows its change. There is no window where the user exists in the stream but not in the table.
• Rules across aggregates. One account per email can’t be checked inside a single aggregate. A unique index in the projection checks it, and a duplicate fails the whole transaction, events included:

internal/platform/es/es_test.go

func TestRepositoryProjectionFailureRollsBack(t *testing.T) {
  failure := errors.New("email taken")
  repo, db, store := newRepository(func(context.Context, uow.Tx, []es.Event) error { return failure })
  c := &counter{}
  id := uuid.New()
  _ = es.Raise(c, opened{ID: id})

  if _, err := save(t.Context(), db, repo, c); !errors.Is(err, failure) {
    t.Fatalf("err = %v, want the projection's error", err)
  }
  if recs, _ := store.Load(t.Context(), id); len(recs) != 0 {
    t.Fatalf("a failed projection kept %d events", len(recs))
  }
}

The user service turns that duplicate-key error into apperr.ErrConflict, which the client sees as a 409.

The user service’s read side

Table	Columns	Rules
users	id, email, google_sub, created_at	email unique. google_sub unique when set, so any number of users can have none.
login_codes	code_hash, purpose, email, user_id, next, expires_at, used_at	Consumed atomically: a code works once, and two concurrent uses can’t both succeed.

In MySQL, google_sub is NULL for a user without Google, and a unique key allows any number of NULLs. In MongoDB, the unique index on google_sub is partial: it covers only documents where it is a string. MongoDB also removes expired login codes itself, through a TTL index on expires_at; every backend checks the time when a code is used regardless.

Login codes aren’t a projection: they are short-lived state written directly by the use cases. They live beside the projection because they share its transaction.

Writing a projection

• Make it idempotent. Running it twice on the same events must leave the same rows. Upserts by ID, not blind inserts.
• Switch on the payload type. The user stores refuse an event they have no case for, so a new event can’t be saved until every backend handles it.
• Write one implementation per backend, each under the service’s store contract suite.
• Keep it in the transaction. It must not call other services or send anything. A side effect is an outbox message.

Add a projection walks through one end to end.

When a projection must be asynchronous

Inline projections are the default because they need no checkpoints, no replay machinery and no eventual-consistency handling in the web app. Sometimes a projection is too slow to sit in the request, or it lives in another system, such as a search index.

Then move it onto the outbox:

1. In the use case, add an outbox message for the projection in the same transaction as the events.
2. Register a relay handler for that topic that updates the read model.
3. Make the handler idempotent by message ID, because the relay delivers at least once.

The read model now lags by the relay’s poll interval, and the web app has to live with that. Outbox and side effects covers delivery and retries.