Event Sourcing in .NET: Aggregates, Event Stores, and Projections

Event sourcing stores an ordered history of events as the authoritative record of an aggregate. In .NET, commands validate business rules, events update state, and an event store appends those events only if the loaded stream version still matches. Projections build queryable views, while snapshots can reduce the cost of loading long histories.

What Are We Building?

An order can move through OrderCreated, LineItemAdded, and OrderConfirmed events. Loading it means replaying that history in order. For the design tradeoffs, start with the introduction to event sourcing.

The following example is a complete in-memory learning implementation, using only the .NET standard libraries. It demonstrates aggregate replay, atomic expected-version checks, a repository, and a projection. The event store loses its contents on restart; the final sections explain durable persistence and snapshots.

Create a console project and put the types below in Ordering.cs:

dotnet new console -n EventSourcingDemo --framework net10.0
cd EventSourcingDemo

Define Events and the Aggregate

The example prices every order in USD. In an HTTP application, load prices from a trusted catalog before calling the aggregate; do not accept a caller-supplied price as authoritative.

public abstract record OrderEvent(Guid OrderId);
public sealed record OrderCreated(Guid OrderId, Guid CustomerId)
    : OrderEvent(OrderId);
public sealed record LineItemAdded(Guid OrderId, Guid ProductId,
    int Quantity, decimal UnitPrice) : OrderEvent(OrderId);
public sealed record OrderConfirmed(Guid OrderId, DateTimeOffset ConfirmedAt)
    : OrderEvent(OrderId);
public sealed record OrderCancelled(Guid OrderId, string Reason)
    : OrderEvent(OrderId);

public sealed record StoredEvent(int Version, OrderEvent Data);
public enum OrderStatus { Draft, Confirmed, Cancelled }

public sealed class Order
{
    private readonly List<OrderEvent> _pending = [];
    private int _lineCount;

    private Order() { }

    public Guid Id { get; private set; }
    public Guid CustomerId { get; private set; }
    public OrderStatus Status { get; private set; }
    public decimal TotalUsd { get; private set; }
    public int Version { get; private set; } = -1;
    public IReadOnlyList<OrderEvent> Pending => _pending.AsReadOnly();

    public static Order Create(Guid customerId)
    {
        if (customerId == Guid.Empty)
            throw new ArgumentException("A customer is required.");

        var order = new Order();
        order.Raise(new OrderCreated(Guid.NewGuid(), customerId));
        return order;
    }

    public void AddLineItem(Guid productId, int quantity, decimal unitPrice)
    {
        if (Status != OrderStatus.Draft)
            throw new InvalidOperationException("Only drafts accept items.");
        if (productId == Guid.Empty || quantity <= 0 || unitPrice < 0)
            throw new ArgumentException("Invalid line item.");

        Raise(new LineItemAdded(Id, productId, quantity, unitPrice));
    }

    public void Confirm(DateTimeOffset now)
    {
        if (Status != OrderStatus.Draft || _lineCount == 0)
            throw new InvalidOperationException("A nonempty draft is required.");

        Raise(new OrderConfirmed(Id, now));
    }

    public void Cancel(string reason)
    {
        ArgumentException.ThrowIfNullOrWhiteSpace(reason);
        if (Status == OrderStatus.Cancelled)
            throw new InvalidOperationException("The order is already cancelled.");

        Raise(new OrderCancelled(Id, reason));
    }

    private void Raise(OrderEvent data)
    {
        Apply(data);
        _pending.Add(data);
    }

    private void Apply(OrderEvent data)
    {
        switch (data)
        {
            case OrderCreated created:
                Id = created.OrderId;
                CustomerId = created.CustomerId;
                Status = OrderStatus.Draft;
                break;
            case LineItemAdded item:
                TotalUsd += item.Quantity * item.UnitPrice;
                _lineCount++;
                break;
            case OrderConfirmed:
                Status = OrderStatus.Confirmed;
                break;
            case OrderCancelled:
                Status = OrderStatus.Cancelled;
                break;
            default:
                throw new InvalidOperationException("Unsupported order event.");
        }
    }

    public static Order Rehydrate(IReadOnlyList<StoredEvent> history)
    {
        if (history.Count == 0 || history[0].Data is not OrderCreated)
            throw new InvalidOperationException("The stream must start with creation.");

        var order = new Order();
        foreach (var stored in history)
        {
            if (stored.Version != order.Version + 1 ||
                (order.Id != Guid.Empty && stored.Data.OrderId != order.Id) ||
                (stored.Version > 0 && stored.Data is OrderCreated))
            {
                throw new InvalidOperationException("Invalid stream history.");
            }

            order.Apply(stored.Data);
            order.Version = stored.Version;
        }
        return order;
    }

    public void MarkCommitted()
    {
        Version += _pending.Count;
        _pending.Clear();
    }
}

Version tracks the last committed event: -1 means no events, and the first event has version 0. Raising events changes working state without advancing that version. The repository advances it only after a successful append, so saving the same aggregate again uses the correct expected version.

Replay calls Apply without raising new events or rerunning current business validation. Structural checks still reject gaps, mixed streams, and unknown event types. Keep timestamps, external lookups, and message delivery out of Apply so the same history always produces the same state.

Append With an Expected Version

Add the store to Ordering.cs. The lock makes the version check and append one atomic operation within this process:

public sealed class StreamConflictException(Guid streamId)
    : Exception($"Stream {streamId} changed after it was loaded.");

public sealed class InMemoryEventStore
{
    private readonly object _gate = new();
    private readonly Dictionary<Guid, List<StoredEvent>> _streams = [];

    public IReadOnlyList<StoredEvent> Read(Guid id)
    {
        lock (_gate)
        {
            return _streams.TryGetValue(id, out var stream)
                ? stream.ToArray()
                : Array.Empty<StoredEvent>();
        }
    }

    public void Append(Guid id, int expectedVersion,
        IReadOnlyList<OrderEvent> events)
    {
        lock (_gate)
        {
            _streams.TryGetValue(id, out var stream);
            var actualVersion = (stream?.Count ?? 0) - 1;
            if (actualVersion != expectedVersion)
                throw new StreamConflictException(id);
            if (events.Any(data => data.OrderId != id))
                throw new ArgumentException("Events belong to another stream.");
            if (events.Count == 0)
                return;

            var next = events.Select((data, index) =>
                new StoredEvent(expectedVersion + index + 1, data)).ToArray();
            stream ??= [];
            stream.AddRange(next);
            _streams[id] = stream;
        }
    }
}

public sealed class OrderRepository(InMemoryEventStore store)
{
    public Order Load(Guid id)
    {
        var history = store.Read(id);
        if (history.Count == 0)
            throw new KeyNotFoundException($"Order {id} was not found.");

        return Order.Rehydrate(history);
    }

    public void Save(Order order)
    {
        store.Append(order.Id, order.Version, order.Pending);
        order.MarkCommitted();
    }
}

Two callers that loaded version 3 both attempt to append version 4. Only one can pass the version check. After a conflict, discard the stale aggregate, reload, and reevaluate the command; a newly confirmed order may no longer accept the requested item.

Use each aggregate instance for one command at a time. The store supports concurrent callers, but an aggregate is a mutable working model and is not thread-safe.

Run the Example

Replace Program.cs with the following:

var store = new InMemoryEventStore();
var repository = new OrderRepository(store);

var order = Order.Create(Guid.NewGuid());
order.AddLineItem(Guid.NewGuid(), 2, 25m);
repository.Save(order); // Created=0, LineItemAdded=1.

order.Confirm(DateTimeOffset.UtcNow);
repository.Save(order); // Confirmed=2, using expected version 1.

var loaded = repository.Load(order.Id);
Console.WriteLine($"{loaded.Status}: USD {loaded.TotalUsd}, v{loaded.Version}");

Run it with dotnet run. The result is a confirmed order for USD 50 at version 2. The second save is intentional: it demonstrates that accepting committed events advances the aggregate version.

Build an Idempotent Projection

A projection turns stored events into a query-oriented view. This pure function updates one order view, ignores an already-applied version, and rejects a gap:

public sealed record OrderView(Guid Id, Guid CustomerId,
    OrderStatus Status, decimal TotalUsd, int Version);

public static class OrderProjection
{
    public static OrderView Apply(OrderView? view, StoredEvent stored)
    {
        if (view is not null && view.Id != stored.Data.OrderId)
            throw new InvalidOperationException("Projection stream mismatch.");
        if (view is not null && stored.Version <= view.Version)
            return view;
        if (stored.Version != (view?.Version ?? -1) + 1)
            throw new InvalidOperationException("A preceding event is missing.");

        return (view, stored.Data) switch
        {
            (null, OrderCreated e) => new(e.OrderId, e.CustomerId,
                OrderStatus.Draft, 0, stored.Version),
            (not null, LineItemAdded e) => view with
            {
                TotalUsd = view.TotalUsd + e.Quantity * e.UnitPrice,
                Version = stored.Version
            },
            (not null, OrderConfirmed) => view with
            {
                Status = OrderStatus.Confirmed, Version = stored.Version
            },
            (not null, OrderCancelled) => view with
            {
                Status = OrderStatus.Cancelled, Version = stored.Version
            },
            _ => throw new InvalidOperationException("Unexpected projection event.")
        };
    }
}

Apply it by iterating store.Read(order.Id) with an initially null view. All four event types affect the read model; leaving out line items would incorrectly show every order with a zero total.

This function demonstrates transitions and duplicate handling, not a durable subscription worker. In a database-backed projector, save the view and its consumed version in the same transaction, with a concurrency token or single-writer ownership. Only acknowledge delivery after that commit.

The committed checkpoint prevents a crash between applying an event and recording progress from double-counting it. For a global subscription, also persist the event store's global cursor; versions are only ordered within each stream. These choices matter when CQRS read models run independently of writes.

Persist Events and Add Snapshots Carefully

A durable store needs an atomic expected-version check and append across application instances. A PostgreSQL implementation commonly uses a unique (stream_id, version) constraint, plus validation that the current stream version equals the caller's expected version. Insert a batch in one transaction so a failed append cannot leave half a command's events committed.

Classify database errors precisely. A violation of that particular stream-version constraint signals a conflict; authentication errors and unavailable connections are different problems. EF Core documents that insert uniqueness violations use provider-specific exceptions, rather than DbUpdateConcurrencyException.

Use stable event names and schema versions in persisted envelopes, not CLR assembly-qualified names. Preserve old event meanings, and add explicit upcasters or compatible handlers when schemas evolve. A network failure after commit has an ambiguous outcome; command IDs and deduplication are needed to prevent reapplying a business operation on retry.

A snapshot caches the aggregate's state and committed version at a point in its stream. For this example that state includes the order ID, customer ID, status, total, and line count. Restore all of them, set Version to the snapshot version, and then replay only events with higher versions, checking for gaps. Starting replay from -1 after loading a snapshot produces incorrect concurrency checks.

Snapshots need their own format version and must not contain pending events. If an old snapshot cannot be read, discard it and replay the complete event stream. Add this optimization only after measuring loading latency; event count alone does not determine whether it is necessary.

For a production system, evaluate Marten or a dedicated event database instead of treating this in-memory store as a persistence layer. You still own event design, retention, compatibility, projection recovery, and operational monitoring.

Summary

• Commands validate rules before raising events; replay rebuilds state deterministically.
• Expected versions detect concurrent changes, and successful saves advance the committed version.
• Projections handle every relevant event and commit state with their checkpoint.
• Durable storage adds cross-process atomicity, schema evolution, and retry deduplication.
• Snapshots cache full committed state and a version; they are optional and rebuildable.

Event sourcing is useful when durable business history and replay justify the added storage and operational complexity. A conventional aggregate with state-based persistence is often sufficient when only current state matters.

Thanks for reading.

And stay awesome!