repository

Documentation

Index

• Variables
• type ChangeDiscarder
• type IsRetryable
• type Option
  • func AfterInsert(fn func(context.Context, aggregate.Aggregate) error) Option
  • func BeforeInsert(fn func(context.Context, aggregate.Aggregate) error) Option
  • func ModifyQueries(...) Option
  • func OnDelete(fn func(context.Context, aggregate.Aggregate) error) Option
  • func OnFailedInsert(fn func(context.Context, aggregate.Aggregate, error) error) Option
  • func ValidateConsistency(validate bool) Option
  • func WithSnapshots(store snapshot.Store, s snapshot.Schedule) Option
• type Repository
  • func New(store event.Store, opts ...Option) *Repository
  • func (r *Repository) Delete(ctx context.Context, a aggregate.Aggregate) error
  • func (r *Repository) Fetch(ctx context.Context, a aggregate.Aggregate) error
  • func (r *Repository) FetchVersion(ctx context.Context, a aggregate.Aggregate, v int) error
  • func (r *Repository) Query(ctx context.Context, q aggregate.Query) (<-chan aggregate.History, <-chan error, error)
  • func (r *Repository) Save(ctx context.Context, a aggregate.Aggregate) error
  • func (r *Repository) Use(ctx context.Context, a aggregate.Aggregate, fn func() error) error
• type RetryTrigger
  • func RetryApprox(interval, deviation time.Duration, maxTries int) RetryTrigger
  • func RetryEvery(interval time.Duration, maxTries int) RetryTrigger
• type RetryTriggerFunc
• type Retryer
• type TypedRepository
  • func NewOf[Aggregate aggregate.TypedAggregate](r aggregate.Repository, makeFunc func(uuid.UUID) Aggregate) *TypedRepository[Aggregate]
  • func Typed[Aggregate aggregate.TypedAggregate](r aggregate.Repository, makeFunc func(uuid.UUID) Aggregate) *TypedRepository[Aggregate]
  • func (r *TypedRepository[Aggregate]) Delete(ctx context.Context, a Aggregate) error
  • func (r *TypedRepository[Aggregate]) Fetch(ctx context.Context, id uuid.UUID) (Aggregate, error)
  • func (r *TypedRepository[Aggregate]) FetchVersion(ctx context.Context, id uuid.UUID, version int) (Aggregate, error)
  • func (r *TypedRepository[Aggregate]) NewFunc() func(uuid.UUID) Aggregate
  • func (r *TypedRepository[Aggregate]) Query(ctx context.Context, q aggregate.Query) (<-chan Aggregate, <-chan error, error)
  • func (r *TypedRepository[Aggregate]) Repository() aggregate.Repository
  • func (r *TypedRepository[Aggregate]) Save(ctx context.Context, a Aggregate) error
  • func (r *TypedRepository[Aggregate]) Use(ctx context.Context, id uuid.UUID, fn func(Aggregate) error) error

Variables

var (
	// ErrVersionNotFound is returned when trying to fetch an aggregate with a
	// version higher than the current version of the aggregate.
	ErrVersionNotFound = errors.New("version not found")

	// ErrDeleted is returned when trying to fetch an aggregate that has been soft-deleted.
	ErrDeleted = errors.New("aggregate was soft-deleted")
)

Types

type ChangeDiscarder

type ChangeDiscarder interface {
	DiscardChanges()
}

A ChangeDiscarder discards changes to the aggregate. The DiscardChanges() method is called each time a Repository.Use() call is retried for the aggregate.

type IsRetryable

type IsRetryable func(error) bool

IsRetryable is a function that determines if an error within a Reposistory.Use() call is retryable.

type Option

type Option func(*Repository)

Option is a repository option.

func AfterInsert

func AfterInsert(fn func(context.Context, aggregate.Aggregate) error) Option

AfterInsert returns an Option that adds fn as a hook to a Repository. fn is called after the changes to an aggregate are inserted into the event store.

func BeforeInsert

func BeforeInsert(fn func(context.Context, aggregate.Aggregate) error) Option

BeforeInsert returns an Option that adds fn as a hook to a Repository. fn is called before the changes to an aggregate are inserted into the event store.

func ModifyQueries

func ModifyQueries(mods ...func(ctx context.Context, q aggregate.Query, prev event.Query) (event.Query, error)) Option

ModifyQueries returns an Option that adds mods as Query modifiers to a Repository. When the Repository builds a Query, it is passed to every modifier before the event store is queried.

func OnDelete

func OnDelete(fn func(context.Context, aggregate.Aggregate) error) Option

OnDelete returns an Option that adds fn as a hook to a Repository. fn is called after an aggregate has been deleted.

func OnFailedInsert

func OnFailedInsert(fn func(context.Context, aggregate.Aggregate, error) error) Option

OnFailedInsert returns an Option that adds fn as a hook to a Repository. fn is called when the Repository fails to insert the changes to an aggregate into the event store.

func ValidateConsistency

func ValidateConsistency(validate bool) Option

ValidateConsistency returns an Option that configures a Repository to validate the consistency of aggregate events before inserting the events into the event store. Defaults to true.

func WithSnapshots

func WithSnapshots(store snapshot.Store, s snapshot.Schedule) Option

WithSnapshots returns an Option that add a Snapshot Store to a Repository.

A Repository that has a Snapshot Store will fetch the latest valid Snapshot for an aggregate before fetching the necessary events to reconstruct the state of the Agrgegate.

An optional Snapshot Schedule can be provided to instruct the Repository to make and save Snapshots into the Snapshot Store when appropriate:

var store snapshot.Store
r := repository.New(store, snapshot.Every(3))

The example above will make a Snapshot of an aggregate every third version of the aggregate.

Aggregates must implement snapshot.Marshaler & snapshot.Unmarshaler in order for Snapshots to work.

type Repository

type Repository struct {
	// contains filtered or unexported fields
}

func New

func New(store event.Store, opts ...Option) *Repository

New returns an event-sourced aggregate Repository. It uses the provided event Store to persist and query aggregates.

func (*Repository) Delete

func (r *Repository) Delete(ctx context.Context, a aggregate.Aggregate) error

Delete deletes an aggregate by deleting its events from the event store.

func (*Repository) Fetch

func (r *Repository) Fetch(ctx context.Context, a aggregate.Aggregate) error

Fetch fetches the events of the provided aggregate from the event store and applies them to it to build its current state.

It is allowed to pass an aggregate that does't have any events in the event store yet.

It is also allowed to pass an aggregate that has already events applied onto it. Only events with a version higher than the current version of the passed Aggregate are fetched from the event store.

func (*Repository) FetchVersion

func (r *Repository) FetchVersion(ctx context.Context, a aggregate.Aggregate, v int) error

FetchVersion does the same as r.Fetch, but only fetches events up until the given version v. If the event store has no event for the provided aggregate with the requested version, ErrVersionNotFound is returned.

func (*Repository) Query

func (r *Repository) Query(ctx context.Context, q aggregate.Query) (<-chan aggregate.History, <-chan error, error)

Query queries the event store for events that match the given Query and returns a stream of aggregate Histories and errors. Use the returned Histories to build the current state of the queried aggregates:

var r *Repository
str, errs, err := r.Query(context.TODO(), query.New(...))
// handle err
histories, err := streams.Drain(context.TODO(), str, errs)
// handle err
for _, his := range histories {
	aggregateName := his.AggregateName()
	aggregateID := his.AggregateID()

	// Create the aggregate from its name and UUID
	foo := newFoo(aggregateID)

	// Then apply its History
	his.Apply(foo)
}

func (*Repository) Save

func (r *Repository) Save(ctx context.Context, a aggregate.Aggregate) error

Save saves the changes to an aggregate into the underlying event store and flushes its changes afterwards (by calling a.FlushChanges).

func (*Repository) Use

func (r *Repository) Use(ctx context.Context, a aggregate.Aggregate, fn func() error) error

Use first fetches the aggregate a, then calls fn(a) and finally saves the aggregate. If the RetryUse() option is used, Use() is retried up to the configured maxTries option.

type RetryTrigger

type RetryTrigger interface {
	// contains filtered or unexported methods
}

A RetryTrigger triggers a retry of Repository.Use().

func RetryApprox

func RetryApprox(interval, deviation time.Duration, maxTries int) RetryTrigger

RetryApprox returns a RetryTrigger that retries approximately every interval up to maxTries. The provided deviation is used to randomize the interval. If the interval is 1s and deviation is 100ms, then the retry is triggered after somewhere between 900ms to 1100ms.

func RetryEvery

func RetryEvery(interval time.Duration, maxTries int) RetryTrigger

RetryEvery returns a RetryTrigger that retries every interval up to maxTries.

type RetryTriggerFunc

type RetryTriggerFunc func(context.Context) error

RetryTriggerFunc allows a function to be used as a RetryTrigger.

type Retryer

type Retryer interface {
	RetryUse() (RetryTrigger, IsRetryable)
}

Retryer is an aggregate that can retry a failed Repository.Use() operation. If the RetryUse() method of the aggregate returns a non-nil RetryTrigger rt, failed Repository.Use() calls will be retried until rt.next() returns a non-nil error. The returned IsRetryable function is used to determine if the error that made Repository.Use() fail is retryable.

The following example retries calls that failed due to consistency errors. It is retried every second, up to 3 times before giving up. If the call does not succeed after 3 tries, the error of the last attempt is returned to the caller.

type Foo struct { *aggregate.Base }
func (f *Foo) RetryUse() (rt repository.RetryTrigger, isRetryable repository.IsRetryable) {
  return repository.RetryEvery(time.Second, 3), aggregate.IsConsistencyError
}

type TypedRepository

type TypedRepository[Aggregate aggregate.TypedAggregate] struct {
	// contains filtered or unexported fields
}

TypedRepository implements aggregate.TypedRepository. Use Typed to create a typed repository from an underlying repository.

type Foo struct { *aggregate.Base }
func NewFoo(id uuid.UUID) *Foo { return &Foo{Base: aggregate.New("foo", id)} }

var repo aggregate.Repository
typed := repository.Typed(repo, NewFoo)

Now you can use the typed repository like this:

foo, err := typed.Fetch(context.TODO(), uuid.UUID{...})

For comparison, using an untyped repository:

var foo Foo
err := repo.Fetch(context.TODO(), &foo)

func NewOf

func NewOf[Aggregate aggregate.TypedAggregate](r aggregate.Repository, makeFunc func(uuid.UUID) Aggregate) *TypedRepository[Aggregate]

NewOf is an alias for Typed.

func Typed

func Typed[Aggregate aggregate.TypedAggregate](r aggregate.Repository, makeFunc func(uuid.UUID) Aggregate) *TypedRepository[Aggregate]

Typed returns a TypedRepository for the given generic aggregate type. The provided makeFunc is used to initialize aggregates when querying from the event store.

type Foo struct { *aggregate.Base }
func NewFoo(id uuid.UUID) *Foo { return &Foo{Base: aggregate.New("foo", id)} }

var repo aggregate.Repository
typed := repository.Typed(repo, NewFoo)

func (*TypedRepository[Aggregate]) Delete

func (r *TypedRepository[Aggregate]) Delete(ctx context.Context, a Aggregate) error

Delete implements aggregate.TypedRepository.Delete.

func (*TypedRepository[Aggregate]) Fetch

func (r *TypedRepository[Aggregate]) Fetch(ctx context.Context, id uuid.UUID) (Aggregate, error)

Fetch implements aggregate.TypedRepository.Fetch.

func (*TypedRepository[Aggregate]) FetchVersion

func (r *TypedRepository[Aggregate]) FetchVersion(ctx context.Context, id uuid.UUID, version int) (Aggregate, error)

FetchVersion implements aggregate.TypedRepository.FetchVersion.

func (*TypedRepository[Aggregate]) NewFunc

func (r *TypedRepository[Aggregate]) NewFunc() func(uuid.UUID) Aggregate

NewFunc returns the underlying constructor function for the aggregate of this repository.

func (*TypedRepository[Aggregate]) Query

func (r *TypedRepository[Aggregate]) Query(ctx context.Context, q aggregate.Query) (<-chan Aggregate, <-chan error, error)

Query implements aggregate.TypedRepository.Query.

func (*TypedRepository[Aggregate]) Repository

func (r *TypedRepository[Aggregate]) Repository() aggregate.Repository

Repository returns the underlying, untyped aggregate.Repository.

func (*TypedRepository[Aggregate]) Save

func (r *TypedRepository[Aggregate]) Save(ctx context.Context, a Aggregate) error

Save implements aggregate.TypedRepository.Save.

func (*TypedRepository[Aggregate]) Use

func (r *TypedRepository[Aggregate]) Use(ctx context.Context, id uuid.UUID, fn func(Aggregate) error) error

Use implements aggregate.TypedRepository.Use.