Repository and UnitOfWork

Package: aggregate

The repository loads aggregates, tracks them in a UnitOfWork, and commits their uncommitted events through the EventStore and the publisher.

Construction and factories

repo := aggregate.NewRepository(store, publisher)
repo.RegisterFactory("bank.Account", func(id uuid.UUID) aggregate.Root {
    return NewAccount(id)
})

A factory builds an empty aggregate of a given type. Load needs one to rehydrate, and snapshot rebuild needs it too. With the façade, app.RegisterAggregate(type, factory) does this for you.

Fields: Store, Snapshots (optional), Publisher, Factories.

UnitOfWork

A UnitOfWork lives on the context of one Execute. It is an identity map plus an ordered list of aggregates.

• Add and Load need a UoW on the context. Without one they return aggregate: no UnitOfWork in context.
• Load returns the instance already in the UoW if the id is known. You never get two mutable copies of one aggregate in one Execute.
• Commit collects uncommitted events from tracked aggregates in load order.
• defer uow.Clear() in Execute empties it when the call ends, success or not.

Do not share a UoW, or any aggregate it holds, across goroutines. Parallelism belongs between requests, not inside one.

Add

acc := NewAccount(id)
if err := acc.OpenAccount("alice", "CNY", opening); err != nil {
    return err
}
return repo.Add(ctx, acc)

Add tracks a new aggregate. Its events are stored at commit.

Load

agg, err := repo.Load(ctx, id, "bank.Account")

Load does the following:

1. Return the tracked instance if there is one.
2. Find the factory for the aggregate type. Error if none: aggregate: no factory for type.
3. If a SnapshotStore is set, load the latest snapshot whose version matches SnapshotVersion(), restore it, and set the aggregate version to the snapshot sequence.
4. Load the stream through Store.LoadStream. Events at or before the snapshot sequence are skipped. ErrStreamNotFound is returned if there is no stream and no snapshot.
5. Replay the remaining events with LoadFromHistory.
6. Check Invariants().
7. Track the aggregate in the UoW.

LoadStream verifies the hash chain while it reads, so tampering surfaces here as ErrHashChainTamper.

Commit

Commit runs once per command, after the handler returns:

1. For each tracked aggregate with uncommitted events, run Invariants().
2. Build a StreamAppend per aggregate: id, type, PartitionKey, ExpectedVersion: agg.Version(), events, UniqueKeys.
3. Call Store.Append(ctx, commandRecord, streams). It writes the command row and all streams atomically, and is where OCC and unique keys are enforced.
4. Call Publisher.Publish with all events, in load order and sequence order. With SyncInTxn, projectors and workflows run here, inside the transaction.
5. Mark each aggregate committed. Its version advances and its uncommitted list clears.

If a step fails, Commit returns the error. Execute rolls back the transaction. If nothing was uncommitted, Commit does nothing.

Event order is load order of aggregates first, then sequence order within each. It does not depend on interleaved Apply calls across aggregates.

Exists

ok, err := repo.Exists(ctx, id)

Reports whether a stream has any events.

Errors you will see

Related

• Write-path consistency
• Snapshotting
• CommandService

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