Skip to content

Persisting Projections

The built-in InMemoryProjectionRepository is perfect for getting started, but real read models need to survive restarts and scale beyond one process. You get that by implementing two contracts against a database of your choice, and by giving the observer a durable checkpoint store to match.

The Two Contracts

A projection is touched from two directions, and EventSourcingKit keeps them apart. The projector writes through IProjectionDataStore<TProjection, TId>:

public interface IProjectionDataStore<TProjection, TId>
    where TProjection : Projection<TId>
{
    Task<TProjection?> GetByIdIfExisting(
        TId id,
        CancellationToken cancellationToken
    );

    Task Create(TProjection projection, CancellationToken cancellationToken);
    Task Update(TProjection projection, CancellationToken cancellationToken);
    Task DeleteById(TId id, CancellationToken cancellationToken);
}

Your application reads through a separate interface, which you name as the fourth type argument to AddProjection. EventSourcingKit ships IProjectionRepository<T, TId> for that purpose:

public interface IProjectionRepository<T, TId>
{
    Task<T> GetById(TId id, CancellationToken cancellationToken);

    Task<IEnumerable<T>> GetManyById(
        IEnumerable<TId> ids,
        CancellationToken cancellationToken
    );

    Task<IEnumerable<T>> GetAll(CancellationToken cancellationToken);
}

You are free to define your own read interface instead, with the queries your application actually needs. One class usually implements both contracts, since reading and writing hit the same table or collection anyway.

Implementing the Store

With EF Core, the repository works against a DbContext and dbContext.Set<TProjection>():

public class ProjectionRepository<TProjection, TId>(MyDbContext dbContext)
    : IProjectionDataStore<TProjection, TId>,
        IProjectionRepository<TProjection, TId>
    where TProjection : Projection<TId>, new()
{
    public async Task<TProjection?> GetByIdIfExisting(
        TId id,
        CancellationToken cancellationToken
    ) =>
        await dbContext.Set<TProjection>().AsNoTracking()
            .FirstOrDefaultAsync(p => p.Id!.Equals(id), cancellationToken);

    public async Task Create(
        TProjection projection,
        CancellationToken cancellationToken
    )
    {
        dbContext.Set<TProjection>().Add(projection);
        await dbContext.SaveChangesAsync(cancellationToken);
    }

    // Update and DeleteById complete IProjectionDataStore;
    // GetById, GetManyById and GetAll serve the read interface.
}

With MongoDB, the same class works against an IMongoCollection<TProjection>. An upsert keeps the write path simple, since the observer may revisit a projection while catching up:

public async Task Update(
    TProjection projection,
    CancellationToken cancellationToken
)
{
    var filter = Builders<TProjection>.Filter.Eq(p => p.Id, projection.Id);
    var options = new FindOneAndReplaceOptions<TProjection> { IsUpsert = true };

    await _collection.FindOneAndReplaceAsync(
        filter,
        projection,
        options,
        cancellationToken
    );
}

Why GetManyById exists

Resolving a list of entities one by one turns a single request into N round trips – the problem a GraphQL data loader exists to solve. GetManyById is shaped for exactly that: hand it the batch of ids a data loader collected, and answer them all in one query.

Registering the Store

Register your database context first, then the projection, naming the projection type, its id type, your data store and the read interface you query through:

builder.Services.AddDbContext<MyDbContext>(options =>
    options.UseNpgsql(connectionString));

builder.Services.AddEventSourcingKit(builder.Configuration, [assembly])
    .AddProjection<
        Book,
        Guid,
        ProjectionRepository<Book, Guid>,
        IProjectionRepository<Book, Guid>
    >();

Keeping the Checkpoint Durable Too

A durable projection store on its own is not enough. The observer remembers its position through ICheckpointStore, and that store defaults to in-memory – so after a restart the observer would replay from the beginning into read models that already hold the data. Implement ICheckpointStore over the same database and register it alongside:

builder.Services.AddEventSourcingKit(builder.Configuration, [assembly])
    .AddCheckpointStore<MyCheckpointStore>()
    .AddProjection<
        Book,
        Guid,
        ProjectionRepository<Book, Guid>,
        IProjectionRepository<Book, Guid>
    >();

Projecting is idempotent, so a replay does not corrupt a read model – but it costs time that grows with your event history. Storing projections and checkpoints in the same database lets both move forward together.

For More Information