4. Aggregates

A quick way to understand aggregates that use event sourcing is to consider the following analogy.

Commands and events

Imagine a car parts factory that is being managed by a director. This director commands this factory by sending instructions from his office down to the production floor. These instructions could be:

• recalibrate all machines;
• make that guy work 16 hours tomorrow;
• ship order to customer A. Now!

Each command is carried over to the plant and executed by workers (as diligently as possible). As this happens, papers and memos are filled (since we need to inform the headquarters about what has happened). Each of these commands can have various outcomes, which depend on the current state of the plant.

For example, when we send command to recalibrate all machines, we can get response that all 10 machines were recalibrated. Another response could be 5 machines were recalibrated, 3 were found to be broken and 2 were found to be missing.

Same with the other commands. When we try to make a worker work 16 hours he might agree or (if he is a Russian), he could just say some rich and powerful words to the management.

When we are trying to ship all orders, one of the following sequence of events might happen:

• given that all order items are prepared, orders get shipped;
• given that some items are missing but can be replaced with equivalent parts, order gets shipped with some replaced parts and then also record about that replacement.
• given that order can’t be fixed by replacing certain items, an apology gets written.

Let’s bring in the analogies between an aggregate root and a factory:

Event contracts

Let’s implement aggregate for this factory without using any external frameworks. First we define an interface for events. The snippets are illustrative; supporting types are omitted.

public interface IEvent { }

We define contracts for our events (I’m using a VS T4+ANTLR Combo here to avoid writing message contract classes by hand or bloating this post).

Events would look like this:

MachineAdded: IEvent(
    Guid factory,
    int machineId,
    DateTime initialCalibrationDate,
    double initialAccuracy)

MachineRecalibrated: IEvent(
    Guid factory,
    int machineId,
    DateTime calibrationDate,
    double accuracy)

MachineReportedStolen: IEvent (
    Guid factory,
    int machineId,
    Guid policeReportNumber)

MachineReportedBroken: IEvent (
    Guid factory,
    int machineId,
    Guid incidentReportId)

OrderShipped: IEvent(
    Guid factory,
    int orderId,
    ShippingReport report)

OrderCantBeShipped: IEvent (
    Guid factory,
    int orderId,
    ShipDenialReport report)

WorkerWorkShiftExtended: IEvent(Guid factory, int workerId, int hours)
WorkerSentMessageToManagement: IEvent(Guid factory, int workerId, string message)
WorkerRefusedToExtendShift: IEvent(Guid factory, int workerId)

Aggregate behavior

Given these events, let’s start by implementing an aggregate called FactoryAggregate. In the following method we hold all behaviors required to recalibrate all factory machines. Here we are taking a dependency on IServiceGuy.

public void RecalibrateAllMachines(IServiceGuy guy)
{
    var id = _state.FactoryId;
    foreach (var machine in _state.Machines.ToArray())
    {
        var report = guy.RecalibrateMachine(machine);

        switch (report.State)
        {
            case CalibrationReport.Missing:
                Apply(new MachineReportedStolen(id, machine.Id, report.Id));
                break;
            case CalibrationReport.Broken:
                Apply(new MachineReportedBroken(id, machine.Id, report.Id));
                break;
            case CalibrationReport.Recalibrated:
                Apply(new MachineRecalibrated(id, machine.Id, report.Date, report.Accuracy));
                break;
            default:
                throw new UnsupportedReport(report);
        }
    }
}

Here is how another method can be implemented:

public void MakeWorkerPullLongShift(int workerId, int hours, IRussianCurseGenerator gen)
{
    var worker = _state.GetWorker(workerId);

    if (worker.AgreesToPullHours(hours))
    {
        Apply(new WorkerWorkShiftExtended(_state.FactoryId, workerId, hours));
    }
    else
    {
        Apply(new WorkerRefusedToExtendShift(_state.FactoryId, workerId));
    }

    if (worker.IsEnraged && worker.IsRussian)
    {
        var curses = gen.GenerateCurses(worker.RageLevel);
        Apply(new WorkerSentMessageToManagement(_state.FactoryId, workerId, curses));
        return;
    }
}

Aggregate state

We have 2 unknown items here: _state and Apply.

_state reflects the current state of things on the factory. Apply is used to record events while also updating this state.

readonly FactoryState _state;
readonly Action<IEvent> _addToUnitOfWork;

public FactoryAggregate(Action<IEvent> addToUnitOfWork, FactoryState state)
{
    _addToUnitOfWork = addToUnitOfWork;
    _state = state;
}

void Apply(IEvent @event)
{
    // Pass the event to state (and let it update itself)
    _state.Apply(@event);
    // Add event to the unit of work
    _addToUnitOfWork(@event);
}

The aggregate state should know how to apply given events.

public sealed class FactoryState
{
    public Guid FactoryId { get; }
    public MachineList Machines { get; } = new MachineList();
    readonly MachineList _brokenMachines = new MachineList();

    public FactoryState(Guid factoryId)
    {
        FactoryId = factoryId;
    }

    public void Apply(IEvent @event)
    {
        RedirectToWhen.InvokeEvent(this, @event);
    }

    public void When(MachineAdded e)
    {
        Machines.AddNew(e.MachineId, e.InitialCalibrationDate, e.InitialAccuracy);
    }
    public void When(MachineReportedStolen e)
    {
        Machines.Remove(e.MachineId);
    }
    public void When(MachineReportedBroken e)
    {
        var m = Machines[e.MachineId];
        _brokenMachines.Add(m);
        Machines.Remove(m);
    }
    public void When(MachineRecalibrated e)
    {
        var m = Machines[e.MachineId];
        m.CalibrationDate = e.CalibrationDate;
        m.Accuracy = e.Accuracy;
    }

    // Other state and handlers omitted.
}

The RedirectToWhen helper is simple (see gist).

Commands are executed against the aggregate. If needed, aggregate uses additional services. While executing commands, aggregate employs complex behaviors which are based on the current state. This execution results in changes to the state, which are expressed as events.

Event sourcing

Let’s see how all these come together in a message handler that accepts command messages and passes them to the aggregate store.

public void Consume(RecalibrateAllMachines cmd)
{
    var guy = _container.Resolve<IServiceGuy>();
    _factoryRepository.Execute(cmd.FactoryId, c =>
    {
        c.RecalibrateAllMachines(guy);
    });
}

public void Consume(MakeWorkerPullLongShift cmd)
{
    var generator = _container.Resolve<IRussianCurseGenerator>();
    _factoryRepository.Execute(cmd.FactoryId, c =>
    {
        c.MakeWorkerPullLongShift(cmd.WorkerId, cmd.Hours, generator);
    });
}

Here’s how event sourcing works:

public void Execute(Guid id, Action<FactoryAggregate> when)
{
    var streamName = "factory-" + id;
    var givenEvents = _stream.ReadAllEvents(streamName);
    // load state from the event history
    // or, if you have snapshot, load it here first
    // we will not do the latter here
    var state = new FactoryState(id);
    foreach (var e in givenEvents)
    {
        state.Apply(e);
    }
    var thenEvents = new List<IEvent>();
    var ar = new FactoryAggregate(thenEvents.Add, state);
    // execute actual command
    when(ar);
    // do something with the events that were produced.
    // for example, append them to the history and then
    // publish in async or do both at once and face 2PC
    _stream.AppendEvents(streamName, thenEvents);
}

We are not relying on any frameworks. Versioning and version checks are not within the scope of this article, yet they could be added to the snippet above as needed.

Creating an aggregate

Event Stream for an aggregate is created by sending a command to the aggregate that does not exist. This command could look like:

new CreateUserCommand
{
    Id = Guid.NewGuid(),
    Name = registerModel.Name,
    Login = registerModel.Login,
    Password = registerModel.Password
};

In the handler we could have:

public void Consume(CreateUserCommand cmd)
{
    var service = _container.Resolve<EmailCheckService>();
    _userRepository.CreateAndExecute(cmd.Id, c =>
    {
        c.CreateNewUser(cmd.Id, cmd.Name, cmd.Login, cmd.Password, service);
    });
}

The only difference between Execute and CreateAndExecute methods is that the latter does not throw any exceptions if the stream does not exist before the call (and it might actually throw an exception if the stream did exist).

In the advanced cases of user registrations we might want to start by creating a RegistrationAggregate (to track all associated information and the actual process of registration), that would then pass control to various instances of CustomerAggregate, SecurityAccountAggregate and UserLoginAggregate. These details depend on your domain and are subject to discussion with domain experts.

Advanced scenarios

Above is a simplified approach to event sourcing that is generally described and explained in articles on the subject. There are ways to reduce development friction and increase expressiveness of the code, if Event centric approach a la Lokad is used.

The situation gets more interesting if:

• All messages (commands and events alike) carry unique identifier that is used at all steps of message processing to enforce message deduplication (required for repartitioning or cloud environment in general).
• Entity identities (i.e.: aggregate root identifiers) that are carried in the transport headers not only simplify our contracts (while still being exposed to the domain code in a decoupled way), but also provide simple foundation for message routing and aggregate repartitioning.
• We keep track of the message causality (event X was caused by command batch Y) in the transport headers along with client vectors. This provides us with the foundation to do partial message ordering (for the cases where cloud environments are stressed and tend to mess up order).

This topic is continued in the post on Tape Storage, which serves as persistence foundation for event sourcing and also enable fully portable systems to be developed and deployed with Lokad.CQRS.

Message builders

Sometimes our messages will be so large, that composing them via constructors would create fragile and non readable code. If such an instantiation happens too often for a single event, we can improve code readability in this specific case by introducing event builders.

Event builder is a stateful class, which provides helpful and expressive syntax to create an instance of an event.