Event Sourcing with reckon-db

Event Sourcing is an architectural pattern where the state of an application is determined by a sequence of events. Instead of storing just the current state, you store the complete history of state changes as immutable events.

What is Event Sourcing?

Traditional CRUD-based systems store only the current state:

User Record: {id: 123, name: "Alice", email: "alice@example.com", balance: 150}

Event-sourced systems store the history of changes:

Event 1: UserCreated {id: 123, name: "Alice", email: "alice@example.com"}
Event 2: BalanceDeposited {user_id: 123, amount: 200}
Event 3: BalanceWithdrawn {user_id: 123, amount: 50}

The current state is derived by replaying these events.

Benefits of Event Sourcing

Complete Audit Trail

Every change is recorded with a timestamp and metadata. This is invaluable for:

• Regulatory compliance (financial systems, healthcare)
• Debugging production issues
• Understanding user behavior

Temporal Queries

You can reconstruct the state at any point in time:

%% Get account balance as of last month
{ok, Events} = reckon_db_streams:read(my_store, <<"account-018f6a7b8c9d4e5f60718293a4b5c6d8">>, 0, 1000, forward),
PastEvents = [E || E <- Events, E#event.timestamp < LastMonthTimestamp],
Balance = lists:foldl(fun apply_event/2, 0, PastEvents).

Event Replay

Rebuild read models, fix bugs in projections, or create new views of historical data:

%% Rebuild a projection from scratch: page through the whole store's log,
%% 1000 events at a time. read/5 reads one stream by its exact id; there is
%% no wildcard id.
replay(Offset) ->
    {ok, Events} = reckon_db_streams:read_all_global(my_store, Offset, 1000),
    lists:foreach(fun(E) -> update_projection(E) end, Events),
    replay_rest(Offset + length(Events), Events).

replay_rest(_Offset, []) -> done;
replay_rest(Offset, _Events) -> replay(Offset).

Decoupled Systems

Events can be consumed by multiple subscribers independently:

Event Sourcing with reckon-db

Streams

A stream is an ordered sequence of events sharing a common identifier (the stream ID). A stream usually holds one aggregate, such as one order or one user. Its id is <prefix>-<32 hex>, for example order-018f6a7b8c9d4e5f60718293a4b5c6d9, and the store refuses any other shape; Stream Ids in reckon-gater states the rule and how to make one.

%% Append events to a stream
Events = [
    #{
        event_type => <<"OrderPlaced">>,
        data => #{order_id => <<"ord-123">>, items => [...], total => 9999},
        metadata => #{user_id => <<"user-456">>, correlation_id => <<"req-789">>}
    }
],
{ok, Version} = reckon_db_streams:append(my_store, <<"order-018f6a7b8c9d4e5f60718293a4b5c6d9">>, -1, Events).

Events

Events are immutable facts that have happened. They should:

• Be named in past tense (e.g., OrderPlaced, not PlaceOrder)
• Contain all information needed to understand what happened
• Be business-meaningful (e.g., AccountOverdrawn, not BalanceUpdated)

%% Event structure
#{
    event_type => <<"OrderPlaced">>,      %% What happened
    data => #{                             %% The event payload
        order_id => <<"ord-123">>,
        customer_id => <<"cust-456">>,
        items => [
            #{product_id => <<"prod-1">>, quantity => 2, price => 1999}
        ],
        total => 3998
    },
    metadata => #{                         %% Cross-cutting concerns
        correlation_id => <<"req-abc">>,   %% Traces related operations
        causation_id => <<"evt-xyz">>,     %% What caused this event
        user_id => <<"user-789">>,         %% Who triggered it
        timestamp => 1703001234567         %% When it happened
    }
}

Note: When using evoq (the CQRS/Event Sourcing framework), events are wrapped in a full envelope with additional fields. The simplified format above shows what you provide; evoq adds event_id, stream_id, version, tags, timestamp, and epoch_us. See evoq Event Envelope Guide for complete details.

Metadata Standardization

Required metadata fields for all events:

metadata => #{
    correlation_id => <<"req-abc">>,   %% Request ID (trace across services)
    causation_id => <<"cmd-xyz">>,     %% What caused this event
    timestamp => erlang:system_time(millisecond)
}

Optional but recommended for user-triggered events:

metadata => #{
    user_id => <<"usr-123">>,          %% Who triggered it
    ip_address => <<"192.168.1.1">>,   %% Where from
    user_agent => <<"...">>             %% Client info
}

Optional but recommended for system events:

metadata => #{
    triggered_by => <<"system">>,
    reason => <<"scheduled_task">>,
    task_id => <<"task-abc">>
}

Optimistic Concurrency

reckon-db uses optimistic concurrency control to prevent conflicting writes:

%% Expected version semantics:
%% -1 (NO_STREAM): Stream must not exist (first write)
%% -2 (ANY_VERSION): Land after whatever is already there
%% N >= 0: Stream version must equal N

%% First write to a new stream
{ok, 0} = reckon_db_streams:append(Store, <<"order-018f6a7b8c9d4e5f60718293a4b5c6d9">>, -1, [Event1]).

%% Subsequent writes must specify expected version
{ok, 1} = reckon_db_streams:append(Store, <<"order-018f6a7b8c9d4e5f60718293a4b5c6d9">>, 0, [Event2]).

%% Concurrent writes will fail with version mismatch
%% Process A reads version 1
%% Process B reads version 1
%% Process A writes with expected version 1 -> succeeds, version is now 2
%% Process B writes with expected version 1 -> fails! (wrong_expected_version)

Two appends that race for the same version never overwrite each other. With ANY_VERSION the loser re-reads the version and tries again with a delay that doubles up to 50 ms, and answers {error, {version_conflict, ExpectedVersion, CurrentVersion}} if it loses 60 times in a row; with an explicit expected version it fails its own check, as above. An append of no events writes nothing and answers the version it read.

A scavenged stream that still has events is refused: the version is the count of live version nodes, a scavenge deletes old ones without renumbering the rest, and the append checks both the node it would write and the node before it, so every prefix shape answers {error, {version_unverifiable, Version, Reason}} (5.12.0 reads a per-stream counter and is exact again). A scavenge that deletes every event is the exception: get_version/2 answers -1 and the next ANY_VERSION append lands at version 0 again - a one-event append answers {ok, 0}, a batch of N answers {ok, N - 1} - nothing is overwritten, and a subscriber that re-reads by version sees 0, 1, ... again. See guides/storage_internals.md for the store's view. reckon-gater does not know either error yet, treats an unrecognised one as transient, and so burns ten retries - about 81 to 89 seconds - before answering {error, {retries_exhausted, Reason}}; reckon-evoq and evoq pass both through unchanged.

Designing Events

Event Naming

Use past tense verbs that describe business facts:

Event Granularity

Events should be atomic business facts. Avoid:

• Generic events like EntityUpdated (not meaningful)
• Overly fine-grained events (one per field change)
• Composite events (multiple unrelated changes)

%% Good: Specific, meaningful events
#{event_type => <<"AddressChanged">>, data => #{
    old_address => OldAddr,
    new_address => NewAddr,
    reason => <<"customer_request">>
}}

%% Bad: Generic, meaningless event
#{event_type => <<"CustomerUpdated">>, data => #{
    field => <<"address">>,
    value => NewAddr
}}

Event Versioning

Events are immutable, but schemas evolve. Use explicit versions:

%% Version 1
#{event_type => <<"OrderPlaced.v1">>, data => #{
    order_id => ...,
    items => [...]
}}

%% Version 2 (added shipping_address)
#{event_type => <<"OrderPlaced.v2">>, data => #{
    order_id => ...,
    items => [...],
    shipping_address => #{}
}}

Handle schema evolution in your projections:

handle_event(#{event_type := <<"OrderPlaced.v1">>} = Event) ->
    %% Default shipping address for v1 events
    upgrade_to_v2(Event);
handle_event(#{event_type := <<"OrderPlaced.v2">>} = Event) ->
    process_order(Event).

Building Aggregates

An aggregate is a domain object that encapsulates state and enforces invariants. In event sourcing, aggregates:

1. Load their state by replaying events
2. Validate commands against current state
3. Emit new events if the command succeeds

-module(order_aggregate).
-export([new/0, apply_event/2, place_order/2, add_item/3]).

-record(order, {
    id,
    status = pending,
    items = [],
    total = 0
}).

%% Create a new aggregate
new() -> #order{}.

%% Apply events to rebuild state
apply_event(#{event_type := <<"OrderPlaced">>} = E, _Order) ->
    Data = maps:get(data, E),
    #order{
        id = maps:get(order_id, Data),
        status = placed,
        items = maps:get(items, Data),
        total = maps:get(total, Data)
    };

apply_event(#{event_type := <<"ItemAdded">>} = E, Order) ->
    Data = maps:get(data, E),
    NewItem = #{
        product_id => maps:get(product_id, Data),
        quantity => maps:get(quantity, Data),
        price => maps:get(price, Data)
    },
    Order#order{
        items = [NewItem | Order#order.items],
        total = Order#order.total + (maps:get(quantity, Data) * maps:get(price, Data))
    };

apply_event(#{event_type := <<"OrderShipped">>}, Order) ->
    Order#order{status = shipped}.

%% Commands that produce events
place_order(OrderId, Items) ->
    Total = lists:sum([Q * P || #{quantity := Q, price := P} <- Items]),
    {ok, [#{
        event_type => <<"OrderPlaced">>,
        data => #{order_id => OrderId, items => Items, total => Total}
    }]}.

add_item(#order{status = placed} = _Order, ProductId, Quantity) ->
    Price = get_product_price(ProductId),
    {ok, [#{
        event_type => <<"ItemAdded">>,
        data => #{product_id => ProductId, quantity => Quantity, price => Price}
    }]};
add_item(#order{status = shipped}, _ProductId, _Quantity) ->
    {error, order_already_shipped}.

Loading Aggregate State

Use reckon_db_aggregator to rebuild aggregate state:

%% Load order aggregate from event stream
load_order(StoreId, OrderId) ->
    StreamId = <<"order-", OrderId/binary>>,

    %% Read events from stream
    {ok, Events} = reckon_db_streams:read(StoreId, StreamId, 0, 10000, forward),

    %% Apply events to initial state using custom apply function
    InitialState = order_aggregate:new(),
    FinalState = lists:foldl(
        fun(Event, Acc) -> order_aggregate:apply_event(Event, Acc) end,
        InitialState,
        Events
    ),
    FinalState.

%% Or use reckon_db_aggregator for tagged value aggregation
%% (useful when events have {sum, N} or {overwrite, V} tagged values)
load_order_with_aggregator(StoreId, OrderId) ->
    StreamId = <<"order-", OrderId/binary>>,
    {ok, Events} = reckon_db_streams:read(StoreId, StreamId, 0, 10000, forward),
    TaggedState = reckon_db_aggregator:foldl(Events),
    reckon_db_aggregator:finalize(TaggedState).

Further Reading

• CQRS Guide - Command Query Responsibility Segregation
• Subscriptions Guide - Real-time event notifications
• Snapshots Guide - Optimizing aggregate loading

References

• Martin Fowler: Event Sourcing
• Greg Young: CQRS and Event Sourcing
• Vaughn Vernon: "Implementing Domain-Driven Design" (Chapters 8-10)