Handling Projections from multiple Aggregates

[klim-dsoft]

Hi, @bounoable!

How can we create a projection from multiple aggregates?

Let's imagine, we have a User and an Order aggregates (with some events):

type UserAggregate struct {
	id string
	name string
	// ...
}

type OrderAggregate struct {
	id string
	userID string
	// ...
}

The goal is to create a User projection:

type UserProjection struct {
	id string
	name string
	orders []Order
	// ...
}

type Order struct {
	id string
	// ...
}

Also, we need to store this projection in the database. So here is a snippet:

type Projector struct {
	repo Repository
}

func (x Projector) Project(ctx context.Context, bus event.Bus, store event.Store) (<-chan error, error) {
	s := schedule.Continuously(bus, store, []string{
		user_created_event.Name,
		user_renamed_event.Name,
		order_created_event.Name,
	}, schedule.Debounce(time.Second))

	return s.Subscribe(ctx, func(job projection.Job) error {
		refs, errs, err := job.Aggregates(ctx, user_aggregate.Name)
		if err != nil {
			return fmt.Errorf("failed to extract user aggregates: %w", err)
		}

		return streams.Walk(ctx, func(ref aggregate.Ref) error {
			var proj *UserProjection
			if proj, err = x.repo.Fetch(ctx, ref.ID); err != nil {
				switch {
				case errors.Is(err, model.ErrNotFound):
					proj = NewUserProjection(ref.ID)
				default:
					return fmt.Errorf("failed to fetch user projection: %w", err)
				}
			}

			if err = job.Apply(ctx, proj); err != nil {
				return fmt.Errorf("failed to apply events to user projection: %w", err)
			}

			if err = x.repo.Save(ctx, proj); err != nil {
				return fmt.Errorf("failed to save user projection: %w", err)
			}

			return nil
		}, refs, errs)
	}, projection.Startup())
}

There are problems with the current solution:

1. Order events won't trigger projection updates;
2. There is a gap between fetching projection and saving it;

Could you please suggest the right approach using goes? We don't want to reinvent the wheel. Thank you!

[klim-dsoft]

@bounoable I have solved the first problem this way. The idea is to get all user aggregate IDs and apply events to them. IMHO, there should be a better way:

return s.Subscribe(ctx, func(job projection.Job) error {
	ch, errs, err := job.Events(ctx)
	if err != nil {
		return fmt.Errorf("failed to extract events: %w", err)
	}

	aggregateIDs := make(map[string]struct{})

	if err = streams.Walk(ctx, func(e event.Event) error {
		switch v := e.Data().(type) {
		case user_created_event.Event:
			aggregateID := pick.AggregateID(e)
			aggregateIDs[aggregateID] = struct{}{}
		case order_created_event.Event:
			aggregateID := v.GetUserID()
			aggregateIDs[aggregateID] = struct{}{}
		}

		return nil
	}, ch, errs); err != nil {
		return fmt.Errorf("failed to walk: %w", err)
	}

	for aggregateID := range aggregateIDs {
		var proj *UserProjection
		if proj, err = x.repo.Fetch(ctx, aggregateID); err != nil {
			switch {
			case errors.Is(err, model.ErrNotFound):
				proj = New UserProjection(aggregateID)
			default:
				return fmt.Errorf("failed to fetch user projection: %w", err)
			}
		}

		if err = job.Apply(ctx, proj); err != nil {
			return fmt.Errorf("failed to apply events to user projection: %w", err)
		}

		if err = x.repo.Save(ctx, proj); err != nil {
			return fmt.Errorf("failed to save user projection: %w", err)
		}

		return nil
	}

	return nil
}, projection.Startup())

[bounoable]

Hi @klim-dsoft, for multi-aggregate projections I would typically use one of the approaches below (there's also an example in here how I mostly do single-aggregate projections). I also use a MongoDB collection for the UserOrder read-model to show how I would query orders for a given user.

Also, I always extract the state of my aggregates and projections into DTOs to make them simple to send over the network but that's not really necessary.

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/google/uuid"
	"github.com/modernice/goes/aggregate"
	"github.com/modernice/goes/backend/mongo"
	"github.com/modernice/goes/event"
	"github.com/modernice/goes/helper/pick"
	"github.com/modernice/goes/helper/streams"
	"github.com/modernice/goes/persistence/model"
	"github.com/modernice/goes/projection"
	"github.com/modernice/goes/projection/schedule"
	"go.mongodb.org/mongo-driver/bson"
	gomongo "go.mongodb.org/mongo-driver/mongo"
	"go.mongodb.org/mongo-driver/mongo/options"
)

type UserRepository = aggregate.TypedRepository[*User]
type OrderRepository = aggregate.TypedRepository[*Order]
type OrderHistoryRepository = model.Repository[*OrderHistory, uuid.UUID]

type UserOrderRepository struct {
	model.Repository[*UserOrder, uuid.UUID]

	col    *gomongo.Collection
	orders OrderRepository
}

func NewUserOrderRepository(ctx context.Context, col *gomongo.Collection, orders OrderRepository) *UserOrderRepository {
	repo := mongo.NewModelRepository[*UserOrder](
		col,
		mongo.ModelFactory(UserOrderOf, true),
		mongo.ModelIDKey("id"),
	)

	if err := repo.CreateIndexes(ctx); err != nil {
		panic(err)
	}

	// To be able to query orders by user id.
	if _, err := col.Indexes().CreateOne(ctx, gomongo.IndexModel{
		Keys:    gomongo.IndexModel{Keys: bson.M{"userId": 1}},
	}); err != nil {
		panic(err)
	}

	return &UserOrderRepository{
		Repository: repo,
		col:        col,
		orders:     orders,
	}
}

func (r *UserOrderRepository) ByUser(ctx context.Context, userID uuid.UUID) ([]*UserOrder, error) {
	cursor, err := r.col.Find(ctx, bson.M{"userId": userID})
	if err != nil {
		return nil, fmt.Errorf("find orders: %w", err)
	}

	orders := make([]*UserOrder, 0)
	if err := cursor.All(ctx, &orders); err != nil {
		return nil, fmt.Errorf("decode orders: %w", err)
	}

	return orders, nil
}

func (r *UserOrderRepository) Run_ReplaceEntireDTO(ctx context.Context, bus event.Bus, store event.Store) (<-chan error, error) {
	s := schedule.Continuously(bus, store, OrderEvents[:], schedule.Debounce(time.Second))

	return s.Subscribe(ctx, func(ctx projection.Job) error {
		refs, errs, err := ctx.Aggregates(ctx)
		if err != nil {
			return fmt.Errorf("extract aggregates: %w", err)
		}

		return streams.Walk(ctx, func(ref aggregate.Ref) error {
			return r.Use(ctx, ref.ID, func(uo *UserOrder) error {
				// I always fetch the entire aggregate from the event store to replace the whole DTO.
				o, err := r.orders.Fetch(ctx, ref.ID)
				if err != nil {
					return fmt.Errorf("fetch order: %w", err)
				}

				uo.UserOrderDTO.OrderDTO = o.OrderDTO

				// ... Here you can add additional data to the user order read-model if you need to ...

				return nil
			})
		}, refs, errs)
	})
}

func (r *UserOrderRepository) Run_ApplyEventsInvidually(ctx context.Context, bus event.Bus, store event.Store) (<-chan error, error) {
	s := schedule.Continuously(bus, store, OrderEvents[:], schedule.Debounce(time.Second))

	return s.Subscribe(ctx, func(ctx projection.Job) error {
		events, errs, err := ctx.Events(ctx)
		if err != nil {
			return fmt.Errorf("extract events: %w", err)
		}

		return streams.Walk(ctx, func(evt event.Event) error {
			return r.Use(ctx, pick.AggregateID(evt), func(uo *UserOrder) error {

				// And here we would need to basically reimplement the whole Order aggregate's event application logic for ALL existing order events.
				// This is simple in this example because there's only 1 event, but in a real-world application this becomes a nightmare.
				// This is the reason why I don't use this approach. I always prefer to just fetch the entire aggregate from the event store and replace the whole DTO.
				switch evt.Name() {
				case OrderCreated:
					uo.UserOrderDTO.OrderDTO = evt.Data().(OrderDTO)
				}

				return nil
			})
		}, events, errs)
	})
}

// I'm using MongoDB for such read-models because it works well with the "bson" tags in the OrderHistoryDTO.
func NewOrderHistoryRepository(ctx context.Context, col *gomongo.Collection) OrderHistoryRepository {
	repo := mongo.NewModelRepository[*OrderHistory](
		col,
		// Set "createIfNotFound" parameter to `true` to create the model if it doesn't exist instead of returning model.ErrNotFound
		mongo.ModelFactory(OrderHistoryOf, true),
		mongo.ModelIDKey("userId"),
	)

	if err := repo.CreateIndexes(ctx); err != nil {
		panic(err)
	}

	return repo
}

const (
	UserAggregate = "user"
	UserCreated   = "user.created"
	UserRenamed   = "user.renamed"
)

type UserCreatedEvent = event.Of[string]
type UserRenamedEvent = event.Of[string]

const (
	OrderAggregate = "order"
	OrderCreated   = "order.created"
)

var OrderEvents = [...]string{
	OrderCreated,
}

type OrderCreatedEvent = event.Of[OrderDTO]

type UserDTO struct {
	ID    uuid.UUID // I like to add aggregate ids to DTOs
	Name  string
	Email string
}

type User struct {
	*aggregate.Base
	UserDTO
}

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

	event.ApplyWith(u, u.create, UserCreated)
	event.ApplyWith(u, u.rename, UserRenamed)

	return u
}

func (u *User) Create(email string) error {
	// ... validation stuff ...
	aggregate.Next(u, UserCreated, email)
	return nil
}

func (u *User) create(evt UserCreatedEvent) {
	u.UserDTO.Email = evt.Data()
}

func (u *User) Rename(name string) error {
	// ... validation stuff ...
	aggregate.Next(u, UserRenamed, name)
	return nil
}

func (u *User) rename(evt UserRenamedEvent) {
	u.UserDTO.Name = evt.Data()
}

type OrderDTO struct {
	ID     uuid.UUID   `bson:"id"`
	UserID uuid.UUID   `bson:"userId"`
	Total  money.Money `bson:"total"`
	// ...
}

type Order struct {
	*aggregate.Base
	OrderDTO
}

func NewOrder(id uuid.UUID) *Order {
	o := &Order{
		Base:     aggregate.New(OrderAggregate, id),
		OrderDTO: OrderDTO{ID: id},
	}

	event.ApplyWith(o, o.create, OrderCreated)

	return o
}

func (o *Order) Create(userID uuid.UUID, total money.Money) error {
	// ... validation stuff ...
	aggregate.Next(o, OrderCreated, OrderDTO{
		ID:     o.AggregateID(),
		UserID: userID,
		Total:  total,
	})
	return nil
}

func (o *Order) create(evt OrderCreatedEvent) {
	o.OrderDTO = evt.Data()
}

type UserOrderDTO struct {
	OrderDTO `bson:",inline"`
	// Additional fields for the user order read-model
	// If you don't need additional fields, you could also just use embed OrderDTO directly into UserOrder
}

type UserOrder struct {
	*projection.Base       `bson:"-"`
	*projection.Progressor `bson:"progressor"` // Progressor makes only really sense to use when using the "ApplyEventsIndividually" approach.
	UserOrderDTO           `bson:",inline"`
}

func UserOrderOf(orderID uuid.UUID) *UserOrder {
	return &UserOrder{
		Base:       projection.New(),
		Progressor: projection.NewProgressor(),
		UserOrderDTO: UserOrderDTO{
			OrderDTO: OrderDTO{ID: orderID},
		},
	}
}

func (uo *UserOrder) ModelID() uuid.UUID {
	return uo.OrderDTO.ID
}

type OrderHistoryDTO struct {
	UserID   uuid.UUID
	UserName string
	Orders   []OrderDTO
}

type OrderHistory struct {
	*projection.Base       `bson:"-"`
	*projection.Progressor // Progressor makes only really sense to use when using the "ApplyEventsIndividually" approach.
	OrderHistoryDTO
}

func (h *OrderHistory) ModelID() uuid.UUID {
	return h.UserID
}

func OrderHistoryOf(userID uuid.UUID) *OrderHistory {
	return &OrderHistory{
		Base:       projection.New(),
		Progressor: projection.NewProgressor(),
		OrderHistoryDTO: OrderHistoryDTO{UserID: userID},
	}
}

var OrderHistoryEvents = [...]string{
	UserRenamed,
	OrderCreated,
}

type OrderHistoryProjector struct {
	users      UserRepository
	orders     OrderRepository
	userOrders *UserOrderRepository
	histories  OrderHistoryRepository
}

func NewOrderHistoryProjector(
	users UserRepository,
	orders OrderRepository,
	userOrders *UserOrderRepository,
	histories OrderHistoryRepository,
) *OrderHistoryProjector {
	return &OrderHistoryProjector{
		users:      users,
		orders:     orders,
		userOrders: userOrders,
		histories:  histories,
	}
}

// This is my preferred approach for most multi-aggregate projections.
func (p *OrderHistoryProjector) Run_RebuildEntireProjection(ctx context.Context, bus event.Bus, store event.Store) (<-chan error, error) {
	s := schedule.Continuously(bus, store, OrderHistoryEvents[:], schedule.Debounce(time.Second))

	return s.Subscribe(ctx, func(ctx projection.Job) error {
		refs, errs, err := ctx.Aggregates(ctx)

		if err != nil {
			return fmt.Errorf("extract aggregates: %w", err)
		}

		return streams.Walk(ctx, func(ref aggregate.Ref) error {
			switch ref.Name {
			case UserAggregate:
				return p.rebuild(ctx, ref.ID)
			case OrderAggregate:
				userID, err := p.getUserFromOrder(ctx, ref.ID)
				if err != nil {
					return fmt.Errorf("get user from order: %w", err)
				}
				return p.rebuild(ctx, userID)
			default:
				return fmt.Errorf("unknown aggregate: %s (%s)", ref.Name, ref.ID)
			}
		}, refs, errs)
	}, projection.Startup())
}

func (p *OrderHistoryProjector) getUserFromOrder(ctx context.Context, orderID uuid.UUID) (uuid.UUID, error) {
	o, err := p.orders.Fetch(ctx, orderID)
	if err != nil {
		return uuid.Nil, fmt.Errorf("fetch order: %w", err)
	}
	return o.UserID, nil
}

func (p *OrderHistoryProjector) rebuild(ctx context.Context, userID uuid.UUID) error {
	return p.histories.Use(ctx, userID, func(h *OrderHistory) error {
		if err := p.buildHistory(ctx, h); err != nil {
			return err
		}
		return h.finalize(ctx, p)
	})
}

// Oftem times, I create a separate "finalize" method to do some final processing on the projection target.
func (h *OrderHistory) finalize(ctx context.Context, p *OrderHistoryProjector) error {
	// ... some final processing ...
	// ... heavy work ...
	// ... sort slices etc. ...
	// ... fetch some more data and add it to the projection ...

	// In this case, we always want to update the user name after updating a projection.
	u, err := p.users.Fetch(ctx, h.UserID)
	if err != nil {
		return fmt.Errorf("fetch user: %w", err)
	}
	h.UserName = u.UserDTO.Name

	return nil
}

func (p *OrderHistoryProjector) buildHistoryOf(ctx context.Context, userID uuid.UUID) error {
	return p.histories.Use(ctx, userID, func(h *OrderHistory) error {
		return p.buildHistory(ctx, h)
	})
}

func (p *OrderHistoryProjector) buildHistory(ctx context.Context, h *OrderHistory) error {
	orders, err := p.userOrders.ByUser(ctx, h.UserID)
	if err != nil {
		return fmt.Errorf("fetch orders: %w", err)
	}

	h.Orders = slices.Map(orders, func(o *UserOrder) OrderDTO {
		return o.OrderDTO
	})

	return nil
}

// This is an alternative approach to the same projection but applying events
// directly to the projection targets by implementing the `ApplyEvent` method.
// This is in theory more efficient because we can just append the order to
// the history directly without having to fetch all orders for the user each
// time the projection is run. In my experience, this approach is not worth the
// effort because the performance difference is negligible (for most cases)
// and the code gets more and more complex as the number of different events
// to check increases (You have to check for each event, instead of just for each aggregate)
func (p *OrderHistoryProjector) Run_ApplyEventsIndividually(ctx context.Context, bus event.Bus, store event.Store) (<-chan error, error) {
	s := schedule.Continuously(bus, store, OrderHistoryEvents[:], schedule.Debounce(time.Second))

	return s.Subscribe(ctx, func(ctx projection.Job) error {
		events, errs, err := ctx.Events(ctx)
		if err != nil {
			return fmt.Errorf("extract events: %w", err)
		}

		return streams.Walk(ctx, func(evt event.Event) error {
			switch evt.Name() {
			case UserRenamed:
				return p.applyJobToHistoryOf(ctx, pick.AggregateID(evt))
			case OrderCreated:
				order := evt.Data().(OrderDTO)
				return p.applyJobToHistoryOf(ctx, order.UserID)
			default:
				return fmt.Errorf("unknown event: %s", evt.Name())
			}
		}, events, errs)
	}, projection.Startup())
}

func (p *OrderHistoryProjector) applyJobToHistoryOf(ctx projection.Job, userID uuid.UUID) error {
	return p.histories.Use(ctx, userID, func(h *OrderHistory) error {
		return ctx.Apply(ctx, h)
	})
}

// ApplyEvent is only needed if you want to use the "ApplyEventsIndividually"
// approach such that you can pass the OrderHistory to the projection job's Apply() method.
func (p *OrderHistory) ApplyEvent(evt event.Event) {
	switch evt.Name() {
	case UserRenamed:
		p.UserName = evt.Data().(string)
	case OrderCreated:
		p.Orders = append(p.Orders, evt.Data().(OrderDTO))
	}
}

func (p *OrderHistoryProjector) Run_MixedApproach(ctx context.Context, bus event.Bus, store event.Store) (<-chan error, error) {
	s := schedule.Continuously(bus, store, OrderHistoryEvents[:], schedule.Debounce(time.Second))

	return s.Subscribe(ctx, func(ctx projection.Job) error {
		events, errs, err := ctx.Events(ctx)
		if err != nil {
			return fmt.Errorf("extract events: %w", err)
		}

		if err := streams.Walk(ctx, func(evt event.Event) error {
			switch evt.Name() {
			// Apply UserRenamed events directly to the order histories.
			case UserRenamed:
				return p.applyJobToHistoryOf(ctx, pick.AggregateID(evt))
			default:
				return nil
			}
		}, events, errs); err != nil {
			return err
		}

		// By default, the Aggregates() method and the Events() method of the projection job run the same query.
		// Because the projection job caches its query results, calling Aggregates() here will _not_ run the query again,
		// so no need to worry about performance here.
		refs, errs, err := ctx.Aggregates(ctx)
		if err != nil {
			return fmt.Errorf("extract aggregates: %w", err)
		}

		// But for Order events, always rebuild the entire order history to make sure all the state is up to date.
		return streams.Walk(ctx, func(ref aggregate.Ref) error {
			switch ref.Name {
			case OrderAggregate:
				return p.buildHistoryOf(ctx, ref.ID)
			default:
				return nil
			}
		}, refs, errs)
	}, projection.Startup())
}