%% @doc Facade for subscription operations via adapter. %% %% Provides a consistent interface for subscribing to events, %% delegating to a configured subscription adapter. %% %% The adapter is responsible for translating between the event %% store's native format and evoq's #evoq_event{} records. %% Subscribers always receive {events, [#evoq_event{}]} messages. %% %% == Configuration (Required) == %% %% You must configure a subscription adapter in your application config: %% %% ``` %% {evoq, [ %% {subscription_adapter, reckon_evoq_adapter} %% ]} %% ''' %% %% == Usage == %% %% ``` %% %% Subscribe to events (typically in a gen_server init) %% {ok, SubId} = evoq_subscriptions:subscribe( %% my_store, event_type, <<"order_placed_v1">>, %% <<"my_projection">>, #{subscriber_pid => self()} %% ), %% %% %% Receive events in handle_info %% handle_info({events, Events}, State) -> %% lists:foreach(fun(#evoq_event{data = Data}) -> %% project(Data) %% end, Events), %% {noreply, State}. %% ''' %% %% @author rgfaber -module(evoq_subscriptions). -include_lib("evoq/include/evoq_types.hrl"). %% API -export([subscribe/5, unsubscribe/2]). -export([ack/4, get_checkpoint/2]). -export([list/1, get_by_name/2]). %% Configuration -export([get_adapter/0, set_adapter/1]). %%==================================================================== %% Configuration %%==================================================================== %% @doc Get the configured subscription adapter. %% Crashes if no adapter is configured. -spec get_adapter() -> module(). get_adapter() -> case application:get_env(evoq, subscription_adapter) of {ok, Adapter} -> Adapter; undefined -> error({not_configured, subscription_adapter}) end. %% @doc Set the subscription adapter (primarily for testing). -spec set_adapter(module()) -> ok. set_adapter(Adapter) -> application:set_env(evoq, subscription_adapter, Adapter). %%==================================================================== %% API %%==================================================================== %% @doc Subscribe to events from a store. %% %% The adapter guarantees that the subscriber_pid receives %% `{events, [#evoq_event{}]}' messages with proper envelope %% structure (event_type, stream_id, version, data, metadata). %% %% Options: %% subscriber_pid - PID to receive events (required for push delivery) %% start_from - Starting position (default: 0) %% pool_size - Number of emitters (default: 1) -spec subscribe(atom(), evoq_subscription_type(), binary() | map(), binary(), map()) -> {ok, binary()} | {error, term()}. subscribe(StoreId, Type, Selector, SubscriptionName, Opts) -> Adapter = get_adapter(), Adapter:subscribe(StoreId, Type, Selector, SubscriptionName, Opts). %% @doc Unsubscribe from events. -spec unsubscribe(atom(), binary()) -> ok | {error, term()}. unsubscribe(StoreId, SubscriptionId) -> Adapter = get_adapter(), Adapter:unsubscribe(StoreId, SubscriptionId). %% @doc Acknowledge an event has been processed. %% Updates the subscription checkpoint. -spec ack(atom(), binary(), binary() | undefined, non_neg_integer()) -> ok | {error, term()}. ack(StoreId, SubscriptionName, StreamId, Position) -> Adapter = get_adapter(), Adapter:ack(StoreId, SubscriptionName, StreamId, Position). %% @doc Get the current checkpoint for a subscription. -spec get_checkpoint(atom(), binary()) -> {ok, non_neg_integer()} | {error, not_found | term()}. get_checkpoint(StoreId, SubscriptionName) -> Adapter = get_adapter(), Adapter:get_checkpoint(StoreId, SubscriptionName). %% @doc List all subscriptions for a store. -spec list(atom()) -> {ok, [evoq_subscription()]} | {error, term()}. list(StoreId) -> Adapter = get_adapter(), Adapter:list(StoreId). %% @doc Get a subscription by name. -spec get_by_name(atom(), binary()) -> {ok, evoq_subscription()} | {error, not_found | term()}. get_by_name(StoreId, SubscriptionName) -> Adapter = get_adapter(), Adapter:get_by_name(StoreId, SubscriptionName).