reckon_evoq_adapter (reckon_evoq v1.1.4)

Gateway adapter implementation for evoq

Implements the adapter behaviors using esdb_gater_api to route all operations through the reckon-gater load balancer.

This adapter ensures that evoq never directly calls reckon-db modules, instead routing through the gateway for: - Automatic retry with exponential backoff - Load balancing across workers - High availability

Summary

Types

event/0

evoq_event/0

evoq_snapshot/0

evoq_subscription/0

evoq_subscription_type/0

Functions

ack(StoreId, SubscriptionName, StreamId, Position)

Acknowledge event via gateway.

append(StoreId, StreamId, ExpectedVersion, Events)

Append events to a stream via gateway.

delete(StoreId, StreamId)

Delete all snapshots via gateway.

delete_at_version(StoreId, StreamId, Version)

Delete snapshot at version via gateway.

delete_stream(StoreId, StreamId)

Delete a stream via gateway.

exists(StoreId, StreamId)

Check if stream exists via gateway.

get_by_name(StoreId, SubscriptionName)

Get subscription by name via gateway.

get_checkpoint(StoreId, SubscriptionName)

Get checkpoint for subscription via gateway.

list(StoreId)

List subscriptions via gateway.

list_streams(StoreId)

List all streams via gateway.

list_versions(StoreId, StreamId)

List snapshot versions via gateway.

read(StoreId, StreamId)

Read the latest snapshot via gateway.

read(StoreId, StreamId, StartVersion, Count, Direction)

Read events from a stream via gateway.

read_all(StoreId, StreamId, Direction)

Read all events from a stream via gateway.

read_at_version(StoreId, StreamId, Version)

Read snapshot at specific version via gateway.

read_by_event_types(StoreId, EventTypes, BatchSize)

Read events by type via gateway.

read_by_tags(StoreId, Tags, BatchSize)

Read events by tags via gateway (default: ANY match).

read_by_tags(StoreId, Tags, Match, BatchSize)

Read events by tags via gateway with match mode.

save(StoreId, StreamId, Version, Data, Metadata)

Save a snapshot via gateway.

subscribe(StoreId, Type, Selector, SubscriptionName, Opts)

Subscribe to events via gateway.

unsubscribe(StoreId, SubscriptionId)

Unsubscribe from events via gateway.

version(StoreId, StreamId)

Get current stream version via gateway.

Types

event/0

-type event() ::
          #event{event_id :: binary(),
                 event_type :: binary(),
                 stream_id :: binary(),
                 version :: non_neg_integer(),
                 data :: map() | binary(),
                 metadata :: map(),
                 tags :: [binary()] | undefined,
                 timestamp :: integer(),
                 epoch_us :: integer(),
                 data_content_type :: binary(),
                 metadata_content_type :: binary()}.

evoq_event/0

-type evoq_event() ::
          #evoq_event{event_id :: binary(),
                      event_type :: binary(),
                      stream_id :: binary(),
                      version :: non_neg_integer(),
                      data :: map() | binary(),
                      metadata :: map(),
                      tags :: [binary()] | undefined,
                      timestamp :: integer(),
                      epoch_us :: integer(),
                      data_content_type :: binary(),
                      metadata_content_type :: binary()}.

evoq_snapshot/0

-type evoq_snapshot() ::
          #evoq_snapshot{stream_id :: binary(),
                         version :: non_neg_integer(),
                         data :: map() | binary(),
                         metadata :: map(),
                         timestamp :: integer()}.

evoq_subscription/0

-type evoq_subscription() ::
          #evoq_subscription{id :: binary(),
                             type :: evoq_subscription_type(),
                             selector :: binary() | map(),
                             subscription_name :: binary(),
                             subscriber_pid :: pid() | undefined,
                             created_at :: integer(),
                             pool_size :: pos_integer(),
                             checkpoint :: non_neg_integer() | undefined,
                             options :: map()}.

evoq_subscription_type/0

-type evoq_subscription_type() :: stream | event_type | event_pattern | event_payload | tags.

Functions

ack(StoreId, SubscriptionName, StreamId, Position)

-spec ack(atom(), binary(), binary() | undefined, non_neg_integer()) -> ok | {error, term()}.

Acknowledge event via gateway.

append(StoreId, StreamId, ExpectedVersion, Events)

-spec append(atom(), binary(), integer(), [map()]) -> {ok, non_neg_integer()} | {error, term()}.

Append events to a stream via gateway.

Events from evoq have a flat structure, but reckon_db expects events with a nested data field. This function transforms the events to the format expected by reckon_db.

delete(StoreId, StreamId)

-spec delete(atom(), binary()) -> ok | {error, term()}.

Delete all snapshots via gateway.

delete_at_version(StoreId, StreamId, Version)

-spec delete_at_version(atom(), binary(), non_neg_integer()) -> ok | {error, term()}.

Delete snapshot at version via gateway.

delete_stream(StoreId, StreamId)

-spec delete_stream(atom(), binary()) -> ok | {error, term()}.

Delete a stream via gateway.

exists(StoreId, StreamId)

-spec exists(atom(), binary()) -> boolean().

Check if stream exists via gateway.

get_by_name(StoreId, SubscriptionName)

-spec get_by_name(atom(), binary()) -> {ok, evoq_subscription()} | {error, not_found | term()}.

Get subscription by name via gateway.

get_checkpoint(StoreId, SubscriptionName)

-spec get_checkpoint(atom(), binary()) -> {ok, non_neg_integer()} | {error, not_found | term()}.

Get checkpoint for subscription via gateway.

list(StoreId)

-spec list(atom()) -> {ok, [evoq_subscription()]} | {error, term()}.

List subscriptions via gateway.

list_streams(StoreId)

-spec list_streams(atom()) -> {ok, [binary()]} | {error, term()}.

List all streams via gateway.

list_versions(StoreId, StreamId)

-spec list_versions(atom(), binary()) -> {ok, [non_neg_integer()]} | {error, term()}.

List snapshot versions via gateway.

read(StoreId, StreamId)

-spec read(atom(), binary()) -> {ok, evoq_snapshot()} | {error, not_found | term()}.

Read the latest snapshot via gateway.

read(StoreId, StreamId, StartVersion, Count, Direction)

-spec read(atom(), binary(), non_neg_integer(), pos_integer(), forward | backward) ->
              {ok, [evoq_event()]} | {error, term()}.

Read events from a stream via gateway.

For event sourcing, a non-existent stream just means no events yet, so we translate stream_not_found to an empty list.

read_all(StoreId, StreamId, Direction)

-spec read_all(atom(), binary(), forward | backward) -> {ok, [evoq_event()]} | {error, term()}.

Read all events from a stream via gateway.

read_at_version(StoreId, StreamId, Version)

-spec read_at_version(atom(), binary(), non_neg_integer()) ->
                         {ok, evoq_snapshot()} | {error, not_found | term()}.

Read snapshot at specific version via gateway.

read_by_event_types(StoreId, EventTypes, BatchSize)

-spec read_by_event_types(atom(), [binary()], pos_integer()) -> {ok, [evoq_event()]} | {error, term()}.

Read events by type via gateway.

Uses the server-side native Khepri filtering for efficient type-based queries. Events are filtered at the database level, avoiding loading all events into memory.

read_by_tags(StoreId, Tags, BatchSize)

-spec read_by_tags(atom(), [binary()], pos_integer()) -> {ok, [evoq_event()]} | {error, term()}.

Read events by tags via gateway (default: ANY match).

Tags provide cross-stream querying for the process-centric model. Use this to find all events related to specific participants.

read_by_tags(StoreId, Tags, Match, BatchSize)

-spec read_by_tags(atom(), [binary()], any | all, pos_integer()) ->
                      {ok, [evoq_event()]} | {error, term()}.

Read events by tags via gateway with match mode.

Tags provide cross-stream querying for the process-centric model. Events are filtered at the database level.

Match modes: any - Return events matching ANY of the tags (union) all - Return events matching ALL of the tags (intersection)

save(StoreId, StreamId, Version, Data, Metadata)

-spec save(atom(), binary(), non_neg_integer(), map() | binary(), map()) -> ok | {error, term()}.

Save a snapshot via gateway.

unsubscribe(StoreId, SubscriptionId)

-spec unsubscribe(atom(), binary()) -> ok | {error, term()}.

Unsubscribe from events via gateway.

version(StoreId, StreamId)

-spec version(atom(), binary()) -> integer().

Get current stream version via gateway.