Aggregates and events

An aggregate is a consistency boundary: one thing whose rules can be checked without looking at anything else, such as a user. It is rebuilt from its own stream of events, and it changes only by raising new ones.

Why events rather than rows? Because the history is the data. You can answer “when did this user change their email, and why?” from the stream, build a new read model from old events, and test a rule by saying what happened before and what should happen next.

An aggregate

An aggregate embeds es.Changes, which tracks its version and the events it raised. Apply changes state from one event and only assigns. Rules are checked in the methods that raise events, before the event exists. This is the user service’s User:

// User is someone who can sign in: an address, and at most one Google
// account.
type User struct {
  es.Changes
  id            uuid.UUID
  email         Email
  googleSubject string
}

// New returns an empty User, to register or to rebuild from its stream.
func New() *User { return &User{} }

// ID implements es.Aggregate.
func (u *User) ID() uuid.UUID { return u.id }

// ChangeEmail moves the account to the address to.
func (u *User) ChangeEmail(to Email) error {
  switch {
  case !u.registered():
    return errNotRegistered
  case to.IsZero():
    return apperr.Invalid("email", "is required")
  case to == u.email:
    return ErrSameEmail
  }
  return es.Raise(u, EmailChanged{Old: u.email, New: to})
}

// Apply implements es.Aggregate.
func (u *User) Apply(event any) error {
  switch e := event.(type) {
  case Registered:
    u.id, u.email = e.UserID, e.Email
  case EmailChanged:
    u.email = e.New
  case GoogleLinked:
    u.googleSubject = e.Subject
  default:
    return fmt.Errorf("user: can't apply %T", event)
  }
  return nil
}

The kit gives you three functions around it. Raise applies an event and records it for the next save. Replay rebuilds an aggregate from stored events. Pending lists what was raised since the aggregate was loaded:

// Changes tracks an aggregate's stored version and the events it raised
// since it was loaded.
type Changes struct {
  version int
  pending []any
}

// Raise applies event to a and records it, to be stored by the next Save.
func Raise(a Aggregate, event any) error {
  if err := a.Apply(event); err != nil {
    return err
  }
  c := a.changes()
  c.pending = append(c.pending, event)
  return nil
}

// Replay rebuilds a from stored events, as Repository.Load does. Tests use
// it to set up an aggregate's history.
func Replay(a Aggregate, events ...any) error {
  for _, e := range events {
    if err := a.Apply(e); err != nil {
      return err
    }
  }
  a.changes().version += len(events)
  return nil
}

// Pending returns the events a raised that aren't stored yet.
func Pending(a Aggregate) []any {
  return slices.Clone(a.changes().pending)
}

Events

An event is a plain struct with JSON tags, named in the past tense. It holds what the fact needs and nothing else:

// Registered is the first event of every user: an account now exists for
// the address.
type Registered struct {
  UserID uuid.UUID `json:"user_id"`
  Email  Email     `json:"email"`
}

// EmailChanged means the user confirmed a new address.
type EmailChanged struct {
  Old Email `json:"old"`
  New Email `json:"new"`
}

// GoogleLinked means the user can sign in with the Google account whose
// subject (the ID token's sub) this is.
type GoogleLinked struct {
  Subject string `json:"subject"`
}

Stored, every event travels in an envelope. Meta says why it happened, using identifiers only:

// Event is one fact in an aggregate's stream, with its payload decoded.
type Event struct {
  ID         uuid.UUID
  StreamID   uuid.UUID
  StreamType string
  // Version is the event's position in its stream, starting at 1.
  Version    int
  Type       string
  OccurredAt time.Time
  Payload    any
  Meta       Meta
}

// Meta says why an event happened. Events are kept forever, so Meta holds
// identifiers only: never tokens, IP addresses or other personal data.
type Meta struct {
  // CorrelationID ties together everything one request caused.
  CorrelationID string `json:"correlation_id,omitzero" bson:"correlation_id,omitempty"`
  // CausationID is the ID of the event or message that caused this one.
  CausationID string `json:"causation_id,omitzero" bson:"causation_id,omitempty"`
  // ActorID is the user or client that caused the event, if any.
  ActorID string `json:"actor_id,omitzero" bson:"actor_id,omitempty"`
}

The user service’s events are user.registered.v1 (user_id, email), user.email_changed.v1 (old, new) and user.google_linked.v1 (subject). The aggregate enforces that a user has at most one Google subject.

Names and versions

Each payload type is registered under an explicit name. The name is stored with the event, so it never changes. Registering by name rather than by Go type means renaming a type can’t orphan the events already stored:

// RegisterEvents names the user events in r. The names are stored with
// every event, so they never change.
func RegisterEvents(r *es.Registry) {
  es.Register[Registered](r, "user.registered.v1")
  es.Register[EmailChanged](r, "user.email_changed.v1")
  es.Register[GoogleLinked](r, "user.google_linked.v1")
}

When a payload’s shape has to change, the new shape gets a new name and an upcaster converts the old JSON as it is read. The stored events stay exactly as they were written:

// Register names the payload type T. Names are stored with every event, so
// they never change: "user.registered.v1". A new shape gets a new name and
// an Upcast from the old one.
//
// Register panics on an empty or repeated name, a type registered twice, or
// a T that isn't a struct, because each is a programming error.
func Register[T any](r *Registry, name string) {
  t := reflect.TypeFor[T]()
  switch {
  case name == "":
    panic("es: Register with an empty name")
  case t.Kind() != reflect.Struct:
    panic(fmt.Sprintf("es: Register %q: payload must be a struct, not %s", name, t))
  }
  if _, ok := r.types[name]; ok {
    panic(fmt.Sprintf("es: event type %q is already registered", name))
  }
  if _, ok := r.upcasters[name]; ok {
    panic(fmt.Sprintf("es: event type %q is upcast to another type", name))
  }
  if other, ok := r.names[t]; ok {
    panic(fmt.Sprintf("es: %s is already registered as %q", t, other))
  }
  r.types[name] = t
  r.names[t] = name
}

// Upcast reads events stored as from as events of type to, converting the
// JSON payload with fn. It lets a payload change shape while the stored
// events stay as they were written. Upcasts may chain.
func (r *Registry) Upcast(from, to string, fn func(payload []byte) ([]byte, error)) {
  if _, ok := r.types[from]; ok {
    panic(fmt.Sprintf("es: Upcast %q: it is a registered type", from))
  }
  if _, ok := r.upcasters[from]; ok {
    panic(fmt.Sprintf("es: Upcast %q: already upcast", from))
  }
  r.upcasters[from] = upcaster{to: to, fn: fn}
}

The user events have no second version yet, so this example comes from the kit’s own tests and their small counter aggregate. Upcasts chain, so a stream written by three versions of the code still loads:

func TestUpcast(t *testing.T) {
  r := registry()
  // v1 of incremented stored the step as "step"; v2 renamed it, and v3 is the current shape.
  r.Upcast("counter.incremented.v0", "counter.incremented.v00", func(old []byte) ([]byte, error) {
    return []byte(strings.Replace(string(old), `"step"`, `"by"`, 1)), nil
  })
  r.Upcast("counter.incremented.v00", "counter.incremented.v1", func(old []byte) ([]byte, error) {
    var v map[string]int
    if err := json.Unmarshal(old, &v); err != nil {
      return nil, err
    }
    return json.Marshal(map[string]int{"by": v["by"] * 2})
  })

  got, err := r.Decode("counter.incremented.v0", []byte(`{"step":2}`))
  if err != nil {
    t.Fatal(err)
  }
  if got != (incremented{By: 4}) {
    t.Errorf("decoded %#v, want by 4", got)
  }
}

Rules for a permanent log

Stored events are kept forever. The rules in internal/platform/es/AGENTS.md follow from that:

Events are immutable facts. To fix data, append a correcting event; never edit or delete a stored one.
A registered name never changes, and neither does the shape stored under it. A new shape gets .v2 and an upcast from .v1.
Events hold no secrets: no tokens, passwords or codes. They hold no personal data beyond what the fact needs, and Meta holds identifiers only.
Apply only assigns. Validation happens before Raise.
Decoding an event type that isn’t registered returns es.ErrUnknownEventType, never a panic.

Next: how events are stored, and what happens when two requests change the same aggregate, in Event store and concurrency.