%% @doc Main API for reckon-gater %% %% Provides the primary interface for accessing reckon-db event stores %% through the gateway with automatic load balancing and retry. %% %% This module mirrors the ExESDBGater.API pattern from the original %% Elixir implementation, providing specific functions for each operation %% rather than a generic call interface. %% %% == Stream Operations == %% %% {ok, Version} = esdb_gater_api:append_events(my_store, StreamId, Events). %% {ok, Events} = esdb_gater_api:get_events(my_store, StreamId, 0, 100, forward). %% {ok, Version} = esdb_gater_api:get_version(my_store, StreamId). %% %% == Subscription Operations == %% %% ok = esdb_gater_api:save_subscription(my_store, by_stream, Selector, Name). %% ok = esdb_gater_api:remove_subscription(my_store, by_stream, Selector, Name). %% %% == Snapshot Operations == %% %% ok = esdb_gater_api:record_snapshot(my_store, SourceId, StreamId, Version, Data). %% {ok, Snapshot} = esdb_gater_api:read_snapshot(my_store, SourceId, StreamId, Version). %% %% @author rgfaber -module(esdb_gater_api). -include("esdb_gater.hrl"). -include("esdb_gater_telemetry.hrl"). %% Worker registration -export([ register_worker/1, register_worker/2, unregister_worker/1, unregister_worker/2, get_workers/1 ]). %% Stream operations -export([ append_events/3, append_events/4, get_events/5, stream_forward/4, stream_backward/4, get_version/2, get_streams/1, delete_stream/2, read_by_event_types/3, read_by_tags/2, read_by_tags/3, read_all_global/3 ]). %% Subscription operations -export([ get_subscriptions/1, get_subscription/2, save_subscription/6, remove_subscription/4, ack_event/4 ]). %% Snapshot operations -export([ record_snapshot/5, delete_snapshot/4, read_snapshot/4, list_snapshots/3 ]). %% Store operations -export([ list_stores/0 ]). %% Health and diagnostics -export([ health/0, verify_cluster_consistency/1, quick_health_check/1, verify_membership_consensus/1, check_raft_log_consistency/1 ]). %% Temporal query operations -export([ read_until/3, read_until/4, read_range/4, read_range/5, version_at/3 ]). %% Scavenge operations -export([ scavenge/3, scavenge_matching/3, scavenge_dry_run/3 ]). %% Causation operations -export([ get_effects/2, get_cause/2, get_causation_chain/2, get_correlated/2, build_causation_graph/2 ]). %% Schema operations -export([ get_schema/2, list_schemas/1, get_schema_version/2, upcast_events/2, register_schema/3, unregister_schema/2 ]). %% Memory pressure operations -export([ get_memory_level/1, get_memory_stats/1 ]). %% Link operations -export([ create_link/2, delete_link/2, get_link/2, list_links/1, start_link/2, stop_link/2, link_info/2 ]). -define(CALL_TIMEOUT, 5000). %%==================================================================== %% Worker Registration %%==================================================================== %% @doc Register current process as a worker for a store -spec register_worker(atom()) -> ok | {error, term()}. register_worker(StoreId) -> register_worker(StoreId, self()). %% @doc Register a specific process as a worker for a store -spec register_worker(atom(), pid()) -> ok | {error, term()}. register_worker(StoreId, Pid) -> esdb_gater_worker_registry:register_worker(StoreId, Pid). %% @doc Unregister current process as a worker for a store -spec unregister_worker(atom()) -> ok | {error, term()}. unregister_worker(StoreId) -> unregister_worker(StoreId, self()). %% @doc Unregister a specific process as a worker for a store -spec unregister_worker(atom(), pid()) -> ok | {error, term()}. unregister_worker(StoreId, Pid) -> esdb_gater_worker_registry:unregister_worker(StoreId, Pid). %% @doc Get all registered workers for a store -spec get_workers(atom()) -> {ok, [worker_entry()]} | {error, term()}. get_workers(StoreId) -> esdb_gater_worker_registry:get_workers(StoreId). %%==================================================================== %% Stream Operations %%==================================================================== %% @doc Append events to a stream (auto-versioned) -spec append_events(atom(), binary(), list()) -> {ok, integer()} | {error, term()}. append_events(StoreId, StreamId, Events) -> route_call(StoreId, {append_events, StoreId, StreamId, Events}). %% @doc Append events to a stream with expected version -spec append_events(atom(), binary(), integer() | any, list()) -> {ok, integer()} | {error, term()} | {error, {wrong_expected_version, integer()}}. append_events(StoreId, StreamId, ExpectedVersion, Events) -> route_call(StoreId, {append_events, StoreId, StreamId, ExpectedVersion, Events}). %% @doc Get events from a stream -spec get_events(atom(), binary(), integer(), integer(), forward | backward) -> {ok, list()} | {error, term()}. get_events(StoreId, StreamId, StartVersion, Count, Direction) -> route_call(StoreId, {get_events, StoreId, StreamId, StartVersion, Count, Direction}). %% @doc Stream events forward from a version -spec stream_forward(atom(), binary(), integer(), integer()) -> {ok, list()} | {error, term()}. stream_forward(StoreId, StreamId, StartVersion, Count) -> route_call(StoreId, {stream_forward, StoreId, StreamId, StartVersion, Count}). %% @doc Stream events backward from a version -spec stream_backward(atom(), binary(), integer(), non_neg_integer()) -> {ok, list()} | {error, term()}. stream_backward(StoreId, StreamId, StartVersion, Count) -> route_call(StoreId, {stream_backward, StoreId, StreamId, StartVersion, Count}). %% @doc Get the current version of a stream -spec get_version(atom(), binary()) -> {ok, integer()} | {error, term()}. get_version(StoreId, StreamId) -> route_call(StoreId, {get_version, StoreId, StreamId}). %% @doc Get all streams in a store -spec get_streams(atom()) -> {ok, list()} | {error, term()}. get_streams(StoreId) -> route_call(StoreId, {get_streams, StoreId}). %% @doc Delete a stream and all its events -spec delete_stream(atom(), binary()) -> ok | {error, term()}. delete_stream(StoreId, StreamId) -> route_call(StoreId, {delete_stream, StoreId, StreamId}). %% @doc Read events by type using native Khepri filtering %% %% This uses the server-side read_by_event_types which performs efficient %% filtering at the database level rather than loading all events. -spec read_by_event_types(atom(), [binary()], pos_integer()) -> {ok, list()} | {error, term()}. read_by_event_types(StoreId, EventTypes, BatchSize) -> route_call(StoreId, {read_by_event_types, StoreId, EventTypes, BatchSize}). %% @doc Read events by tags (default: ANY match, batch_size 1000). %% %% Queries events across all streams that have matching tags. %% By default, returns events matching ANY of the provided tags (union). -spec read_by_tags(atom(), [binary()]) -> {ok, list()} | {error, term()}. read_by_tags(StoreId, Tags) -> read_by_tags(StoreId, Tags, #{}). %% @doc Read events by tags with options. %% %% Queries events across all streams that have matching tags. %% Tags are typically used for cross-stream querying in the process-centric model. %% %% Options: %% - match: any (default) returns events matching ANY tag (union), %% all returns events matching ALL tags (intersection) %% - batch_size: Maximum events to return (default 1000) %% %% @param StoreId The store identifier %% @param Tags List of tag binaries to match %% @param Opts Options map with match and batch_size keys -spec read_by_tags(atom(), [binary()], map()) -> {ok, list()} | {error, term()}. read_by_tags(StoreId, Tags, Opts) -> Match = maps:get(match, Opts, any), BatchSize = maps:get(batch_size, Opts, 1000), route_call(StoreId, {read_by_tags, StoreId, Tags, Match, BatchSize}). %% @doc Read all events across all streams sorted by epoch_us (global ordering). %% %% This is used for catch-up subscriptions where a consumer needs to replay %% all historical events from a given offset. %% %% Offset is the number of events to skip (not a version number). %% BatchSize controls how many events to return per call. -spec read_all_global(atom(), non_neg_integer(), pos_integer()) -> {ok, list()} | {error, term()}. read_all_global(StoreId, Offset, BatchSize) -> route_call(StoreId, {read_all_global, StoreId, Offset, BatchSize}). %%==================================================================== %% Subscription Operations %%==================================================================== %% @doc Get all subscriptions for a store -spec get_subscriptions(atom()) -> {ok, list()} | {error, term()}. get_subscriptions(StoreId) -> route_call(StoreId, {get_subscriptions, StoreId}). %% @doc Get a specific subscription by name %% %% Returns the subscription details including the checkpoint. -spec get_subscription(atom(), binary()) -> {ok, map()} | {error, term()}. get_subscription(StoreId, SubscriptionName) -> route_call(StoreId, {get_subscription, StoreId, SubscriptionName}). %% @doc Save a subscription -spec save_subscription(atom(), atom(), binary() | map(), binary(), non_neg_integer(), pid() | undefined) -> ok. save_subscription(StoreId, Type, Selector, SubscriptionName, StartFrom, Subscriber) -> route_cast(StoreId, {save_subscription, StoreId, Type, Selector, SubscriptionName, StartFrom, Subscriber}), ok. %% @doc Remove a subscription -spec remove_subscription(atom(), atom(), binary() | map(), binary()) -> ok. remove_subscription(StoreId, Type, Selector, SubscriptionName) -> route_cast(StoreId, {remove_subscription, StoreId, Type, Selector, SubscriptionName}), ok. %% @doc Acknowledge receipt of an event by a subscriber -spec ack_event(atom(), binary(), pid(), map()) -> ok. ack_event(StoreId, SubscriptionName, SubscriberPid, Event) -> route_cast(StoreId, {ack_event, StoreId, SubscriptionName, SubscriberPid, Event}), ok. %%==================================================================== %% Snapshot Operations %%==================================================================== %% @doc Record a snapshot -spec record_snapshot(atom(), binary(), binary(), non_neg_integer(), map()) -> ok. record_snapshot(StoreId, SourceUuid, StreamUuid, Version, SnapshotRecord) -> route_cast(StoreId, {record_snapshot, StoreId, SourceUuid, StreamUuid, Version, SnapshotRecord}), ok. %% @doc Delete a snapshot -spec delete_snapshot(atom(), binary(), binary(), non_neg_integer()) -> ok. delete_snapshot(StoreId, SourceUuid, StreamUuid, Version) -> route_cast(StoreId, {delete_snapshot, StoreId, SourceUuid, StreamUuid, Version}), ok. %% @doc Read a snapshot -spec read_snapshot(atom(), binary(), binary(), non_neg_integer()) -> {ok, map()} | {error, term()}. read_snapshot(StoreId, SourceUuid, StreamUuid, Version) -> route_call(StoreId, {read_snapshot, StoreId, SourceUuid, StreamUuid, Version}). %% @doc List snapshots -spec list_snapshots(atom(), binary() | any, binary() | any) -> {ok, [map()]} | {error, term()}. list_snapshots(StoreId, SourceUuid, StreamUuid) -> route_call(StoreId, {list_snapshots, StoreId, SourceUuid, StreamUuid}). %%==================================================================== %% Store Operations %%==================================================================== %% @doc List all managed stores in the cluster -spec list_stores() -> {ok, list()} | {error, term()}. list_stores() -> %% Need at least one worker to route to case esdb_gater_worker_registry:get_all_workers() of {ok, Workers} when map_size(Workers) > 0 -> %% Pick any store to route the request [StoreId | _] = maps:keys(Workers), route_call(StoreId, {list_stores}); {ok, _} -> {error, no_workers}; {error, _} = Error -> Error end. %%==================================================================== %% Health and Diagnostics %%==================================================================== %% @doc Get gateway health status -spec health() -> {ok, map()}. health() -> case esdb_gater_worker_registry:get_all_workers() of {ok, Workers} -> StoreStats = maps:fold( fun(StoreId, Entries, Acc) -> maps:put(StoreId, length(Entries), Acc) end, #{}, Workers ), {ok, #{ status => healthy, stores => StoreStats, total_workers => maps:fold(fun(_, C, A) -> A + C end, 0, StoreStats), node => node(), timestamp => erlang:system_time(millisecond) }}; {error, Reason} -> {ok, #{ status => degraded, reason => Reason, node => node(), timestamp => erlang:system_time(millisecond) }} end. %% @doc Verify cluster consistency for a store -spec verify_cluster_consistency(atom()) -> {ok, map()} | {error, term()}. verify_cluster_consistency(StoreId) -> route_call(StoreId, {verify_cluster_consistency, StoreId}). %% @doc Quick health check for a store -spec quick_health_check(atom()) -> {ok, map()} | {error, term()}. quick_health_check(StoreId) -> route_call(StoreId, {quick_health_check, StoreId}). %% @doc Verify membership consensus for a store -spec verify_membership_consensus(atom()) -> {ok, map()} | {error, term()}. verify_membership_consensus(StoreId) -> route_call(StoreId, {verify_membership_consensus, StoreId}). %% @doc Check Raft log consistency for a store -spec check_raft_log_consistency(atom()) -> {ok, map()} | {error, term()}. check_raft_log_consistency(StoreId) -> route_call(StoreId, {check_raft_log_consistency, StoreId}). %%==================================================================== %% Temporal Query Operations %%==================================================================== %% @doc Read events up to a timestamp -spec read_until(atom(), binary(), integer()) -> {ok, list()} | {error, term()}. read_until(StoreId, StreamId, Timestamp) -> route_call(StoreId, {read_until, StoreId, StreamId, Timestamp}). %% @doc Read events up to a timestamp with options -spec read_until(atom(), binary(), integer(), map()) -> {ok, list()} | {error, term()}. read_until(StoreId, StreamId, Timestamp, Opts) -> route_call(StoreId, {read_until, StoreId, StreamId, Timestamp, Opts}). %% @doc Read events in a time range -spec read_range(atom(), binary(), integer(), integer()) -> {ok, list()} | {error, term()}. read_range(StoreId, StreamId, FromTimestamp, ToTimestamp) -> route_call(StoreId, {read_range, StoreId, StreamId, FromTimestamp, ToTimestamp}). %% @doc Read events in a time range with options -spec read_range(atom(), binary(), integer(), integer(), map()) -> {ok, list()} | {error, term()}. read_range(StoreId, StreamId, FromTimestamp, ToTimestamp, Opts) -> route_call(StoreId, {read_range, StoreId, StreamId, FromTimestamp, ToTimestamp, Opts}). %% @doc Get stream version at a specific timestamp -spec version_at(atom(), binary(), integer()) -> {ok, integer()} | {error, term()}. version_at(StoreId, StreamId, Timestamp) -> route_call(StoreId, {version_at, StoreId, StreamId, Timestamp}). %%==================================================================== %% Scavenge Operations %%==================================================================== %% @doc Scavenge a stream (delete old events) -spec scavenge(atom(), binary(), map()) -> {ok, map()} | {error, term()}. scavenge(StoreId, StreamId, Opts) -> route_call(StoreId, {scavenge, StoreId, StreamId, Opts}). %% @doc Scavenge streams matching a pattern -spec scavenge_matching(atom(), binary(), map()) -> {ok, list()} | {error, term()}. scavenge_matching(StoreId, Pattern, Opts) -> route_call(StoreId, {scavenge_matching, StoreId, Pattern, Opts}). %% @doc Dry-run scavenge (preview what would be deleted) -spec scavenge_dry_run(atom(), binary(), map()) -> {ok, map()} | {error, term()}. scavenge_dry_run(StoreId, StreamId, Opts) -> route_call(StoreId, {scavenge_dry_run, StoreId, StreamId, Opts}). %%==================================================================== %% Causation Operations %%==================================================================== %% @doc Get events caused by an event -spec get_effects(atom(), binary()) -> {ok, list()} | {error, term()}. get_effects(StoreId, EventId) -> route_call(StoreId, {get_effects, StoreId, EventId}). %% @doc Get the event that caused another -spec get_cause(atom(), binary()) -> {ok, map()} | {error, term()}. get_cause(StoreId, EventId) -> route_call(StoreId, {get_cause, StoreId, EventId}). %% @doc Get the full causation chain for an event -spec get_causation_chain(atom(), binary()) -> {ok, list()} | {error, term()}. get_causation_chain(StoreId, EventId) -> route_call(StoreId, {get_causation_chain, StoreId, EventId}). %% @doc Get all events with the same correlation ID -spec get_correlated(atom(), binary()) -> {ok, list()} | {error, term()}. get_correlated(StoreId, CorrelationId) -> route_call(StoreId, {get_correlated, StoreId, CorrelationId}). %% @doc Build a causation graph for visualization -spec build_causation_graph(atom(), binary()) -> {ok, map()} | {error, term()}. build_causation_graph(StoreId, Id) -> route_call(StoreId, {build_causation_graph, StoreId, Id}). %%==================================================================== %% Schema Operations %%==================================================================== %% @doc Get a schema by event type -spec get_schema(atom(), binary()) -> {ok, map()} | {error, term()}. get_schema(StoreId, EventType) -> route_call(StoreId, {get_schema, StoreId, EventType}). %% @doc List all schemas -spec list_schemas(atom()) -> {ok, list()} | {error, term()}. list_schemas(StoreId) -> route_call(StoreId, {list_schemas, StoreId}). %% @doc Get the version of a schema -spec get_schema_version(atom(), binary()) -> {ok, integer()} | {error, term()}. get_schema_version(StoreId, EventType) -> route_call(StoreId, {get_schema_version, StoreId, EventType}). %% @doc Upcast events to current schema version -spec upcast_events(atom(), list()) -> {ok, list()} | {error, term()}. upcast_events(StoreId, Events) -> route_call(StoreId, {upcast_events, StoreId, Events}). %% @doc Register a schema -spec register_schema(atom(), binary(), map()) -> ok. register_schema(StoreId, EventType, Schema) -> route_cast(StoreId, {register_schema, StoreId, EventType, Schema}), ok. %% @doc Unregister a schema -spec unregister_schema(atom(), binary()) -> ok. unregister_schema(StoreId, EventType) -> route_cast(StoreId, {unregister_schema, StoreId, EventType}), ok. %%==================================================================== %% Memory Pressure Operations %%==================================================================== %% @doc Get current memory pressure level -spec get_memory_level(atom()) -> {ok, atom()} | {error, term()}. get_memory_level(StoreId) -> route_call(StoreId, {get_memory_level, StoreId}). %% @doc Get memory statistics -spec get_memory_stats(atom()) -> {ok, map()} | {error, term()}. get_memory_stats(StoreId) -> route_call(StoreId, {get_memory_stats, StoreId}). %%==================================================================== %% Link Operations %%==================================================================== %% @doc Create a new link -spec create_link(atom(), map()) -> ok. create_link(StoreId, LinkSpec) -> route_cast(StoreId, {create_link, StoreId, LinkSpec}), ok. %% @doc Delete a link -spec delete_link(atom(), binary()) -> ok. delete_link(StoreId, LinkName) -> route_cast(StoreId, {delete_link, StoreId, LinkName}), ok. %% @doc Get a link by name -spec get_link(atom(), binary()) -> {ok, map()} | {error, term()}. get_link(StoreId, LinkName) -> route_call(StoreId, {get_link, StoreId, LinkName}). %% @doc List all links -spec list_links(atom()) -> {ok, list()} | {error, term()}. list_links(StoreId) -> route_call(StoreId, {list_links, StoreId}). %% @doc Start a link -spec start_link(atom(), binary()) -> ok. start_link(StoreId, LinkName) -> route_cast(StoreId, {start_link, StoreId, LinkName}), ok. %% @doc Stop a link -spec stop_link(atom(), binary()) -> ok. stop_link(StoreId, LinkName) -> route_cast(StoreId, {stop_link, StoreId, LinkName}), ok. %% @doc Get detailed info about a link -spec link_info(atom(), binary()) -> {ok, map()} | {error, term()}. link_info(StoreId, LinkName) -> route_call(StoreId, {link_info, StoreId, LinkName}). %%==================================================================== %% Internal Functions %%==================================================================== %% @private Route a synchronous call to a random worker with retry -spec route_call(atom(), term()) -> {ok, term()} | {error, term()}. route_call(StoreId, Request) -> route_call(StoreId, Request, ?CALL_TIMEOUT). -spec route_call(atom(), term(), timeout()) -> {ok, term()} | {error, term()}. route_call(StoreId, Request, Timeout) -> StartTime = erlang:monotonic_time(microsecond), emit_request_start(StoreId, Request), Result = esdb_gater_retry:with_retry(StoreId, fun() -> do_route_call(StoreId, Request, Timeout) end), Duration = erlang:monotonic_time(microsecond) - StartTime, emit_request_complete(StoreId, Request, Duration, Result), Result. %% @private Execute a single call attempt to a worker -spec do_route_call(atom(), term(), timeout()) -> {ok, term()} | {error, term()}. do_route_call(StoreId, Request, Timeout) -> case select_worker(StoreId) of {ok, #worker_entry{pid = Pid}} -> try Result = gen_server:call(Pid, Request, Timeout), %% Don't double-wrap if result is already {ok, _} or {error, _} case Result of {ok, _} -> Result; {error, _} -> Result; ok -> {ok, ok}; _ -> {ok, Result} end catch exit:{timeout, _} -> {error, timeout}; exit:{noproc, _} -> {error, worker_down}; exit:{{nodedown, _}, _} -> {error, node_down}; Class:Reason -> {error, {Class, Reason}} end; {error, no_workers} -> {error, no_workers} end. %% @private Route an asynchronous cast to a random worker -spec route_cast(atom(), term()) -> ok. route_cast(StoreId, Request) -> case select_worker(StoreId) of {ok, #worker_entry{pid = Pid}} -> gen_server:cast(Pid, Request); {error, _Reason} -> logger:warning("No workers available for cast to store ~p", [StoreId]) end, ok. %% @private Select a worker using round-robin -spec select_worker(atom()) -> {ok, worker_entry()} | {error, no_workers}. select_worker(StoreId) -> case esdb_gater_worker_registry:get_workers(StoreId) of {ok, []} -> {error, no_workers}; {ok, Workers} -> %% Simple round-robin using process dictionary Index = case get({worker_index, StoreId}) of undefined -> 0; N -> N end, Worker = lists:nth((Index rem length(Workers)) + 1, Workers), put({worker_index, StoreId}, Index + 1), {ok, Worker}; {error, _} = Error -> Error end. %% @private Emit telemetry for request start -spec emit_request_start(atom(), term()) -> ok. emit_request_start(StoreId, Request) -> RequestType = case Request of {Type, _, _} -> Type; {Type, _, _, _} -> Type; {Type, _, _, _, _} -> Type; {Type, _, _, _, _, _} -> Type; {Type} -> Type; _ -> unknown end, telemetry:execute( ?GATER_REQUEST_START, #{system_time => erlang:system_time(millisecond)}, #{store_id => StoreId, request_type => RequestType} ). %% @private Emit telemetry for request completion -spec emit_request_complete(atom(), term(), non_neg_integer(), {ok, term()} | {error, term()}) -> ok. emit_request_complete(StoreId, Request, Duration, {ok, _}) -> RequestType = case Request of {Type, _, _} -> Type; {Type, _, _, _} -> Type; {Type, _, _, _, _} -> Type; {Type, _, _, _, _, _} -> Type; {Type} -> Type; _ -> unknown end, telemetry:execute( ?GATER_REQUEST_STOP, #{duration => Duration}, #{store_id => StoreId, request_type => RequestType, result => success} ); emit_request_complete(StoreId, Request, Duration, {error, Reason}) -> RequestType = case Request of {Type, _, _} -> Type; {Type, _, _, _} -> Type; {Type, _, _, _, _} -> Type; {Type, _, _, _, _, _} -> Type; {Type} -> Type; _ -> unknown end, telemetry:execute( ?GATER_REQUEST_ERROR, #{duration => Duration}, #{store_id => StoreId, request_type => RequestType, reason => Reason} ).