Add a query or projection

Queries never read the event store. They read projections: tables or collections shaped for the question, kept current by a function that runs in the same transaction as every append. Read CQRS and projections first for why.

Decide the shape from the query. Write down the question and the lookup it needs: “find a user by email” needs a unique index on email. Add only the columns the query reads.
Change the schema on each backend.
- MySQL: add a goose migration under internal/user/store/mysql/migrations/. Migrations run with user migrate, from the Helm hook in a cluster, or at start when MIGRATE_ON_START=true.
- MongoDB: add the index to Indexes in internal/user/store/mongodb. The same user migrate, or MIGRATE_ON_START, ensures them, and ensuring them again changes nothing.
- Memory: add the field in internal/user/store/memory and, for a unique rule, a map that enforces it.

Update the projection function. It receives the events of one save and the transaction. Switch on each payload type, and write with upserts so that running it twice changes nothing. The user stores’ Project refuses an event it has no case for, so every event needs a case in each backend, even one that changes nothing. In MongoDB the write is the bson.D in project; userDoc is only for reads.

// 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
}

The repository calls it on every save:

func newRepository(project es.Projection) (*es.Repository[*counter], *memdb.DB, *memdb.EventStore) {
  db := memdb.New()
  store := memdb.NewEventStore(db)
  at := time.Date(2026, 10, 24, 9, 0, 0, 123_456_789, time.UTC)
  return es.NewRepository(es.RepositoryConfig[*counter]{
    StreamType: "counter",
    New:        func() *counter { return &counter{} },
    Store:      store,
    Registry:   registry(),
    Project:    project,
    Now:        func() time.Time { return at },
  }), db, store
}

Map a duplicate key to a conflict. A unique index is how a projection enforces a rule across aggregates. Turn the backend’s duplicate-key error into apperr.ErrConflict, and the client sees a 409. The whole save rolls back, events included:

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))
  }
}

Add the rule to the store contract suite in internal/user/store/storetest, then make each backend pass it. A rule that lives in one backend only is a bug waiting for someone to switch stores.
Write the query. A query method on the store reads the projection, and a use case or handler calls it. Queries take the actor from the identity, like commands, and only return what that actor may see.
Run the backend suites:
Terminal window
```
go test -race ./internal/user/store/...
```
Without -short, this starts MySQL and MongoDB containers. With it, only the memory store runs.

Rebuilding a projection

Events are kept, so a projection added to a running system can be backfilled from them. es.Store has no way to list every stream, so a backfill reads the backend’s events table or collection directly, stream by stream in version order, and runs the projection function over each. After that, the inline projection keeps it current. Make the backfill idempotent, for the same reason the projection is. No backfill tool ships with the repository.