CustomerRevenueNode.cs

public class CustomerRevenueNode :
    Node<CustomerRevenueNode.State>,
    Node<CustomerRevenueNode.State>.IHandle<OrderPlaced>,
    Node<CustomerRevenueNode.State>.IHandle<OrderCancelled>,
    Node<CustomerRevenueNode.State>.IHandle<OrderDelivered>
{
    public class State
    {
        public Guid CustomerId { get; set; }
        public decimal AggregatedRevenue { get; set; }
        public IDictionary<Guid, decimal> PendingOrderAmounts { get; } = new Dictionary<Guid, decimal>();
    }
    
    private readonly ICustomerRevenueStore _store;

    public CustomerRevenueNode(ICustomerRevenueStore store)
    {
        _store = store;
    }

    public async Task<IMemory> AccessMemory(OrderPlaced orderPlaced)
    {
        var snapshot = await _store.GetByCustomerId(orderPlaced.CustomerId).ConfigureAwait(false);
        return new Memory(snapshot, nextState => _store.Save(nextState));
    }

    public async Task<IMemory> AccessMemory(OrderCancelled orderCancelled)
    {
        var snapshot = await _store.GetByOrderId(orderCancelled.OrderId).ConfigureAwait(false);
        return new Memory(snapshot, nextState => _store.Save(nextState));
    }

    public async Task<IMemory> AccessMemory(OrderDelivered orderDelivered)
    {
        var snapshot = await _store.GetByOrderId(orderDelivered.OrderId).ConfigureAwait(false);
        return new Memory(snapshot, nextState => _store.Save(nextState));
    }

    public ITransformResult Transform(OrderPlaced orderPlaced, CustomerRevenueNode.State state)
    {
        var nextState = state ?? new CustomerRevenueNode.State
        {
            CustomerId = orderPlaced.CustomerId
        };
        nextState.PendingOrderAmounts.Add(orderPlaced.Id, orderPlaced.Amount);
        return new Node<CustomerRevenueNode.State>.TransformResult(nextState);
    }

    public ITransformResult Transform(OrderCancelled orderCancelled, CustomerRevenueNode.State state)
    {
        var nextState = state;
        nextState.PendingOrderAmounts.Remove(orderCancelled.OrderId);
        return new Node<CustomerRevenueNode.State>.TransformResult(nextState);
    }

    public ITransformResult Transform(OrderDelivered orderDelivered, CustomerRevenueNode.State state)
    {
        var nextState = state;
        nextState.AggregatedRevenue += nextState.PendingOrderAmounts[orderDelivered.OrderId];
        nextState.PendingOrderAmounts.Remove(orderDelivered.OrderId);
        return new Node<CustomerRevenueNode.State>.TransformResult(nextState);
    }
}

public interface ICustomerRevenueStore
{
    Task<CustomerRevenueNode.State> GetByCustomerId(Guid customerId);
    Task<CustomerRevenueNode.State> GetByOrderId(Guid orderId);
    Task Save(CustomerRevenueNode.State state);
}