lookup

README

Lookups

The lookup package provides an event-driven lookup table for your aggregates.

Example: User Signup

Lookup tables are often needed to find a specific aggregate by a field other than its UUID. An example for this is a user signup where an email address must be unique across all users.

Events

package auth

const UserRegistered = "auth.user.registered"
const EmailLookup = "email"

type UserRegisteredData struct {
	Email string
}

func (data UserRegisteredData) ProvideLookup(p lookup.Provider) {
	p.Provide(EmailLookup, data.Email)
}

User

package auth

const UserAggregate = "auth.user"

type User struct {
	*aggregate.Base

	Name string
	Email string
}

func NewUser(id uuid.UUID) *User {
	u := &User{Base: aggregate.New(UserAggregate, id)}

	aggregate.ApplyWith(u, UserRegistered, u.register)

	return u
}

func (u *User) Register(
	ctx context.Context,
	emails *lookup.Lookup,
	email string,
) error {
	if lookup.Contains[string](
		ctx, emails, UserAggregate, EmailLookup, u.AggregateID(),
	) {
		return errors.New("email address is already in use")
	}

	aggregate.Next(u, UserRegistered, email)
}

func (u *User) register(e event.Of[string]) {
	u.Email = e.Data()
}

Run

package example

func registerUser(email string) {
	var store event.Store
	var bus event.Bus

	l := lookup.New(store, bus, []string{UserRegistered})
	
	errs, err := l.Run(context.TODO())
	if err != nil {
		panic(err)
	}

	<-l.Ready()

	u := NewUser(uuid.New())
	if err := u.Register(ctx, emails, email); err != nil {
		panic(err)
	}
}

Documentation

Index

• Variables
• func Contains[Value any](ctx context.Context, l lookup, aggregateName, key string, ...) bool
• func Expect[Value any](ctx context.Context, l lookup, aggregateName, key string, ...) (Value, error)
• type Data
• type Lookup
  • func New(store event.Store, bus event.Bus, events []string, opts ...Option) *Lookup
  • func (l *Lookup) ApplyEvent(evt event.Event)
  • func (l *Lookup) ApplyJob(ctx projection.Job) error
  • func (l *Lookup) Lookup(ctx context.Context, aggregateName, key string, aggregateID uuid.UUID) (any, bool)
  • func (l *Lookup) Map() map[string]map[uuid.UUID]map[any]any
  • func (l *Lookup) Provider(aggregateName string, aggregateID uuid.UUID) Provider
  • func (l *Lookup) Ready() <-chan struct{}
  • func (l *Lookup) Reverse(ctx context.Context, aggregateName, key string, value any) (uuid.UUID, bool)
  • func (l *Lookup) Run(ctx context.Context) (<-chan error, error)
  • func (l *Lookup) Schedule() *schedule.Continuous
• type Option
  • func ApplyEventsWith(fn func(evt event.Event, original func(event.Event))) Option
  • func ScheduleOptions(opts ...schedule.ContinuousOption) Option
• type Provider

Variables

var (
	// ErrNotFound is returned by Expect when the value for the given key cannot be found.
	ErrNotFound = errors.New("value for key not found")

	// ErrWrongType is returned by Expect when the value for the given key is not of the expected type.
	ErrWrongType = errors.New("value is not of the expected type")
)

Functions

func Contains

func Contains[Value any](ctx context.Context, l lookup, aggregateName, key string, aggregateID uuid.UUID) bool

Contains returns whether the lookup table contains the given key for the given aggregate.

func Expect

func Expect[Value any](ctx context.Context, l lookup, aggregateName, key string, aggregateID uuid.UUID) (Value, error)

Expect calls l.Lookup with the given arguments and casts the result to the given generic type. If the lookup value cannot be found, an error that unwraps to ErrNotFound is returned. If the lookup value is not of the expected type, an error that unwraps to ErrWrongType is returned.

Types

type Data

type Data interface {
	// ProvideLookup accepts a Provider that can be used to update the lookup
	// table of a specific aggregate.
	ProvideLookup(Provider)
}

Data is the interface that must be implemented by events that want to populate the lookup table. The Provider that is passed to ProvideLookup allows the event to update the lookup table of a specific aggregate. The Provider that is provided to the ProvideLookup method is restricted to the aggregate that the event belongs to. The Provider is not thread-safe and must not be used outside of the ProvideLookup method. To retrieve a thread-safe Provider, use the l.Provider() method on the *Lookup type.

// UserRegisteredData is the event data for a registered user.
type UserRegisteredData struct {
	Email string
}

func (data UserRegisteredData) ProvideLookup(p lookup.Provider) {
	p.Provide("email", data.Email)
}

type Lookup

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

Lookup is a projection that provides a lookup table for a given set of events. The lookup table is populated by events that implment the Data interface. A *Lookup is thread-safe.

func New

func New(store event.Store, bus event.Bus, events []string, opts ...Option) *Lookup

New returns a new lookup table. The lookup table becomes ready after the first projection job has been applied. Use the l.Ready() method of the returned *Lookup to wait for the lookup table to become ready. Use l.Run() to start the projection of the lookup table.

func (*Lookup) ApplyEvent

func (l *Lookup) ApplyEvent(evt event.Event)

ApplyEvent implements projection.EventApplier.

func (*Lookup) ApplyJob

func (l *Lookup) ApplyJob(ctx projection.Job) error

ApplyJob applies the given projection job on the lookup table.

func (*Lookup) Lookup

func (l *Lookup) Lookup(ctx context.Context, aggregateName, key string, aggregateID uuid.UUID) (any, bool)

Lookup returns the lookup value for the given key of the given aggregate, or false if no value for the key exists.

func (*Lookup) Map

func (l *Lookup) Map() map[string]map[uuid.UUID]map[any]any

Map returns the all values in the lookup table as a nested map, structured as follows:

map[AGGREGATE_NAME]map[AGGREGATE_ID]map[LOOKUP_KEY]LOOKUP_VALUE

func (*Lookup) Provider

func (l *Lookup) Provider(aggregateName string, aggregateID uuid.UUID) Provider

Provider returns the lookup provider for the given aggregate. The returned Provider is thread-safe.

func (*Lookup) Ready

func (l *Lookup) Ready() <-chan struct{}

Ready returns a channel that is closed when the lookup table is ready. The lookup table becomes ready after the first projection job has been applied. Call l.Run() to start the projection of the lookup table.

func (*Lookup) Reverse

func (l *Lookup) Reverse(ctx context.Context, aggregateName, key string, value any) (uuid.UUID, bool)

Reverse returns the aggregate id that has the given value as the lookup value for the given lookup key. If value is not comparable or if the value does not exists for the given aggregate, uuid.Nil and false are returned. Otherwise the aggregate id and true are returned.

func (*Lookup) Run

func (l *Lookup) Run(ctx context.Context) (<-chan error, error)

Run runs the projection of the lookup table until ctx is canceled. Any asynchronous errors are sent into the returned channel.

func (*Lookup) Schedule

func (l *Lookup) Schedule() *schedule.Continuous

Schedule returns the projection schedule for the lookup table.

type Option

type Option func(*Lookup)

Option is a type that represents an option for configuring a *Lookup. Options are used as arguments in the constructor function New.

func ApplyEventsWith

func ApplyEventsWith(fn func(evt event.Event, original func(event.Event))) Option

ApplyEventsWith returns an Option that overrides the default function that applies events to the lookup table. When an event is applied, the provided function is called with the event as its first argument and the original event applier function as its second argument.

func ScheduleOptions

func ScheduleOptions(opts ...schedule.ContinuousOption) Option

ScheduleOptions returns an Option that configures the continuous schedule that is created by the lookup.

type Provider

type Provider interface {
	// Provide provides the lookup value for the given key.
	Provide(key string, value any)

	// Remove removes the lookup values for the given keys.
	Remove(keys ...string)
}

Provider allows events to update the lookup table of a specific aggregate.