Event store and concurrency

The event store does two things: append events to the end of a stream, and load a stream in order. Everything else, such as decoding payloads and rebuilding aggregates, sits above it in es.Repository. A small interface is what lets three very different databases implement it the same way.

// ErrConcurrency means the stream changed after it was loaded. The caller
// may reload and retry. It is an apperr.ErrConflict.
var ErrConcurrency = apperr.New(apperr.ErrConflict, "this was changed by another request; reload and try again")

// Store keeps event streams.
type Store interface {
  // Append adds records to the end of a stream as part of tx. Their
  // versions run from expectedVersion+1 up. When the stream's current
  // version isn't expectedVersion, Append writes nothing and returns
  // ErrConcurrency, including when a concurrent transaction appended the
  // same version first.
  Append(ctx context.Context, tx uow.Tx, streamID uuid.UUID, expectedVersion int, records []Record) error
  // Load returns a stream's committed records in version order, or none
  // for a stream that doesn't exist.
  Load(ctx context.Context, streamID uuid.UUID) ([]Record, error)
}

Optimistic concurrency

Two requests can load the same user, change it, and save. Without a check, both appends succeed and one change silently overwrites the other’s assumptions.

So every append says which version it expects the stream to be at. The repository passes the version the aggregate was loaded at. If another request appended first, the stream is ahead, nothing is written, and Append returns es.ErrConcurrency:

func TestRepositoryConflict(t *testing.T) {
  repo, db, _ := newRepository(nil)
  id := uuid.New()
  c := &counter{}
  _ = es.Raise(c, opened{ID: id})
  if _, err := save(t.Context(), db, repo, c); err != nil {
    t.Fatal(err)
  }

  first, _ := repo.Load(t.Context(), id)
  second, _ := repo.Load(t.Context(), id)
  _ = first.increment(1)
  _ = second.increment(1)
  if _, err := save(t.Context(), db, repo, first); err != nil {
    t.Fatal(err)
  }
  _, err := save(t.Context(), db, repo, second)
  if !errors.Is(err, es.ErrConcurrency) || !errors.Is(err, apperr.ErrConflict) {
    t.Fatalf("second writer: err = %v, want ErrConcurrency, an ErrConflict", err)
  }
}

ErrConcurrency is an apperr.ErrConflict, so it reaches an HTTP client as a 409 with the message “this was changed by another request; reload and try again”. The client reloads and decides again. Stores never retry it themselves: only the caller knows whether the change still makes sense.

One transaction for everything

Append takes a uow.Tx. Projections and the outbox take the same one, so a change’s events, its read models and the messages it causes commit together:

// Tx is one open transaction. Each backend has its own type; its stores
// unwrap it and return an error for another backend's Tx.
type Tx interface {
  // Backend names the backend that opened the transaction, for errors.
  Backend() string
}

// UnitOfWork runs functions in transactions.
type UnitOfWork interface {
  // Do runs fn in a new transaction. It commits when fn returns nil and
  // rolls back when fn returns an error, which Do then returns. A backend
  // may run fn more than once to retry a transient failure, so fn must do
  // nothing but write through tx.
  Do(ctx context.Context, fn func(tx Tx) error) error
}

A backend may run the function more than once to retry a transient failure, such as a MongoDB write conflict. That is why the function must only write through tx and do nothing else: no HTTP calls and no emails. A side effect goes into the outbox instead.

The repository is configured once per aggregate type, with its store, registry and projection:

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
}

The backends

Backend	Package	How the version check works
Memory	`internal/platform/memdb`	A mutex-guarded map. A transaction holds the write lock and undoes its writes if it fails. Data is lost when the process stops.
MySQL	`internal/platform/mysql`	An `events` table with `UNIQUE KEY events_stream_version (stream_id, version)`. Append checks the head version inside the transaction, and a duplicate key also maps to `es.ErrConcurrency`.
MongoDB	`internal/platform/mongodb`	An `events` collection with a unique index on `{stream_id: 1, version: 1}`, written in a session transaction. Needs a replica set, even a single-node one.

Loading always sorts by version, never by time. Timestamps are kept to the millisecond, the finest every backend stores, and two events can share one.

The contract suite

Every backend runs the same tests: append then load, continue a stream, a stale or future version conflicts, two concurrent appends where exactly one wins, a rollback leaves nothing, an unknown stream is empty, and streams don’t affect each other.

// OpenStore returns an empty store and the unit of work its transactions
// come from. Each call must return a store that shares no data with others.
type OpenStore func(t *testing.T) (es.Store, uow.UnitOfWork)

// TestStore checks the rules every es.Store must follow. Each backend's
// tests call it:
//
//  func TestEventStore(t *testing.T) {
//    estest.TestStore(t, func(t *testing.T) (es.Store, uow.UnitOfWork) {
//      db := memdb.New()
//      return memdb.NewEventStore(db), db
//    })
//  }
func TestStore(t *testing.T, open OpenStore) {
  t.Run("AppendThenLoad", func(t *testing.T) { testAppendThenLoad(t, open) })
  t.Run("AppendContinuesStream", func(t *testing.T) { testAppendContinuesStream(t, open) })
  t.Run("StaleVersionConflicts", func(t *testing.T) { testVersionConflicts(t, open, 1) })
  t.Run("FutureVersionConflicts", func(t *testing.T) { testVersionConflicts(t, open, 3) })
  t.Run("ConcurrentAppendsOneWins", func(t *testing.T) { testConcurrentAppends(t, open) })
  t.Run("RollbackDiscardsAppend", func(t *testing.T) { testRollback(t, open) })
  t.Run("UnknownStreamIsEmpty", func(t *testing.T) { testUnknownStream(t, open) })
  t.Run("StreamsAreIndependent", func(t *testing.T) { testIndependentStreams(t, open) })
}

A backend’s own test only says how to open it:

func TestEventStore(t *testing.T) {
  estest.TestStore(t, func(*testing.T) (es.Store, uow.UnitOfWork) {
    db := memdb.New()
    return memdb.NewEventStore(db), db
  })
}

func TestOutbox(t *testing.T) {
  outboxtest.TestStore(t, func(*testing.T) (outbox.Store, uow.UnitOfWork) {
    db := memdb.New()
    return memdb.NewOutbox(db), db
  })
}

The MySQL and MongoDB suites start a container per test binary through testcontainers, and skip under go test -short. CI runs them in full.

Adding a backend

To keep events in another database, such as DynamoDB or PostgreSQL, write a package that provides:

a uow.UnitOfWork whose Tx your stores can unwrap,
an es.Store and an outbox.Store that write through that Tx,
a test that passes estest.TestStore and outboxtest.TestStore,
the service stores (the user projection and login codes, auth’s clients and grants), each passing its service’s contract suite.

Then add a cmd/<service>/store_<backend>.go with a new build tag that opens it. Nothing above the store changes. If the suite passes, the backend follows the same rules as the other three.