Event Sourcing in C#: Storing State as Event Streams

Event sourcing is an architectural pattern where changes to application state are stored as an immutable sequence of events rather than updating records directly in a database. This article explores the core concepts of event sourcing, examines how it differs from traditional CRUD persistence, and demonstrates how to implement an aggregate root, capture state changes, and reconstruct state from event streams using modern C#.

Understanding Event Sourcing

In a traditional relational model (CRUD), if a user changes their address, the application executes an UPDATE statement that overwrites the existing row. The previous state is lost unless explicit audit logging is implemented.

Event sourcing approaches persistence differently. Instead of storing the current state, the application persists the full history of domain events that led to that state. Each event represents a business fact that occurred in the past (e.g., UserRegistered, AddressChanged, OrderPlaced). Because events represent history, they are append-only and immutable; once written, an event is never updated or deleted.

The current state of any entity is derived by replaying its historical stream of events in chronological order from the beginning.

Key Concepts

Domain Event: An immutable object representing something that happened in the domain.
Event Stream: The ordered sequence of events associated with a specific entity instance (aggregate).
Aggregate Root: The domain object responsible for validating commands, producing new events, and applying those events to update its internal state.
Rehydration: The process of instantiating an aggregate and replaying its event stream to rebuild its current state.

Implementing Event Sourcing in C

Modern C# features, such as record types, make defining and handling events concise and robust.

1. Define Domain Events

Events should be lightweight, immutable data structures representing business facts:

public interface IEvent
{
    Guid AggregateId { get; }
    DateTime OccurredOn { get; }
}

public record BankAccountCreated(Guid AccountId, string Owner, decimal InitialDeposit, DateTime OccurredOn) : IEvent
{
    public Guid AggregateId => AccountId;
}

public record MoneyDeposited(Guid AccountId, decimal Amount, DateTime OccurredOn) : IEvent
{
    public Guid AggregateId => AccountId;
}

public record MoneyWithdrawn(Guid AccountId, decimal Amount, DateTime OccurredOn) : IEvent
{
    public Guid AggregateId => AccountId;
}

2. Create the Aggregate Root

The aggregate enforces business invariants, stages uncommitted events, and updates its internal state via event handlers:

public class BankAccount
{
    private readonly List<IEvent> _uncommittedEvents = new();

    public Guid Id { get; private set; }
    public string Owner { get; private set; } = string.Empty;
    public decimal Balance { get; private set; }
    public int Version { get; private set; }

    public IReadOnlyCollection<IEvent> UncommittedEvents => _uncommittedEvents.AsReadOnly();

    public BankAccount() { }

    // Command: Open Account
    public static BankAccount Open(Guid id, string owner, decimal initialDeposit)
    {
        if (initialDeposit < 0)
            throw new ArgumentException("Initial deposit cannot be negative.");

        var account = new BankAccount();
        account.RaiseEvent(new BankAccountCreated(id, owner, initialDeposit, DateTime.UtcNow));
        return account;
    }

    // Command: Deposit
    public void Deposit(decimal amount)
    {
        if (amount <= 0)
            throw new ArgumentException("Deposit amount must be positive.");

        RaiseEvent(new MoneyDeposited(Id, amount, DateTime.UtcNow));
    }

    // Command: Withdraw
    public void Withdraw(decimal amount)
    {
        if (amount <= 0)
            throw new ArgumentException("Withdrawal amount must be positive.");

        if (Balance < amount)
            throw new InvalidOperationException("Insufficient funds.");

        RaiseEvent(new MoneyWithdrawn(Id, amount, DateTime.UtcNow));
    }

    // Record and apply event
    private void RaiseEvent(IEvent @event)
    {
        Apply(@event);
        _uncommittedEvents.Add(@event);
    }

    // Mutate internal state
    private void Apply(IEvent @event)
    {
        switch (@event)
        {
            case BankAccountCreated e:
                Id = e.AccountId;
                Owner = e.Owner;
                Balance = e.InitialDeposit;
                break;
            case MoneyDeposited e:
                Balance += e.Amount;
                break;
            case MoneyWithdrawn e:
                Balance -= e.Amount;
                break;
        }

        Version++;
    }

    // Replay historical events
    public void LoadFromHistory(IEnumerable<IEvent> history)
    {
        foreach (var @event in history)
        {
            Apply(@event);
        }
    }

    public void ClearUncommittedEvents() => _uncommittedEvents.Clear();
}

3. Storing and Appending to the Stream

An event store writes new events to an append-only store and fetches existing streams by aggregate identifier:

public interface IEventStore
{
    Task AppendEventsAsync(Guid aggregateId, IEnumerable<IEvent> events, int expectedVersion);
    Task<IReadOnlyList<IEvent>> GetStreamAsync(Guid aggregateId);
}

public class InMemoryEventStore : IEventStore
{
    private readonly Dictionary<Guid, List<IEvent>> _streams = new();

    public Task AppendEventsAsync(Guid aggregateId, IEnumerable<IEvent> events, int expectedVersion)
    {
        if (!_streams.TryGetValue(aggregateId, out var stream))
        {
            stream = new List<IEvent>();
            _streams[aggregateId] = stream;
        }

        if (stream.Count != expectedVersion)
        {
            throw new InvalidOperationException("Concurrency conflict detected.");
        }

        stream.AddRange(events);
        return Task.CompletedTask;
    }

    public Task<IReadOnlyList<IEvent>> GetStreamAsync(Guid aggregateId)
    {
        if (_streams.TryGetValue(aggregateId, out var stream))
        {
            return Task.FromResult<IReadOnlyList<IEvent>>(stream.AsReadOnly());
        }

        return Task.FromResult<IReadOnlyList<IEvent>>(Array.Empty<IEvent>());
    }
}

4. Executing Commands and Rehydrating State

The complete workflow involves fetching past events, rebuilding aggregate state, executing logic, and appending new events back to the stream:

public class BankAccountService
{
    private readonly IEventStore _eventStore;

    public BankAccountService(IEventStore eventStore)
    {
        _eventStore = eventStore;
    }

    public async Task ProcessDepositAsync(Guid accountId, decimal amount)
    {
        // 1. Fetch historical event stream
        var history = await _eventStore.GetStreamAsync(accountId);
        if (!history.Any())
            throw new KeyNotFoundException("Account does not exist.");

        // 2. Rehydrate aggregate to current state
        var account = new BankAccount();
        account.LoadFromHistory(history);
        var originalVersion = account.Version;

        // 3. Execute domain logic
        account.Deposit(amount);

        // 4. Persist uncommitted events
        await _eventStore.AppendEventsAsync(accountId, account.UncommittedEvents, originalVersion);
        account.ClearUncommittedEvents();
    }
}

Advantages and Trade-offs

Event sourcing provides a 100% accurate audit trail by default, enables time-travel debugging (reconstructing the exact state at any timestamp), and eliminates object-relational mapping impedance mismatches.

However, it introduces architectural complexity. Querying data across streams requires read-model projections (often implemented using CQRS), and long-lived streams may necessitate snapshots to prevent slow aggregate rehydration over time.