erl-evoq-esdb

Adapter for connecting erl-evoq CQRS/ES framework to erl-esdb event store via erl-esdb-gater gateway.

Overview

erl-evoq-esdb is a thin adapter layer that implements the erl-evoq behavior interfaces:

• evoq_adapter - Event store operations (append, read, delete)
• evoq_snapshot_adapter - Snapshot operations (save, read, delete)
• evoq_subscription_adapter - Subscription operations (subscribe, ack, checkpoint)

All operations are routed through erl-esdb-gater, which provides:

• Automatic retry with exponential backoff
• Load balancing across gateway workers
• High availability with failover support

Installation

Add to your rebar.config:

{deps, [
    {erl_evoq_esdb, "~> 0.3.0"}
]}.

Dependencies

This adapter requires:

• erl-evoq >= 0.3.0 - CQRS/ES framework with behavior definitions
• erl-esdb-gater >= 0.6.2 - Gateway API for load balancing and retry
• erl-esdb >= 0.4.3 - The underlying event store (must be running)

Quick Start

1. Configure erl-evoq to use this adapter

In your sys.config:

{erl_evoq, [
    {adapter, evoq_esdb_gater_adapter},
    {snapshot_adapter, evoq_esdb_gater_adapter},
    {subscription_adapter, evoq_esdb_gater_adapter}
]}

2. Ensure erl-esdb is running

The adapter requires an erl-esdb cluster to be available. Gateway workers automatically discover and connect to erl-esdb nodes.

3. Use erl-evoq normally

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

%% Dispatch through evoq (uses this adapter internally)
ok = evoq_dispatcher:dispatch(Command).

API Reference

Event Store Operations

%% Append events to a stream
evoq_esdb_gater_adapter:append(StoreId, StreamId, ExpectedVersion, Events).
%% Returns: {ok, NewVersion} | {error, Reason}

%% Read events from a stream
evoq_esdb_gater_adapter:read(StoreId, StreamId, StartVersion, Count, Direction).
%% Direction: forward | backward
%% Returns: {ok, [Event]} | {error, Reason}

%% Read all events from a stream (batched internally)
evoq_esdb_gater_adapter:read_all(StoreId, StreamId, Direction).
%% Returns: {ok, [Event]} | {error, Reason}

%% Read events by type (server-side Khepri filtering)
evoq_esdb_gater_adapter:read_by_event_types(StoreId, EventTypes, BatchSize).
%% Returns: {ok, [Event]} | {error, Reason}

%% Get stream version
evoq_esdb_gater_adapter:version(StoreId, StreamId).
%% Returns: Version (integer) | -1 (no stream)

%% Check if stream exists
evoq_esdb_gater_adapter:exists(StoreId, StreamId).
%% Returns: boolean()

%% List all streams
evoq_esdb_gater_adapter:list_streams(StoreId).
%% Returns: {ok, [StreamId]} | {error, Reason}

%% Delete a stream
evoq_esdb_gater_adapter:delete_stream(StoreId, StreamId).
%% Returns: ok | {error, Reason}

Snapshot Operations

%% Save a snapshot
evoq_esdb_gater_adapter:save(StoreId, StreamId, Version, Data, Metadata).
%% Returns: ok | {error, Reason}

%% Read latest snapshot
evoq_esdb_gater_adapter:read(StoreId, StreamId).
%% Returns: {ok, #snapshot{}} | {error, not_found}

%% Read snapshot at specific version
evoq_esdb_gater_adapter:read_at_version(StoreId, StreamId, Version).
%% Returns: {ok, #snapshot{}} | {error, not_found}

%% Delete all snapshots for stream
evoq_esdb_gater_adapter:delete(StoreId, StreamId).
%% Returns: ok

%% Delete snapshot at specific version
evoq_esdb_gater_adapter:delete_at_version(StoreId, StreamId, Version).
%% Returns: ok | {error, Reason}

%% List snapshot versions
evoq_esdb_gater_adapter:list_versions(StoreId, StreamId).
%% Returns: {ok, [Version]} | {error, Reason}

Subscription Operations

%% Subscribe to events
evoq_esdb_gater_adapter:subscribe(StoreId, Type, Selector, Name, Opts).
%% Type: stream | event_type | event_pattern | event_payload
%% Returns: {ok, SubscriptionId} | {error, Reason}

%% Unsubscribe
evoq_esdb_gater_adapter:unsubscribe(StoreId, SubscriptionId).
%% Returns: ok | {error, Reason}

%% Acknowledge event
evoq_esdb_gater_adapter:ack(StoreId, SubscriptionName, StreamId, Position).
%% Returns: ok

%% Get checkpoint (subscription position)
evoq_esdb_gater_adapter:get_checkpoint(StoreId, SubscriptionName).
%% Returns: {ok, Position} | {error, not_found}

%% List all subscriptions
evoq_esdb_gater_adapter:list(StoreId).
%% Returns: {ok, [#subscription{}]} | {error, Reason}

%% Get subscription by name
evoq_esdb_gater_adapter:get_by_name(StoreId, SubscriptionName).
%% Returns: {ok, #subscription{}} | {error, not_found}

Architecture

+-------------+     +------------------+     +----------------+     +-----------+
|  erl-evoq   | --> | erl-evoq-esdb    | --> | erl-esdb-gater | --> | erl-esdb  |
| (Framework) |     | (This Adapter)   |     | (Gateway/LB)   |     | (Storage) |
+-------------+     +------------------+     +----------------+     +-----------+
                    Implements behaviors    Load balancing,        Khepri/Ra
                    from erl-evoq           retry, failover        Raft consensus

Local Development

For local development with symlinks:

mkdir -p _checkouts
ln -s /path/to/erl-evoq _checkouts/erl_evoq
ln -s /path/to/erl-esdb-gater _checkouts/erl_esdb_gater

Then compile:

rebar3 compile
rebar3 eunit

Retry Behavior

The adapter inherits retry behavior from erl-esdb-gater:

• Base delay: 100ms
• Max delay: 30 seconds
• Max retries: 10
• Backoff: Exponential with jitter

Failed operations are automatically retried. If all retries fail, the original error is returned.

Failover Behavior

During leader failover in erl-esdb:

1. Gateway detects worker unavailability
2. Requests are routed to healthy workers
3. In-flight writes are retried automatically
4. Reads may return stale data briefly (eventual consistency)

Version History

See CHANGELOG.md for version history.

Related Projects

• erl-evoq - CQRS/ES framework
• erl-esdb - BEAM-native Event Store
• erl-esdb-gater - Gateway API

License

Apache-2.0