Muflone.Projections.MongoDb

Read model projections for Muflone on MongoDB.

Two events of the same aggregate, consumed off different queues, can reach a projection out of order. Whichever lands last wins — and that is not the one that happened last. This package makes the write refuse to be overtaken, and recover when it can.

The version filter

ProjectionPersister<T> writes under a filter on ProjectionVersion, the stream revision of the last event applied. A write carrying an older revision does not overwrite a newer document: it is rejected, and says so.

public sealed class CartPersister(IMongoDatabase database, ILoggerFactory loggerFactory)
    : ProjectionPersister<Cart>(database, loggerFactory)
{
    public Task<bool> SetStatus(Guid id, string status, long revision, DateTime editDate, CancellationToken ct)
        => TryApply(id, revision, Builders<Cart>.Update.Set(x => x.Status, status),
                    upsert: false, editDate, ct: ct);
}

The coordinator

A rejection has three causes, and only one of them is a loss. IProjectionCoordinator tells them apart by reading the version actually stored:

stored version	meaning	what happens
absent	the event's predecessor is still in flight	retry, with a short ladder
older than the event	transient conflict	retry
equal	broker redelivery	ignored, quietly
newer	overtaken, permanently lost	logged at Error, no retry

It also orders writes per aggregate and serialises them, so the retries do not race each other.

services.AddSingleton<IProjectionCoordinator>(_ => new ProjectionCoordinator());

Register it as a singleton: its per-aggregate structures are the only thing stopping two read models from overtaking each other on the same stream.

var applied = await coordinator.ApplyAsync(
    new ProjectionWrite(
        cartId,
        eventRevision,
        ct => persister.SetStatus(cartId, status, eventRevision, editDate, ct),
        ct => persister.GetProjectionVersionAsync(cartId, ct),
        logger,
        nameof(CartService)),
    ct);

Documents

Implement IProjectionDocument: Id, UpdatedAt, ProjectionVersion. The last two are written by the persister — do not set them yourself.

License

MIT.