Part 4: Aggregates

TIP

Further information about Aggregates can be found in the documentation. The starting point is: Documentation

What is an Aggregate?

Aggregates are the core building blocks of the domain model. They represent a collection of domain objects that are treated as a single unit. Aggregates are responsible for enforcing business rules and maintaining the consistency of the domain model. They encapsulate the state of the domain objects and provide methods for interacting with them. Aggregates are the primary entities that are persisted in the event store and are used to reconstruct the state of the domain model.

From a user's perspective, an aggregate is a collection of model (or entities) that are treated as a single unit. The aggregate acts as the interface between the user and its underlying entities or the aggregate itself — essentially serving as the primary entity, often referred to as the "root aggregate".

Unlike traditional models, an aggregate is initially loaded through its events. This means the aggregate starts empty and is reconstructed by applying its events sequentially. These events represent the history of changes that have occurred within the aggregate.

The AggregateRepository is responsible for loading and saving aggregates and is commonly used in commands. When loading an aggregate, the AggregateRepository fetches all events associated with the aggregate from the EventStore in the correct order and applies them one by one to a new NewAggregate instance.

When defining aggregates, we don't need to worry about the complexities of loading or managing them — this is handled by the comby framework. All we need to do is implement the required fields and methods and define the domain events that the aggregate should produce. The framework takes care of the rest.

Aggregate

Like all key components in comby, an aggregate is built on a base class called BaseAggregate. This class provides a robust foundation for implementing aggregates, including essential functions and structures for event processing and domain event handlers. Let’s start with the definition of the aggregate:

// simple/domain/order/aggregate/aggregate.go
package aggregate

import (
	"github.com/gradientzero/comby/v2"
)

// assume simple lifecycle of an Order
const ORDER_STATUS_PLACED = "placed"
const ORDER_STATUS_CANCELLED = "cancelled"
const ORDER_STATUS_PAID = "paid"
const ORDER_STATUS_SHIPPED = "shipped"
const ORDER_STATUS_DELIVERED = "delivered"

type Item struct {
	ItemUid  string
	Sku      string
	Quantity int64
}

type Order struct {
	*comby.BaseAggregate

	// References
	PaymentUuid string

	// Entities
	Item *Item

	// Value Objects
	Status          string
	ShippingAddress string
	CancelReason    string

	// For simple update demonstration
	Comment       string
	AnyOtherField string
}

// NewAggregate creates new aggregate object
func NewAggregate() *Order {
	agg := &Order{}
	agg.BaseAggregate = comby.NewBaseAggregate()
	agg.Domain = "Order"

	// register domain event handlers
	comby.AddDomainEventHandler(agg, agg.OrderPlacedEvent)
	comby.AddDomainEventHandler(agg, agg.OrderCancelledEvent)
	comby.AddDomainEventHandler(agg, agg.OrderPaidEvent)
	comby.AddDomainEventHandler(agg, agg.OrderShippedEvent)
	comby.AddDomainEventHandler(agg, agg.OrderDeliveredEvent)
	comby.AddDomainEventHandler(agg, agg.OrderDeletedEvent)
	comby.AddDomainEventHandler(agg, agg.OrderUpdatedEvent)
	return agg
}

An aggregate typically has two critical components: the NewAggregate function and the aggregate itself (e.g., Order), which embeds the BaseAggregate class. Everything else is optional and can be extended based on specific requirements.

In practice, it is effective to structure an aggregate into three sections:

• Fields for references to other aggregates
• Fields for internal sub-models (entities)
• Fields for aggregate attributes (value objects)

For instance, our Order aggregate includes a reference to a Payment once the order is paid. To achieve this, we define a PaymentUuid field. In the context of the Order aggregate, we are not concerned with the details of the payment itself —o nly with the reference. This is because the payment lies outside the Order aggregate's boundary (also known as a bounded context). However, if we want to display details about the payment inside our domain (named "Order"), we can use this reference to retrieve the information and present it in a Readmodel or as part of a query.

Additionally, we have defined a simple entity Item which is treated as part of the Order aggregate and not as a standalone aggregate. This entity is used within the Order aggregate to store the order's item. An Item cannot exist independently of an Order (in our example).

In practice, it has proven useful to assign entities a unique identifier (Uid), allowing them to be modified through specific methods (intentions) if needed. This EntityUid, combined with the OrderUuid, can also be used for precise referencing by other aggregates. (Note: References must always include the UUID of an aggregate and may optionally include the Uid of its entities.) As a convention, entities use the term Uid instead of Uuid. Technically, both are UUIDv4 strings, but this distinction ensures clarity when dealing with aggregates and their entities.

Finally, we define our value objects, which are simple fields that contain information about the order, such as its status, comments, and other related details.

The NewAggregate method serves as the entry point for an aggregate and is utilized by comby, commands, and repositories. This method initializes the BaseAggregate, sets the domain name ("Order"), and registers the domain event handlers. For the latter, comby provides a helper function that simplifies the process and reduces manual effort.

Domain Event Handling

After setting up our aggregate, we move on to the individual intentions, their domain events, and the domain event handling. Aggregates are the only components capable of producing domain events, and they are also loaded from these domain events. This means a domain event typically consists of three components:

• Domain Event as a structure.
• Intention as a method (business logic and creation of new domain events).
• Domain Event Handler as a method (applying a domain event to load or update the aggregate state).

Let’s start with a domain event:

Place New Order

Here, we define our domain event, OrderPlacedEvent. Since a domain event is always embedded within an event, the corresponding aggregate UUID can always be derived from the surrounding event. Therefore, it is not necessary to store this information directly within the domain event itself. The rest of the domain event, such as Item in this example, is user-defined and contains the order item we want to process.

// simple/domain/order/aggregate/place.go
package aggregate

import (
	"context"
	"fmt"

	"github.com/gradientzero/comby/v2"
)

type OrderPlacedEvent struct {
	Item *Item `json:"items"`
}

// intention on current aggregate
func (agg *Order) Place(item *Item) error {

	// business logic
	if item.Quantity < 1 {
		return fmt.Errorf("quantity of item must be greater than 0")
	}

	// create new event based on the current aggregate
	domainEvt := &OrderPlacedEvent{
		Item: item,
	}

	// apply event on the current aggregate
	return comby.ApplyDomainEvent(agg, domainEvt, true)
}

// event handler for domain event
func (agg *Order) OrderPlacedEvent(ctx context.Context, evt comby.Event, domainEvt *OrderPlacedEvent) error {
	agg.Item = domainEvt.Item
	agg.Status = ORDER_STATUS_PLACED
	return nil
}

Next, we define two receiver methods.

The first method, Place, represents our intention to create a new order. The method name should clearly indicate that it aims to modify the aggregate. Within this method, we enforce our business logic by validating the number of items being ordered. If the business logic passes, we create a new domain event and apply it to the aggregate using ApplyDomainEvent.

This generates a finalized Event containing the new domain event and stores it within the aggregate. These events are temporarily tracked as uncommitted changes and can later be retrieved by commands using GetUncommittedEvents. This ensures that any changes made to the aggregate are only persisted (and should only be persisted) through a command.

Side note: If we use comby.ApplyDomainEvent(agg, domainEvt, false), the domain event would simply be applied to the aggregate without being added to the uncommitted changes. This is particularly useful for scenarios like read models or projections. This brings us to the next method:

The second method, OrderPlacedEvent, is our domain event handler responsible for processing the domain event. The method name can be freely chosen; however, it should clearly convey that the method is solely applying an already existing domain event and not generating new ones. This method is invoked whenever the domain event OrderPlacedEvent is loaded and needs to be applied to the aggregate.

The domain event is applied to the aggregate to update its current state. In our case, the method sets the Item information from the domain event and updates the status to placed. Subsequent domain events can make further changes to the aggregate. The order of events is crucial, which is why aggregates have versions. These versions ensure that events are applied in the correct sequence and are automatically managed by the BaseAggregate class and the comby framework.

Additional Domain Events

All other domain events and their corresponding domain event handlers follow a technically similar structure. They validate the provided parameters against the current state of the aggregate using business logic, generate new domain events as needed, apply these events to the current aggregate, and handle the domain event to update the aggregate's state accordingly.

Update Order

// simple/domain/order/aggregate/update.go
package aggregate

import (
	"context"

	"github.com/gradientzero/comby/v2"
)

type OrderUpdatedEvent struct {
	Comment       string `json:"comment,omitempty"`
	AnyOtherField string `json:"anyOtherField,omitempty"`
}

// intention on current aggregate
func (agg *Order) Update(comment, anyOtherField string) error {

	// business logic
	// ...

	// create new event based on the current aggregate
	domainEvt := &OrderUpdatedEvent{
		Comment:       comment,
		AnyOtherField: anyOtherField,
	}

	// apply event on the current aggregate
	return comby.ApplyDomainEvent(agg, domainEvt, true)
}

// event handler for domain event
func (agg *Order) OrderUpdatedEvent(ctx context.Context, evt comby.Event, domainEvt *OrderUpdatedEvent) error {
	agg.Comment = domainEvt.Comment
	agg.AnyOtherField = domainEvt.AnyOtherField
	return nil
}

Delete Order

// simple/domain/order/aggregate/delete.go
package aggregate

import (
	"context"
	"fmt"

	"github.com/gradientzero/comby/v2"
)

type OrderDeletedEvent struct {
	Reason string `json:"reason"`
}

func (agg *Order) Delete() error {
	// business logic
	if agg.Deleted {
		return fmt.Errorf("order already deleted")
	}

	// create new event based on the current aggregate
	domainEvt := &OrderDeletedEvent{
		Reason: "deleted",
	}

	// apply event on the current aggregate
	return comby.ApplyDomainEvent(agg, domainEvt, true)
}

func (agg *Order) OrderDeletedEvent(ctx context.Context, evt comby.Event, domainEvt *OrderDeletedEvent) error {
	agg.Deleted = true
	return nil
}

Mark Cancelled

// simple/domain/order/aggregate/mark.cancelled.go
package aggregate

import (
	"context"

	"github.com/gradientzero/comby/v2"
)

type OrderCancelledEvent struct {
	Reason string `json:"items"`
}

// intention on current aggregate
func (agg *Order) MarkCancelled(reason string) error {

	// business logic
	// ...

	// create new event based on the current aggregate
	domainEvt := &OrderCancelledEvent{
		Reason: reason,
	}

	// apply event on the current aggregate
	return comby.ApplyDomainEvent(agg, domainEvt, true)
}

// event handler for domain event
func (agg *Order) OrderCancelledEvent(ctx context.Context, evt comby.Event, domainEvt *OrderCancelledEvent) error {
	agg.Status = ORDER_STATUS_CANCELLED
	agg.CancelReason = domainEvt.Reason
	return nil
}

Mark Paid

// simple/domain/order/aggregate/mark.paid.go
package aggregate

import (
	"context"
	"fmt"

	"github.com/gradientzero/comby/v2"
)

type OrderPaidEvent struct {
	PaymentUuid string `json:"paymentUuid"`
}

// intention on current aggregate
func (agg *Order) MarkPaid(paymentUuid string) error {

	// business logic
	if len(paymentUuid) < 1 {
		return fmt.Errorf("payment uuid must be set")
	}
	if agg.Status != ORDER_STATUS_PLACED {
		return fmt.Errorf("order can only be paid in status: %s", ORDER_STATUS_PLACED)
	}

	// create new event based on the current aggregate
	domainEvt := &OrderPaidEvent{
		PaymentUuid: paymentUuid,
	}

	// apply event on the current aggregate
	return comby.ApplyDomainEvent(agg, domainEvt, true)
}

// event handler for domain event
func (agg *Order) OrderPaidEvent(ctx context.Context, evt comby.Event, domainEvt *OrderPaidEvent) error {
	agg.Status = ORDER_STATUS_PAID
	agg.PaymentUuid = domainEvt.PaymentUuid
	return nil
}

Mark Shipped

// simple/domain/order/aggregate/mark.shipped.go
package aggregate

import (
	"context"
	"fmt"

	"github.com/gradientzero/comby/v2"
)

type OrderShippedEvent struct {
	ShippingAddress string `json:"shippingAddress"`
}

// intentions on current aggregate
func (agg *Order) MarkShipped(address string) error {
	// business logic
	if agg.Status != ORDER_STATUS_PAID {
		return fmt.Errorf("order not paid yet")
	}
	if len(address) < 1 {
		return fmt.Errorf("no shipping address set")
	}

	// create new event based on the current aggregate
	domainEvt := &OrderShippedEvent{
		ShippingAddress: address,
	}

	// apply event on the current aggregate
	return comby.ApplyDomainEvent(agg, domainEvt, true)
}

// event handler for domain event
func (agg *Order) OrderShippedEvent(ctx context.Context, evt comby.Event, domainEvt *OrderShippedEvent) error {
	agg.Status = ORDER_STATUS_SHIPPED
	agg.ShippingAddress = domainEvt.ShippingAddress
	return nil
}

Mark Delivered

// simple/domain/order/aggregate/mark.delivered.go
package aggregate

import (
	"context"
	"fmt"

	"github.com/gradientzero/comby/v2"
)

type OrderDeliveredEvent struct {
	Reason string `json:"reason"`
}

// intentions on current aggregate
func (agg *Order) MarkDelivered() error {
	// business logic
	if agg.Status != ORDER_STATUS_SHIPPED {
		return fmt.Errorf("order not shipped yet")
	}

	// create new event based on the current aggregate
	domainEvt := &OrderDeliveredEvent{
		Reason: "",
	}

	// apply event on the current aggregate
	return comby.ApplyDomainEvent(agg, domainEvt, true)
}

// event handler for domain event
func (agg *Order) OrderDeliveredEvent(ctx context.Context, evt comby.Event, domainEvt *OrderDeliveredEvent) error {
	agg.Status = ORDER_STATUS_DELIVERED
	agg.CancelReason = domainEvt.Reason
	return nil
}

Register in Facade

Finally, we need to register our aggregate in the facade. This is done by calling the RegisterAggregate method. Notice that we need to provide the NewAggregate method as an argument. This method is used by the facade to create a new instance of the aggregate when loading it from the event store.

// simple/domain/order/register.go
func Register(ctx context.Context, fc *comby.Facade) error {
    // ...
    if err := comby.RegisterAggregate(fc, aggregate.NewAggregate); err != nil {
        return err
    }
    // ...
}