Koto.EventSourcing.Marten 0.4.0

Koto.EventSourcing.Marten

Event Sourcing on PostgreSQL via Marten. Part of the Koto suite.

Install

dotnet add package Koto.EventSourcing.Marten

What's included

Usage

1 — Define an event-sourced aggregate

public sealed class Order : EventSourcedAggregateRoot<OrderId>
{
    public decimal Total { get; private set; }
    public bool IsShipped { get; private set; }

    public static Order Place(OrderId id, decimal total)
    {
        var order = new Order();
        order.RaiseEvent(new OrderPlaced(Guid.NewGuid(), DateTime.UtcNow, id, total));
        return order;
    }

    public void Ship() => RaiseEvent(new OrderShipped(Guid.NewGuid(), DateTime.UtcNow));

    protected override void Apply(IDomainEvent @event)
    {
        switch (@event)
        {
            case OrderPlaced e: Total = e.Total; break;
            case OrderShipped:  IsShipped = true; break;
        }
    }

    private Order() { }
}

2 — Register

builder.Services.AddKotoMarten(
    builder.Configuration.GetConnectionString("Db")!,
    opts =>
    {
        // optional extra Marten config
        opts.Projections.Add<OrderSummaryProjection>(ProjectionLifecycle.Async);
    });

3 — Use the repository

public class PlaceOrderHandler : ICommandHandler<PlaceOrderCommand, OrderId>
{
    private readonly IEventSourcedRepository<Order, OrderId> _orders;

    public async Task<Result<OrderId>> HandleAsync(PlaceOrderCommand cmd, CancellationToken ct = default)
    {
        var order = Order.Place(OrderId.New(), cmd.Total);
        await _orders.SaveAsync(order, ct);
        return order.Id;
    }
}

4 — Write a projection

public sealed class OrderSummaryProjection : AsyncProjection<OrderSummary, Guid>
{
    public OrderSummary Create(OrderPlaced e) => new(e.OrderId, e.Total, false);
    public OrderSummary Apply(OrderShipped _, OrderSummary current) =>
        current with { IsShipped = true };
}

Using Marten alongside EF Core

Marten and Koto.Infrastructure.EFCore can share the same PostgreSQL instance. Marten manages event streams (mt_events, mt_streams); EF Core manages read models and other entities.

Atomic multi-stream commits

// перевод: события в двух стримах + идемпотентный маркер — одна транзакция
accounts.Append(from);
accounts.Append(to);
session.Store(new ProcessedOperation($"transfer:{operationId}"));
await unitOfWork.CommitAsync(ct);   // MartenUnitOfWork, регистрируется AddKotoMarten (TryAddScoped)

SaveAsync остаётся одноагрегатным шорткатом (Append + commit). Uncommitted events агрегатов очищаются только после успешного коммита; RollbackAsync отбрасывает всё застейдженное.