evoq

GitHub Sponsors

Erlang CQRS/Event Sourcing framework built on reckon-db.

Architecture Overview

Features

Installation

Add to your rebar.config:

{deps, [
    {evoq, "~> 1.26"}
]}.

Versions

Component Version
evoq (this repo) 1.26.1
telemetry (dep) 1.3.0
Erlang/OTP 27+

evoq is a standalone framework: it has no Reckon dependencies. telemetry is its only runtime dependency. Pair it with any event store through an adapter (see guides/adapters.md); the reckon-evoq adapter wires it to a Reckon store. The CCC payload-condition features require a store and adapter that expose payload indexes (reckon-gater >= 3.7 / reckon-db >= 5.3 via reckon-evoq >= 2.7), and only for decisions that use payload leaves.

Quick Start

Defining an Aggregate

-module(bank_account).
-behaviour(evoq_aggregate).

-export([init/1, execute/2, apply/2]).

init(_AccountId) ->
    {ok, #{balance => 0, status => active}}.

execute(#{status := closed}, _Command) ->
    {error, account_closed};
execute(_State, #{command_type := open_account, initial_balance := B}) ->
    {ok, [#{event_type => <<"AccountOpened">>, data => #{balance => B}}]};
execute(#{balance := Bal}, #{command_type := deposit, amount := A}) ->
    {ok, [#{event_type => <<"MoneyDeposited">>, data => #{amount => A}}]}.

apply(State, #{event_type := <<"AccountOpened">>, data := #{balance := B}}) ->
    State#{balance => B};
apply(#{balance := B} = State, #{event_type := <<"MoneyDeposited">>, data := #{amount := A}}) ->
    State#{balance => B + A}.

Dispatching Commands

%% Create a command
Command = evoq_command:new(
    deposit,           %% command type
    bank_account,      %% aggregate module
    <<"acc-123">>,     %% aggregate id
    #{amount => 100}   %% payload
),

%% Dispatch it
{ok, Version, Events} = evoq_router:dispatch(Command).

Event Handlers

-module(notification_handler).
-behaviour(evoq_event_handler).

-export([interested_in/0, init/1, handle_event/4]).

interested_in() ->
    [<<"AccountOpened">>, <<"LargeDeposit">>].

init(_Config) ->
    {ok, #{}}.

handle_event(<<"AccountOpened">>, Event, _Metadata, State) ->
    send_welcome_email(Event),
    {ok, State};
handle_event(<<"LargeDeposit">>, Event, _Metadata, State) ->
    send_deposit_alert(Event),
    {ok, State}.

%% The side effect is the handler's own: post the mail, page the operator,
%% publish a fact. Here it is a log line, so the example runs as it stands.
send_welcome_email(Event) ->
    logger:notice("welcome: ~p", [Event]),
    ok.

send_deposit_alert(Event) ->
    logger:warning("large deposit: ~p", [Event]),
    ok.

Process Managers

-module(order_fulfillment_pm).
-behaviour(evoq_process_manager).

-export([interested_in/0, correlate/2, handle/3, apply/2]).

interested_in() ->
    [<<"OrderPlaced">>, <<"PaymentReceived">>, <<"ItemShipped">>].

%% The last event of the process correlates as {stop, _}; anything else
%% as false, so correlate/2 never raises.
correlate(#{event_type := <<"ItemShipped">>, data := #{order_id := OrderId}}, _Meta) ->
    {stop, OrderId};
correlate(#{data := #{order_id := OrderId}}, _Meta) ->
    {continue, OrderId};
correlate(_Event, _Meta) ->
    false.

handle(State, #{event_type := <<"OrderPlaced">>, data := #{order_id := OrderId}}, _Meta) ->
    Cmd = evoq_command:new(process_payment, payment, OrderId, #{}),
    {ok, State, [Cmd]};
handle(State, #{event_type := <<"PaymentReceived">>, data := #{order_id := OrderId}}, _Meta) ->
    Cmd = evoq_command:new(ship_item, shipping, OrderId, #{}),
    {ok, State, [Cmd]};
handle(State, #{event_type := <<"ItemShipped">>}, _Meta) ->
    {ok, State#{status => completed}}.

apply(State, _Event) ->
    State.

Projections

-module(account_summary_projection).
-behaviour(evoq_projection).

-export([interested_in/0, init/1, project/4]).

interested_in() ->
    [<<"AccountOpened">>, <<"MoneyDeposited">>, <<"MoneyWithdrawn">>].

init(_Config) ->
    {ok, ReadModel} = evoq_read_model:new(evoq_read_model_ets, #{}),
    {ok, #{}, ReadModel}.

%% The read model is keyed on the account, which is the event's business
%% id. stream_id is the stream's name, so derive the entity from the data.
project(#{event_type := <<"AccountOpened">>, data := #{account_id := Id, balance := B}},
        _Metadata, State, ReadModel) ->
    {ok, NewRM} = evoq_read_model:put(Id, #{balance => B, tx_count => 0}, ReadModel),
    {ok, State, NewRM};
project(#{event_type := <<"MoneyDeposited">>, data := #{account_id := Id, amount := A}},
        _Metadata, State, ReadModel) ->
    {ok, Current} = evoq_read_model:get(Id, ReadModel),
    Updated = Current#{
        balance => maps:get(balance, Current) + A,
        tx_count => maps:get(tx_count, Current) + 1
    },
    {ok, NewRM} = evoq_read_model:put(Id, Updated, ReadModel),
    {ok, State, NewRM}.

Core Behaviors

Domain Artifacts

Domain artifacts stay inside the bounded context. They use atom keys and Erlang terms.

evoq_command

Commands represent intentions to change state. Formal contract for command modules.

-behaviour(evoq_command).

%% Required
-callback command_type() -> atom().
-callback new(Params :: map()) -> {ok, Command :: term()} | {error, Reason :: term()}.
-callback to_map(Command :: term()) -> map().

%% Optional
-callback validate(Command :: term()) -> ok | {ok, Command :: term()} | {error, Reason :: term()}.
-callback from_map(Map :: map()) -> {ok, Command :: term()} | {error, Reason :: term()}.

evoq_event

Events represent facts that have happened. Immutable once stored.

-behaviour(evoq_event).

%% Required
-callback event_type() -> atom() | binary().
-callback new(Params :: map()) -> Event :: term().
-callback to_map(Event :: term()) -> map().

%% Optional
-callback from_map(Map :: map()) -> {ok, Event :: term()} | {error, Reason :: term()}.

Integration Artifacts

Integration artifacts cross bounded context boundaries. They use binary keys and are JSON-serializable.

evoq_fact

Facts translate domain events into payloads for external consumption via pg or mesh.

-behaviour(evoq_fact).

%% Required
-callback fact_type() -> binary().          %% e.g., <<"ventures.venture_initiated">>
-callback from_event(EventType :: atom(), EventData :: map(), Metadata :: map()) ->
    {ok, Payload :: map()} | skip.

%% Optional (defaults use OTP 27 json module)
-callback serialize(Payload :: map()) -> {ok, binary()} | {error, Reason :: term()}.
-callback deserialize(Binary :: binary()) -> {ok, map()} | {error, Reason :: term()}.
-callback schema() -> map().

evoq_hope

Hopes are outbound RPC requests between agents. Unlike facts (fire-and-forget), hopes expect a response.

-behaviour(evoq_hope).

%% Required
-callback hope_type() -> binary().
-callback new(Params :: map()) -> {ok, Hope :: term()} | {error, Reason :: term()}.
-callback to_payload(Hope :: term()) -> map().
-callback from_payload(Payload :: map()) -> {ok, Hope :: term()} | {error, Reason :: term()}.

%% Optional
-callback validate(Hope :: term()) -> ok | {error, Reason :: term()}.

See Artifacts Guide for detailed documentation and examples.

evoq_aggregate

Aggregates maintain business invariants and produce events.

-callback state_module() -> module().
-callback init(AggregateId :: binary()) -> {ok, State :: term()}.
-callback execute(State :: term(), Command :: map()) ->
    {ok, [Event :: map()]} | {error, Reason :: term()}.
-callback apply(State :: term(), Event :: map()) -> NewState :: term().

%% Optional: snapshotting
-callback snapshot(State :: term()) -> SnapshotData :: term().
-callback from_snapshot(SnapshotData :: term()) -> State :: term().

evoq_aggregate_lifespan

Controls aggregate lifecycle (TTL, passivation).

-type action() :: timeout() | infinity | hibernate | stop.

-callback after_event(Event :: map()) -> action().
-callback after_command(Command :: map()) -> action().
-callback after_error(Error :: term()) -> action().
-callback on_timeout(State :: term()) -> {ok, action()} | {snapshot, action()}.

Default: 30-minute idle timeout, snapshot on passivation.

evoq_event_handler

Subscribe to events by type (not by stream).

-callback interested_in() -> [EventType :: binary()].
-callback init(Config :: map()) -> {ok, State :: term()} | {error, Reason :: term()}.
-callback handle_event(EventType :: binary(), Event :: map(),
                       Metadata :: map(), State :: term()) ->
    {ok, NewState :: term()} | {error, Reason :: term()}.
%% Optional: skip for a handler with side effects, see guides/event_handlers.md
-callback replay_policy() -> skip | deliver.

evoq_process_manager

Coordinate long-running business processes.

-callback interested_in() -> [EventType :: binary()].
-callback correlate(Event :: map(), Metadata :: map()) ->
    {start | continue | stop, ProcessId :: binary()} | false.
-callback handle(State :: term(), Event :: map(), Metadata :: map()) ->
    {ok, NewState :: term()} |
    {ok, NewState :: term(), Commands :: [#evoq_command{}]} |
    {error, Reason :: term()}.
-callback apply(State :: term(), Event :: map()) -> NewState :: term().

%% Optional: compensation for a failed command
-callback compensate(State :: term(), FailedCommand :: #evoq_command{}) ->
    {ok, [CompensatingCommand :: #evoq_command{}]} | skip.

evoq_projection

Build read models from events.

-callback interested_in() -> [EventType :: binary()].
-callback init(Config :: map()) ->
    {ok, State :: term(), ReadModel :: evoq_read_model:read_model()} |
    {error, Reason :: term()}.
-callback project(Event :: map(), Metadata :: map(),
                  State :: term(), ReadModel :: evoq_read_model:read_model()) ->
    {ok, NewState :: term(), NewReadModel :: evoq_read_model:read_model()} |
    {skip, State :: term(), ReadModel :: evoq_read_model:read_model()}.

evoq_middleware

Intercept command dispatch.

-callback before_dispatch(Pipeline :: #evoq_pipeline{}) -> #evoq_pipeline{}.
-callback after_dispatch(Pipeline :: #evoq_pipeline{}) -> #evoq_pipeline{}.
-callback after_failure(Pipeline :: #evoq_pipeline{}) -> #evoq_pipeline{}.

Configuration

%% sys.config
[{evoq, [
    {store_id, my_store},

    {aggregate_defaults, #{
        idle_timeout => 1800000,    %% 30 minutes
        hibernate_after => 60000,   %% 1 minute
        snapshot_every => 100       %% events
    }},

    {aggregate_partitions, 4},

    {memory_monitor, #{
        check_interval => 10000,    %% 10 seconds
        elevated_threshold => 0.70,
        critical_threshold => 0.85
    }}
]}].

Consistency is a dispatch option, not application config and not a handler option: evoq_router:dispatch(Command, #{consistency => strong}) waits for every event handler and projection subscribed to the events the command appended to have written them, and #{consistency => {handlers, [...]}} waits for those alone. Only a write is acknowledged: a handler whose own error policy dropped the event answers the dispatch {error, {consistency_dropped, Handler}} instead. See the consistency section of guides/event_handlers.md for what is and is not in the wait.

Memory Pressure Handling

The memory monitor adjusts aggregate TTLs based on system memory usage:

Pressure Level Memory Usage TTL Factor
normal < 70% 1.0x
elevated 70-85% 0.5x
critical > 85% 0.1x

Telemetry Events

All events follow the pattern: [evoq, component, action, stage]

Units follow telemetry's convention (since 1.26.1): a duration is elapsed time from the monotonic clock in native units, and a system_time is wall-clock time in native units. Convert with erlang:convert_time_unit(D, native, millisecond). The one exception is delay on [evoq, handler, retry], which is milliseconds, the unit of the retry backoff.

Aggregate Events

Handler Events

Process Manager Events

Projection Events

Testing

# Unit tests
rebar3 eunit --dir=test/unit

# Integration tests
rebar3 ct

# Dialyzer
rebar3 dialyzer

# All tests with coverage
rebar3 do eunit, ct, cover

Key Design Decisions

Per-Event-Type Subscriptions

Unlike stream-based subscriptions, evoq subscribes by event type. This prevents subscription explosion when you have millions of aggregates.

Default TTL (Not Infinity!)

Aggregates have a 30-minute default idle timeout. This prevents unbounded memory growth that occurs with infinite lifespan defaults.

Partitioned Supervision

Aggregates are distributed across 4 partition supervisors using consistent hashing, preventing single-supervisor bottlenecks.

Documentation

The full, grouped guide index with suggested reading orders lives in guides/README.md. Highlights:

Reckon stack

evoq is one library in the Reckon event-sourcing ecosystem. In dependency order (a library only knows about the ones above it):

License

Apache-2.0