0

I have workflow where I have complex command handlers encapsulated inside aggregate. These handlers emit some events, and then further logic based on result of these events can emit more events. Sample pseudocode of architecture:

public class ExampleAggregate : Aggregate
{
    public void Execute(CommandA c)
    {
        ChangeState(c.params); // applies events

        // this relies on synchronous application of events inside ChangeState,
        // real code checks more state, in multiple entities inside aggregate
        if (this.state == Y)
        {
            ChangeOtherState(c.otherParams); // applies events
        }
    }

    private State state;

    private void ChangeState(Params params) // can be called from multiple command handlers
    {
        if (params == ... && this.state == X)
        {
            Apply(new StateChanged { state = Y });
        }
        if (params == ... && this.state == Y)
        {
            Apply(new StateChanged { state = Z });
        }
        (...)
    }

    private void ChangeOtherState(...) => /* do some logic and apply some events*/;

    private void On(StateChanged e) => this.state = e.state;
}

(...)

// later, after command is processed, events are pulled and passed further
var aggregate = database.LoadAggregate(command.aggregateId);
aggregate.Execute(command); // synchronous
var events = aggregate.PullEvents();
// in-memory aggregate can be discarded now
database.Publish(events);


This example is simple, but I have more complex examples when applying events and logic is intertwined in loops, for example changing multiple entities inside aggregate and then logic (based on these multiple entities changed and multiple events applied) can emit more events.

I wonder if this is valid and correct approach? Also if there is any alternative or smart pattern which can be used?


EDIT:

"state" variable is only example - there are multiple fields which can be checked and drive execution of other methods.

Command handler is method, which can apply events or call "command" methods, which can also apply events.

Event handlers cannot call "command" methods, create commands, make any logic, etc. - these are only for changing state based on events. No state change can happen outside event handlers.

Single command handler execution is synchronous and rather fast, it's important that aggregate is most of the time cached in memory so command handlers execute very fast. But at the same time, if aggregate is unloaded, it can be easily restored from events (only events are persisted in database).

Maybe it's also important that aggregate contains multiple nested entities - command handlers can execute "command" methods on these entities, which can also apply events on aggregate.


EDIT 2: more real-world (but still made up) example, it shows that after every change to order, it has to update discounts on entities. In my code, there are more dependencies like this, where applying events influences further state changes.

public class Order : Aggregate
{
    public void Execute(ApplyOrderDiscountCount c)
    {
        /* validate c.DiscountCode */
        Apply(new OrderDiscountCodeApplied { NewDiscountCode = c.DiscountCode });
        UpdateDiscounts();
    }

    public void Execute(IncreaseOrderProductCount c)
    {
        var product = FindProduct(c.ProductId);

        if (product == null)
        {
            /* validate c.ProductId */
            Apply(new OrderProductAdded
            {
                ProductId = c.ProductId,
                Count = 1,
                ItemPrice = /* can query some service */,
                ...
            });
        }
        else
        {
            product.IncreaseCount();
        }
        
        UpdateDiscounts();
    }

    public void Execute(DecreaseOrderProductCount c)
    {
        var product = FindProduct(c.ProductId);

        if (product == null)
        {
            return; // product not found
        }

        product.DecreaseCount();
        
        UpdateDiscounts();
    }

    public void Execute(DecreaseAllOrderProductsCount c)
    {
        for (product in _products)
        {
            product.DecreaseCount();
        }
        
        //  discount is updated only once after all products are updated
        UpdateDiscounts();
    }

    public void On(OrderDiscountCodeApplied e) => _discountCode = e.NewDiscountCode;
    public void On(OrderProductAdded e) => _products.Add(new OrderProduct(this, e.Id, e.Count, e.ItemPrice));
    public void On(IncreaseOrderProductCount e) => FindProduct(e.ProductId).On(e);
    public void On(DecreaseOrderProductCount e) => FindProduct(e.ProductId).On(e);

    private List<OrderProduct> _products;
    private string _discountCode;

    private OrderProduct FindProduct(string productId)
        => _products.FindIndex(_ => _.Id == productId);

    private void UpdateDiscounts()
    {
        var discountByProductId = /* some logic calculating discount,
            e.g buy two items and get third for 50%, or second product from same category 
            gets 10% discount, discount code can be applied to whole order etc.
            This logic can be embedded in Order class, or it can be queried from external service.
        */;

        foreach (product in _products)
        {
            product.SetDiscount(discountByProductId[product.Id]);
        }
    }
}

public class OrderProduct : Entity<Order>
{
    public string Id { get; }
    public int Count { get; private set; }
    public decimal ItemPrice { get; }
    public decimal Discount { get; private set; }

    public OrderProduct(Order order, string id, int count, decimal itemPrice): base(order) { ... }

    public void IncreaseCount()
    {
        Apply(new OrderProductCountIncreased { ProductId = Id, NewCount = Count + 1 });
    }

    public void DecreaseCount()
    {
        if (Count <= 0)
        {
            return; // already empty
        }

        Apply(new OrderProductCountDecreased { ProductId = Id, NewCount = Count - 1 });
    }

    public void SetDiscount(decimal discount)
    {
        if (Discount == discount)
        {
            return;
        }

        Apply(new OrderProductDiscountChanged { ProductId = Id, NewDiscount = discount );
    }

    public void On(OrderProductCountIncreased e) => Count = e.NewCount;
    public void On(OrderProductCountDecreased e) => Count = e.NewCount;
    public void On(OrderProductDiscountChanged e) => Discount = e.NewDiscount;
}
5
  • 1
    Seems like FSMs intertwined with event-driven task processing. (I picture a spaghetti monster on a mecha frame... on a trebuchet.) Can you tell the bigger goal outside of that scheme? Also, an execution graph diagram could be really helpful. More specifically I'd like to know if there are recursions, and how a command is different from any other code (eg event handlers) that "sets things into motion". And how long does it keeps executing after the command is done. Apr 20 at 11:07
  • Post was updated to contain additional information - let me know if it's enough @ShadowsInRain
    – chris
    Apr 20 at 18:38
  • It's good that FSM is ruled out, the picture is much clearer now. Apr 20 at 19:37
  • Can an event cause another event, or what sort of logic causes subsequent events to be emitted? You have stated that events cannot have any logic — does it mean that something is watching or reading the event stream in order to emit more events? Apr 20 at 21:21
  • Directly after applying event, command handler can read state changed synchronously in event handler (either by direct apply of event, or nested in some command methods): Apply(new SomeEvent { fieldX = Y }); if (this.fieldX == Z) { Apply(new OtherEvent { ... }); } - ofc this directly does not make sense, but application of SomeEvent can be nested and Y value can be calculated based on state of aggregate/entities
    – chris
    Apr 21 at 10:37
1

Can't really tell from here, but there is a smell: changing multiple entities inside aggregate.

Maybe the solution is to have smaller aggregates and some Policy/Saga coordinating them. It really feels like your aggregates are too "fat".

1

Answering the title, the natural alternative is to use a background process chain. In the CQRS world, this is called a saga. This approach is pretty straightforward, but requires require some preliminary setup and a significant amount of additional work. More specifically, a saga would require explicit handling of possible data inconsistencies that are inevitable in the async context.


That said, your issues seem to be caused by granularity mismatch between commands and events. Which means that you should either make commands smaller or make events bigger. In turn, you may consider redesigning the domain.

• Making commands smaller will most probably lead back to sagas. Assuming commands were designed well, there is some unavoidable amount of processing that must be done, therefore splitting it in smaller pieces will lead to chaining multiple commands. Which is similar to what you're doing already, albeit more standardized.

• Making events bigger may be a good idea in some cases. If certain entities are (mostly/often) modified together, it makes sense to bundle them into a single event. For example, consider making a local clone of the state and modifying it directly, then assembling an event data from the difference between the two states. Alternatively, simply make events to do more work and/or span more entities.

• Redesign the aggregate. Fitting and entire department of affairs into a single aggregate is not unheard of, although it's certainly uncommon. Can't say much here without knowing the specifics. Perhaps it's a cause for another question.

4
  • I updated question with second example - I considered saga, but it has two cons for me: 1. It breaks encapsulation of order, where management of state is "outsourced" 2. I think it will complicate whole state management - now it's clear when discount has to be updated, when state changes happen etc. My current approach has that con, that it is not that clear command -> event(s) flow, but more like command -> event(s) -> logic -> event(s) -> (...) Look especially on method DecreaseAllOrderProductsCount - where all entites are updated, and then discount is updated once.
    – chris
    Apr 25 at 13:55
  • @chris I've worked with something pretty similar to your example: a humongous salad of entities and rules. The second bullet in my answer is the way to go. Try to manage everything in the memory. Perhaps build a specialized in-memory domain just for that. It may sound hard, but it should be simpler than interacting with persistence. In that domain, events shouldn't be needed anymore, you could replace events with meaningfully named functions that set dirty flags, then call a single centralized method that tries to clear said dirty flags in a loop. Apr 25 at 16:10
  • "events shouldn't be needed anymore" - events are one of my requirements - I need to go forward and back in time to see how objects changed and reason of that change (so simple info about raw state change is too little)
    – chris
    Apr 26 at 12:57
  • @chris You don't trust your commands that much? Well you could try good old logging for things that happen withing that big diff. In CQRS, rolling back a command partially makes little to no sense — because a command defines a logical boundary for changes — so I can only assume you want events for debugging purposes. Otherwise, it's another sign that you should make your commands more granular and chain them into a saga. Apr 26 at 19:29

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.