reckon-evoq

Hex.pm Hex Docs GitHub Sponsors

Adapter for connecting the evoq CQRS/ES framework to a reckon-db event store, reached through the reckon-gater gateway API.

reckon-evoq depends on evoq and reckon_gater only. It does not depend on reckon_db: it reaches the store through the gater API (see Reckon stack).

Overview

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

Every callback is a thin translation: it maps evoq-shaped terms to the gater types, calls reckon_gater_api, and maps the result back through events_to_evoq/1 and friends. The adapter holds no state and runs no logic of its own.

The DCB passthrough is what backs evoq's evoq_decision behaviour, see evoq's decisions guide for the high-level pattern. The CCC callbacks extend that to payload-conditioned decisions: declared {payload, Key} / {payload_hash, [Keys]} indexes drive conditional reads, and the introspection callbacks let evoq's decision runtime fail loudly on an undeclared payload index.

All operations are routed through reckon-gater, which provides:

Installation

Add to your rebar.config:

{deps, [

    {reckon_evoq, "~> 2.7"}

]}.

Versions

Component Version
reckon_evoq (this repo) 2.7.2
evoq (dep) ~> 1.22
reckon_gater (dep) ~> 3.7
telemetry (dep) ~> 1.3
Erlang/OTP 27+

reckon-evoq depends on evoq and reckon_gater directly. The underlying reckon_db event store must be running and reachable through the gateway, but is not a build dependency of this adapter.

Dependencies

This adapter requires:

Quick Start

1. Configure evoq to use this adapter

In your sys.config:

{evoq, [

    {event_store_adapter, reckon_evoq_adapter},

    {snapshot_store_adapter, reckon_evoq_adapter},

    {subscription_adapter, reckon_evoq_adapter}

]}

2. Ensure reckon-db is running

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

3. Use 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

reckon_evoq_adapter:append(StoreId, StreamId, ExpectedVersion, Events).

%% Returns: {ok, NewVersion} | {error, Reason}



%% Read events from a stream

reckon_evoq_adapter:read(StoreId, StreamId, StartVersion, Count, Direction).

%% Direction: forward | backward

%% Returns: {ok, [Event]} | {error, Reason}



%% Read all events from a stream (batched internally)

reckon_evoq_adapter:read_all(StoreId, StreamId, Direction).

%% Returns: {ok, [Event]} | {error, Reason}



%% Read events by type (server-side Khepri filtering)

reckon_evoq_adapter:read_by_event_types(StoreId, EventTypes, BatchSize).

%% Returns: {ok, [Event]} | {error, Reason}



%% Get stream version

reckon_evoq_adapter:version(StoreId, StreamId).

%% Returns: Version (integer) | -1 (no stream)



%% Check if stream exists

reckon_evoq_adapter:exists(StoreId, StreamId).

%% Returns: boolean()



%% List all streams

reckon_evoq_adapter:list_streams(StoreId).

%% Returns: {ok, [StreamId]} | {error, Reason}



%% Delete a stream

reckon_evoq_adapter:delete_stream(StoreId, StreamId).

%% Returns: ok | {error, Reason}

Conditional Append, DCB and CCC

%% DCB conditional append: append only if no event matches TagFilter past SeqCutoff

reckon_evoq_adapter:append_if_no_tag_matches(StoreId, TagFilter, SeqCutoff, Events).

%% TagFilter: reckon_gater_types:tag_filter()  SeqCutoff: integer() (-1 = saw nothing)

%% Returns: {ok, LastSeq} | {error, {context_changed, MaxSeq}}



%% CCC payload-conditioned reads (2.7.0+, evoq 1.22 / reckon-gater 3.7)

reckon_evoq_adapter:ccc_read_by_payload(StoreId, Key, Value, BatchSize).

reckon_evoq_adapter:ccc_read_by_payload_hash(StoreId, Keys, Values, BatchSize).

%% Keys/Values: equal-length [binary()]  BatchSize: pos_integer()

%% Returns: {ok, [Event]} | {error, Reason}



%% CCC index introspection (used by evoq's decision runtime)

reckon_evoq_adapter:payload_indexes(StoreId).

reckon_evoq_adapter:payload_hash_indexes(StoreId).

%% Returns: the declared payload / payload-hash index key lists

Cross-Cutting Queries

%% Read events by tags across all streams

reckon_evoq_adapter:read_by_tags(StoreId, Tags, BatchSize).

reckon_evoq_adapter:read_by_tags(StoreId, Tags, Match, BatchSize).

%% Tags: [binary()]  Match: any | all  BatchSize: pos_integer()

%% Returns: {ok, [Event]} | {error, Reason}



%% Read events whose metadata Key equals Value

reckon_evoq_adapter:read_by_metadata(StoreId, Key, Value).

%% Returns: {ok, [Event]} | {error, Reason}

Snapshot Operations

%% Save a snapshot

reckon_evoq_adapter:save(StoreId, StreamId, Version, Data, Metadata).

%% Returns: ok | {error, Reason}



%% Read latest snapshot

reckon_evoq_adapter:read(StoreId, StreamId).

%% Returns: {ok, #evoq_snapshot{}} | {error, not_found}



%% Read snapshot at specific version

reckon_evoq_adapter:read_at_version(StoreId, StreamId, Version).

%% Returns: {ok, #evoq_snapshot{}} | {error, not_found}

%%

%% reckon-db 5.5.2+ stores a snapshot's data and metadata in their own

%% fields; older stores (and rows written by them) keep the whole

%% #{data, metadata, timestamp} wrapper in data, and are read through that

%% shape. Use 2.7.2 or later with any reckon-db (2.7.1 was never published).



%% Delete all snapshots for stream

reckon_evoq_adapter:delete(StoreId, StreamId).

%% Returns: ok



%% Delete snapshot at specific version

reckon_evoq_adapter:delete_at_version(StoreId, StreamId, Version).

%% Returns: ok | {error, Reason}



%% List snapshot versions

reckon_evoq_adapter:list_versions(StoreId, StreamId).

%% Returns: {ok, [Version]} | {error, Reason}

Subscription Operations

%% Subscribe to events

reckon_evoq_adapter:subscribe(StoreId, Type, Selector, Name, Opts).

%% Type: stream | event_type | event_pattern | event_payload

%% Returns: {ok, SubscriptionId} | {error, Reason}



%% Unsubscribe

reckon_evoq_adapter:unsubscribe(StoreId, SubscriptionId).

%% Returns: ok | {error, Reason}



%% Acknowledge event

reckon_evoq_adapter:ack(StoreId, SubscriptionName, StreamId, Position).

%% Returns: ok



%% Get checkpoint (subscription position)

reckon_evoq_adapter:get_checkpoint(StoreId, SubscriptionName).

%% Returns: {ok, Position} | {error, not_found}



%% List all subscriptions

reckon_evoq_adapter:list(StoreId).

%% Returns: {ok, [#subscription{}]} | {error, Reason}



%% Get subscription by name

reckon_evoq_adapter:get_by_name(StoreId, SubscriptionName).

%% Returns: {ok, #subscription{}} | {error, not_found}

Checkpoint Store

%% Configure which ReckonDB store holds projection checkpoints

application:set_env(reckon_evoq, checkpoint_store_id, my_store).



%% Use with evoq_projection

evoq_projection:start_link(MyProjection, Config, #{

    checkpoint_store => reckon_evoq_checkpoint_store

}).



%% Checkpoints persist as ReckonDB snapshots.

%% On restart, projections resume from last checkpoint

%% instead of replaying all events.

Architecture

reckon-evoq Architecture

Local Development

For local development with symlinks:

mkdir -p _checkouts

ln -s /path/to/evoq _checkouts/evoq

ln -s /path/to/reckon-gater _checkouts/reckon_gater

Then compile:

rebar3 compile

rebar3 eunit

Retry Behavior

The adapter inherits retry behavior from reckon-gater:

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

Failover Behavior

During leader failover in reckon-db:

  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

Reckon stack

reckon-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