Benzene

Event Sourcing

Status: DRAFT v0.1 — part of the Patterns "at enterprise scale" set.

Regulated and audit-heavy domains — trading, payments, ledgers — often cannot store just the current state of an aggregate; they must store every change that led to it, as an immutable, ordered log. That is event sourcing: the aggregate's state is a fold over its event history, the log is the source of truth, and "what did this account look like last Tuesday?" and "prove how we got to this balance" are answerable by construction.

Benzene keeps its event-sourcing surface deliberately small. The Benzene.EventSourcing package gives you an append-only event store with optimistic concurrency — IEventStore (AppendAsync(streamId, expectedVersion, events) / ReadAsync), an InMemoryEventStore, and a DynamoDB-backed store in Benzene.EventSourcing.DynamoDb — but there is no aggregate-rehydration helper, no snapshot or replay framework: those you compose. Add a change-data-capture transport that makes the projection half nearly free, and you have the whole pattern. This document is explicit about that line, and works a trade ledger through it.

What Benzene gives you, and what you build

Piece Status
Command ingest — validate and decide Built-in — a message handler ([Message("account:debit")] IMessageHandler<Debit, …>) is the command handler.
The event log — append-only, ordered Built-in — Benzene.EventSourcing's IEventStore.AppendAsync(streamId, expectedVersion, events) is the append, with optimistic concurrency (a version clash throws EventStoreConcurrencyException); use InMemoryEventStore or the DynamoDB store (Benzene.EventSourcing.DynamoDb, one item per (streamId, version)). You can still hand-roll the write against your own data layer if you prefer.
Projections — turn events into read state Built-in transport — point DynamoDB Streams at the log and consume it with UseDynamoDb ([Message("ledger:INSERT")]); Benzene delivers each appended event, in shard order, to a projector. This is CQRS fed by the log.
Idempotency / exactly-once effect Built-in middleware — UseIdempotency() (+ a durable store) so a replayed event projects once.
Durable event evolution — old events, new code Built-in — AddPayloadVersioning upcasts historical event schemas to the current shape at the pipeline edge (below).
Aggregate rehydration, snapshots, replay orchestration App-level — you write these. Benzene.EventSourcing gives you the store (IEventStore), but deliberately imposes no aggregate base class, snapshotting, or replay driver.

The honest summary: Benzene gives you the ingest, the event store, the stream-projection consumer, idempotency, and event versioning; you own the rehydration fold, snapshots, and the replay driver. That is a deliberate small surface — the storage and concurrency are handled, but the event-sourcing conventions above vary enough that a framework abstraction usually gets in the way.

How you build it with Benzene

Event sourcing: a command handler appends events to an ordered log; a projector consumes the log via change-data-capture into a read model; the command handler rehydrates its state by folding the log before deciding.

1. The command handler appends an event

A command handler validates, loads the aggregate's current state (a fold of its events, below), decides, and appends the resulting event(s) to the log — an ordinary write to the append store:

(informative, .NET)

[Message("account:debit")]
public class DebitHandler : IMessageHandler<Debit, DebitAccepted>
{
    private readonly IEventLog _log;   // Benzene.EventSourcing's IEventStore, or your own store
    public DebitHandler(IEventLog log) => _log = log;

    public async Task<IBenzeneResult<DebitAccepted>> HandleAsync(Debit cmd)
    {
        var account = await Rehydrate(cmd.AccountId);           // fold events → current state (app code)
        if (account.Balance < cmd.Amount)
            return BenzeneResult.Invalid("insufficient-funds");  // a decision, returned as a result
        var evt = new AccountDebited(cmd.AccountId, cmd.Amount, account.NextSequence);
        await _log.AppendAsync(evt);        // the append IS the DynamoDB write; ordered by sequence
        return BenzeneResult.Ok(new DebitAccepted(evt.Sequence));
    }
}

Two Benzene-shaped details that matter for correctness:

2. The log emits its events — via change data capture

You do not publish the event from the handler (that would be the dual-write problem). Instead, the append is captured off the log's stream and projected — the transactional outbox, applied to the event log itself:

app.UseAwsLambda(events => events
    .UseDynamoDb(cdc => cdc
        .UseIdempotency()
        .UseMessageHandlers()));

[Message("ledger:INSERT")]                 // a newly-appended event, in shard order
public class ProjectBalance : IMessageHandler<AccountEvent>
{
    private readonly IReadStore _view;
    public ProjectBalance(IReadStore view) => _view = view;
    public Task<IBenzeneResult> HandleAsync(AccountEvent e)
        => _view.ApplyAsync(e);            // fold the event into the read model
}

Because DynamoDB Streams are shard-ordered and processed sequentially, resuming from the first failed record, events project in order, at least once — exactly what a ledger needs. The read models this feeds are ordinary CQRS views: current balances, statements, positions — each a projection of the same authoritative log.

3. Rehydration and snapshots (your code, Benzene-friendly shapes)

4. Events are immutable — so version them, never rewrite them

An event, once written, is history and cannot be edited. When the event's shape must evolve, you keep the old events exactly as written and upcast them to the current shape as they are read — Benzene's payload-schema versioning does this at the pipeline edge:

// AddPayloadVersioning is registered on the Benzene container (inside UsingBenzene).
services.UsingBenzene(x => x.AddPayloadVersioning(v => v.ForContext<DynamoDbRecordContext>()
    .Topic("ledger:INSERT", t => t
        .Version<AccountDebitedV1>("v1")
        .Version<AccountDebitedV2>("v2")
        .Upcast<AccountDebitedV1, AccountDebitedV2>(f => f.RegisterInitValue(e => e.Currency, "USD")))));

One projection handler on the latest schema; a decade of historical v1 events upcast on read; the caster graph validated at startup (a missing conversion path throws at boot, not on a 2015 event in production). Chain composition handles a long back-catalog. This is the mechanism that makes an append-only log survivable across years of schema change.

Worked example: a trade ledger

The system: every change to a trading account — trades booked, cash moved, fees applied — must be an immutable, ordered, auditable record; current positions and balances must be queryable fast; and "reconstruct the account as of any past instant" must be possible for audit and dispute.

When to use it

Checklist

Event sourcing is well-formed when:

See also: transactional outbox (the CDC mechanism that emits appended events), CQRS & read models (what the log projects into), and map-reduce (for replaying/aggregating a large log in parallel).