Fetch including uncommited during rebuild of projection

[ChristPetitjean]

Hello,

I have a "simple" use case but it seems to be impossible to tackle.

I have this simple aggregate:

// Stars is the global rating over users rates
public record House(Guid Id, string Name, string Address, int NumberOfRooms, decimal? Stars);

I have this event:

public sealed record UserRate(int Rate);

I have also this simple projection with this Apply:

public class HouseProjection : SingleStreamProjection<House>
{
    // ....

    public async Task<House> Apply(UserRate @event, House house, IQuerySession session)
    {
        // When multiple Apply happens before being commited, this does not work as
        // it only read commited from databse
        // So, how to include uncommited changes during rebuild ?
        var events = await session.Events.FetchStreamAsync(house.Id);

        var stars = events
            .Where(e => e.EventType == typeof(UserRate))
            .Select(e => Convert.ToDecimal(((UserRate)e.Data).Rate))
            .Union([Convert.ToDecimal(@event.Rate)]);
        
        return house with
        {
            Stars = stars.Average(), // So this average does not work when there is more than one UserRate  in the transaction
        };
    }
   
}

This seems to work if there is only one UserRate event in the transction.

BUT

If I have multiple UserRate events in the same transaction, it is going wrong as it only include commited events.
A simple exemple of this is the rebuild of the projection that will commit everything at once.

[jakobt]

As i understand what is not working for you is if you append multiple events your projection value will be based on the persisted event + the most recent event?

i.e. you do:

• UserRate(2)
  ... SaveChanges -> Average = 2
• UserRate(3)
  ... SaveChanges -> Average = 2.5
• UserRate(1)
  ... SaveChanges -> Average = 2

However if you do

• UserRate (2)
  ... SaveChanges -> Average 2
• UserRate(3)
• UserRate(1)
  ... SaveChanges -> Average 1.5

Is that correctly understood, or is there something else not working?

You can make this work in multiple ways however I would probably change the design instead, but here are some options if you are hellbent on this way of doing things or if you encounter another issue where you need access to the events being applied in the current session:

1. Save the current events on the aggregation and use them in your calculation, so having a List of UserRates that you append to when you see the event and use that in your calculation something like:

public List<UserRate> _seenUserRates = new List<UserRate>();

//...

public async Task<House> Apply(UserRate @event, House house, IQuerySession session)
{
    _seenUserRates.Add(@event);
    var events = await session.Events.FetchStreamAsync(house.Id);

    var stars = events
        .Where(e => e.EventType == typeof(UserRate))
        .Select(e => Convert.ToDecimal(((UserRate)e.Data).Rate))
        .Union(_seenUserRates.Select(x => Convert.ToDecimal(x.Rate)));
    
    return house with
    {
        Stars = stars.Average(), 
    };
}

2. Implement an IProjection which default behaviour is that you process all outstanding events for the stream and not single events. See https://martendb.io/events/projections/custom.html

However as mentioned I would simplify this code by storing something like NumberOfRatings and TotalStars and calculate the average based on that so something like:

public async Task<House> Apply(UserRate @event, House house, IQuerySession session)
{   
    var numberOfRatings = house.NumberOfRatings + 1;
    var totalStars = house.TotalStars + @event.Rate;
    return house with
    {
        NumberOfRatings = numberOfRatings,
        TotalStars = totalStars,
        Stars = totalStars/numberOfRatings
    };
}

[ChristPetitjean]

You perfectly understood the issue. So the optimal workaround saoule be to redesign a littéralement bit. That is fair enough to add the number of rates in the aggregate but the issue only happens during rebuild currently.

So I was wondering if there was a way and as you mentionned, I could create a custom projection.

Thanks a lot.

[jeremydmiller]

@ChristPetitjean Slow down a second first, why are you calling this inside your projection: var events = await session.Events.FetchStreamAsync(house.Id);

That's very highly not idiomatic, and in any kind of rebuild or async scenario, that's a giant N+1 you don't want. You'd ideally want your projection to work by incrementally applying changes to the snapshot, but never need to look backwards through the event stream.

[ChristPetitjean]

Hello, that is a concept I just come with. You are right aggregate should be incremental, this solve a lots of problems.

Thanks for the highlight