Event Sourced Aggregates

An aggregate is a domain object whose state is rebuilt by replaying its event history. Aggregates enforce business rules before new events are written to a stream — they are the write-side counterpart to projections.

The pattern

1. Load every event for the aggregate's stream.
2. Apply each event to rebuild the current state.
3. Invoke a command method on the aggregate.
4. The aggregate enforces invariants and emits zero or more new events.
5. The new events are appended to the stream and broadcast to projections.

Example

final class BasketAggregate extends Aggregate implements IsAggregate
{
    public function __construct(
        private AggregateId $id,
        private Price $totalAmount,
    ) {
        parent::__construct();
    }

    public function addItemToBasket(ProductId $productId, ItemName $name, Price $amount): void
    {
        if ($this->totalAmount->exceeds(Price::limit())) {
            throw new BasketLimitReached();
        }

        $this->addNewEvent(new ItemAddedToBasket($productId, $name, $amount));
    }

    protected function applyItemAddedToBasket(ItemAddedToBasket $event): void
    {
        $this->totalAmount = $this->totalAmount->plus($event->amount);
    }
}

Loading an aggregate

$events  = $client->readStream(StreamId::fromString('basket-42'));
$basket  = BasketAggregate::reconstitute($events);

$basket->addItemToBasket($productId, $name, $price);

$client->append($basket->pullNewEvents());

Combine aggregates (write model) with projections (read model) to get a clean CQRS architecture. See CQRS Scaffolding for tooling that generates the boilerplate for you.