Skip to content
go-sequentCQRS + Event Sourcing for Go financial systems

Reference

AggregateRoot

Reference for go-sequent aggregates: the Root interface, embedding aggregate.Base, pure appliers, Apply, invariants, unique keys, partition keys and uncommitted events.

AggregateRoot

Package: aggregate

An aggregate root is the consistency boundary of your domain model. Its state is the fold of its event stream. Commands change it by calling methods that check rules and then Apply events.

The interface

type Root interface {
    ID() uuid.UUID
    AggregateType() string
    UncommittedEvents() []event.Event
    ClearUncommitted()
    Version() int64
    LoadFromHistory(events []event.Event) error
    Invariants() error
    UniqueKeys() map[string]any
    PartitionKey() string
}

You rarely implement this by hand. Embed aggregate.Base, which supplies all of it. Invariants returns nil by default and UniqueKeys returns whatever you stored with SetUniqueKeys; override either on your aggregate when you need them. Applying an event that has no registered applier panics.

Rules

  1. Appliers are pure state changes. They set fields from the event. No I/O, no clock reads, no random values.
  2. Command methods validate, then Apply. A method either returns an error and emits nothing, or emits one or more events.
  3. No cross-aggregate reads while deciding. The aggregate uses only its own state and the arguments it receives.

Rule 1 is what makes replay correct. Loading history must yield the same state as the live run that produced it.

Embedding aggregate.Base

type Account struct {
    aggregate.Base
    Owner    string
    Currency string
    Balance  types.Money
    Open     bool
}

func NewAccount(id uuid.UUID) *Account {
    a := &Account{}
    a.Init(id, "bank.Account")
    a.On(&AccountOpened{}, func(e event.Event) {
        ev := e.(*AccountOpened)
        a.Owner, a.Currency, a.Balance, a.Open = ev.Owner, ev.Currency, ev.Opening, true
    })
    return a
}
  • Init(id, typeName) sets the aggregate id and type name. Call it in the constructor before any Apply.
  • On(sample, applier) registers an applier for the Go type of sample. Pass a pointer to an empty event struct.
  • Apply(event) runs the applier, sets the event’s aggregate id, sequence number and time, and appends it to the uncommitted list.
  • Version() is the last committed sequence number. It is the ExpectedVersion for the next append.

Command methods

func (a *Account) Deposit(amount types.Money) error {
    if !a.Open {
        return fmt.Errorf("account not open")
    }
    if amount.IsZero() || amount.IsNegative() {
        return fmt.Errorf("deposit amount must be positive")
    }
    if amount.Currency != a.Currency {
        return fmt.Errorf("currency mismatch")
    }
    a.Apply(NewMoneyDeposited(amount))
    return nil
}

Return errors for business rule violations. A failed method leaves nothing in the uncommitted list, and the transaction rolls back.

Invariants

Invariants() error holds rules that must be true of the state. The repository checks it after Load, and again for every aggregate with uncommitted events right before append.

func (a *Account) Invariants() error {
    if a.Open && a.Balance.IsNegative() {
        return fmt.Errorf("invariant: balance negative")
    }
    return nil
}

Unique keys

UniqueKeys() returns a map from scope to value. The store enforces uniqueness on append and reports ErrUniqueKeyViolation on a clash. Base provides SetUniqueKeys so the aggregate can refresh keys from an applier. Include the tenant in the scope. See Unique keys and OCC.

Partition key

PartitionKey() returns a hint stored with the aggregate. For multi-tenant systems it is usually derived from the tenant:

acc.SetPartitionKey(tenancy.PartitionKey(tenantID))

It supports physical partitioning of events and tenant-scoped unique keys inside the shared schema. See Tenancy and Partitioning.

Uncommitted events

Events from Apply sit in UncommittedEvents() until Repository.Commit appends them. After a successful commit the repository clears the list and advances the aggregate’s version. You do not clear it yourself.

Snapshots

An aggregate can opt in to JSON snapshots by implementing SnapshotVersion, MarshalSnapshot and UnmarshalSnapshot. See Snapshotting.

Concurrency

Never share an aggregate instance between goroutines. Each Execute has its own UnitOfWork, and the aggregates it loads belong to that call.

Testing

testkit.For(id, factory).Given(events...).When(fn).Run() rebuilds an aggregate from events and captures what a method produces. See Testing.

Docs track the go-sequent library. When APIs drift, the source wins.