reckon_gater

%% @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} = reckon_gater_api:append_events(my_store, StreamId, Events).
%%   {ok, Events} = reckon_gater_api:get_events(my_store, StreamId, 0, 100, forward).
%%   {ok, Version} = reckon_gater_api:get_version(my_store, StreamId).
%%
%% == Subscription Operations ==
%%
%%   {ok, Key} = reckon_gater_api:save_subscription(
%%       my_store, by_stream, Selector, Name, 0, self()).
%%   ok = reckon_gater_api:remove_subscription(my_store, by_stream, Selector, Name).
%%
%% == Snapshot Operations ==
%%
%%   ok = reckon_gater_api:record_snapshot(my_store, SourceId, StreamId, Version, Data).
%%   {ok, Snapshot} = reckon_gater_api:read_snapshot(my_store, SourceId, StreamId, Version).
%%
%% @author rgfaber

-module(reckon_gater_api).

-include("reckon_gater.hrl").
-include("reckon_gater_telemetry.hrl").

%% Worker registration
-export([
    register_worker/1,
    register_worker/2,
    unregister_worker/1,
    unregister_worker/2,
    get_workers/1,
    pick_worker/3
]).

%% DCB read operations (reckon-gater 3.4.1+)
-export([
    dcb_read_log/3,
    dcb_all_tags/1,
    dcb_all_event_types/1
]).

%% CCC payload-index reads (reckon-gater 3.6.0+)
-export([
    ccc_read_by_payload/4,
    ccc_read_by_payload_hash/4
]).

%% Store index introspection (reckon-gater 3.7.0+)
-export([
    get_payload_indexes/1,
    get_payload_hash_indexes/1
]).

%% Stream operations
-export([
    append_events/3,
    append_events/4,
    append_if_no_tag_matches/4,
    get_events/5,
    stream_forward/4,
    stream_backward/4,
    get_version/2,
    has_events/1,
    get_streams/1,
    global_event_count/1,
    integrity_status/1,
    delete_stream/2,
    read_by_event_types/3,
    read_by_tags/2,
    read_by_tags/3,
    read_by_metadata/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
]).

%% 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,
    get_resource_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
]).

%% Store inspector operations
-export([
    store_stats/1,
    list_all_snapshots/1,
    list_store_subscriptions/1,
    subscription_lag/2,
    event_type_summary/1,
    stream_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) ->
    reckon_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) ->
    reckon_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) ->
    reckon_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 Conditionally append events under the DCB pseudo-stream
%% (Dynamic Consistency Boundary, reckon-db 3.1.0+).
%%
%% Unlike append_events/3,4, the precondition is NOT a stream-version
%% check; it is a tag-filter context query. Returns
%% {error, {context_changed, MaxSeq}} when any event matching
%% TagFilter has seq greater than SeqCutoff. Nothing is written in
%% the conflict case.
%%
%% v1 backend (reckon-db 3.1.0) refuses on stores with integrity
%% enabled; returns {error, integrity_not_supported_in_dcb_v1}.
%%
%% See: PLAN_DCB_IMPLEMENTATION.md in reckon-db for the full design.
-spec append_if_no_tag_matches(
    StoreId   :: atom(),
    TagFilter :: tag_filter(),
    SeqCutoff :: seq_cutoff(),
    Events    :: list()
) ->
      {ok, LastSeq :: non_neg_integer()}
    | {error, {context_changed, non_neg_integer()}}
    | {error, no_events}
    | {error, integrity_not_supported_in_dcb_v1}
    | {error, term()}.
append_if_no_tag_matches(StoreId, TagFilter, SeqCutoff, Events) ->
    route_call(StoreId,
        {append_if_no_tag_matches, StoreId, TagFilter, SeqCutoff, Events}).

%%====================================================================
%% DCB Read Operations
%%====================================================================

%% @doc Read DCB events in ascending seq order.
%%
%% Returns {ok, #{events => Events, total_count => N}} on success.
%% FromSeq is inclusive (0 = beginning). Limit is capped to 200 on
%% the gateway side for admin UI use.
-spec dcb_read_log(atom(), non_neg_integer(), pos_integer()) ->
    {ok, map()} | {error, term()}.
dcb_read_log(StoreId, FromSeq, Limit) ->
    route_call(StoreId, {dcb_read_log, StoreId, FromSeq, Limit}).

%% @doc Return all tags in the DCB log with event counts.
%%
%% Returns {ok, [{Tag, Count}]} sorted descending by Count.
-spec dcb_all_tags(atom()) -> {ok, [{binary(), non_neg_integer()}]} | {error, term()}.
dcb_all_tags(StoreId) ->
    route_call(StoreId, {dcb_all_tags, StoreId}).

%% @doc Return all event types in the DCB log with event counts.
%%
%% Returns {ok, [{EventType, Count}]} sorted descending by Count.
-spec dcb_all_event_types(atom()) -> {ok, [{binary(), non_neg_integer()}]} | {error, term()}.
dcb_all_event_types(StoreId) ->
    route_call(StoreId, {dcb_all_event_types, StoreId}).

%% @doc Read DCB events that have a specific payload field value.
%%
%% Requires the store to declare {payload, Key} in its index config.
%% Returns {ok, Events} with at most Limit events in ascending seq order.
-spec ccc_read_by_payload(atom(), binary(), binary(), pos_integer()) ->
    {ok, [event()]} | {error, term()}.
ccc_read_by_payload(StoreId, Key, Value, Limit) ->
    route_call(StoreId, {dcb_read_by_payload, StoreId, Key, Value, Limit}).

%% @doc Read DCB events matching a composite payload field combination.
%%
%% Requires {payload_hash, Keys} declared in the store's index config.
%% All Keys must match their corresponding Values; field order is ignored.
%% Returns {ok, Events} with at most Limit events in ascending seq order.
-spec ccc_read_by_payload_hash(atom(), [binary()], [binary()], pos_integer()) ->
    {ok, [event()]} | {error, term()}.
ccc_read_by_payload_hash(StoreId, Keys, Values, Limit) ->
    route_call(StoreId, {dcb_read_by_payload_hash, StoreId, Keys, Values, Limit}).

%% @doc Return the payload field keys that are individually indexed in a store.
%%
%% Each returned binary is a key that was declared as {payload, Key} in the
%% store's index config. Use these to know which ccc_read_by_payload/4 calls
%% will be O(matches) rather than a full-store scan.
-spec get_payload_indexes(atom()) -> {ok, [binary()]} | {error, term()}.
get_payload_indexes(StoreId) ->
    route_call(StoreId, {get_payload_indexes, StoreId}).

%% @doc Return the payload field key-sets that are hash-indexed in a store.
%%
%% Each element is a list of keys declared as {payload_hash, Keys} in the
%% store's index config. Use these to know which ccc_read_by_payload_hash/4
%% combinations will be O(matches) rather than a full-store scan.
-spec get_payload_hash_indexes(atom()) -> {ok, [[binary()]]} | {error, term()}.
get_payload_hash_indexes(StoreId) ->
    route_call(StoreId, {get_payload_hash_indexes, StoreId}).

%% @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 Check if a store contains at least one event.
-spec has_events(atom()) -> boolean().
has_events(StoreId) ->
    case route_call(StoreId, {has_events, StoreId}) of
        {ok, Bool} when is_boolean(Bool) -> Bool;
        {ok, _} -> true;
        {error, _} -> false
    end.

%% @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 The store's monotonic total event count — an O(1) read of the
%% counter maintained on every append, for ingest-rate dashboards.
-spec global_event_count(atom()) -> {ok, non_neg_integer()} | {error, term()}.
global_event_count(StoreId) ->
    route_call(StoreId, {global_event_count, StoreId}).

%% @doc The store's public event-integrity status (enabled + algo + key id;
%% never the key bytes). A store-owned property, so a catalogue gateway that
%% does not host the store dispatches for it rather than reading local state.
-spec integrity_status(atom()) -> {ok, map()} | {error, term()}.
integrity_status(StoreId) ->
    route_call(StoreId, {integrity_status, 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 events whose metadata key = value.
%%
%% The sanctioned primitive for cross-cutting lookups by a metadata field
%% (e.g. an application's causation_id / correlation_id read model). When
%% the store declared the `{meta, Key}' secondary index this is an
%% O(matches) indexed read; otherwise the server falls back to a whole-store
%% scan. The store does NOT interpret what the key means — lineage
%% traversal/graphs are the application's job.
%%
%% @param StoreId The store identifier
%% @param Key The metadata key (binary)
%% @param Value The value to match (binary)
-spec read_by_metadata(atom(), binary(), binary()) ->
    {ok, list()} | {error, term()}.
read_by_metadata(StoreId, Key, Value) ->
    route_call(StoreId, {read_by_metadata, StoreId, Key, Value}).

%% @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 (create) a subscription.
%%
%% Synchronous as of reckon-gater 2.1.2 — the previous
%% `route_cast' fire-and-forget swallowed errors like
%% `{invalid_filter, _}', so a client subscribe call would see
%% gRPC ok while no events ever flowed. Now returns the worker's
%% real result; consumers (gateway handlers, application code)
%% can translate to a proper user-visible error.
%%
%% Returns `{ok, Key}' on success (Key is the subscription's
%% internal id), `{error, Reason}' on validation / filter
%% failure (non-retriable; see {@link reckon_gater_retry}).
-spec save_subscription(atom(), atom(), binary() | map(), binary(), non_neg_integer(), pid() | undefined) ->
    {ok, binary()} | {error, term()}.
save_subscription(StoreId, Type, Selector, SubscriptionName, StartFrom, Subscriber) ->
    route_call(StoreId, {save_subscription, StoreId, Type, Selector, SubscriptionName, StartFrom, Subscriber}).

%% @doc Remove a subscription.
%%
%% Returns `ok' on success (including the idempotent case where the
%% subscription was already gone), `{error, Reason}' on transport
%% failure. Removing a non-existent subscription is NOT an error;
%% removal is the desired terminal state regardless of starting state.
-spec remove_subscription(atom(), atom(), binary() | map(), binary()) ->
    ok | {error, term()}.
remove_subscription(StoreId, Type, Selector, SubscriptionName) ->
    route_call(StoreId, {remove_subscription, StoreId, Type, Selector, SubscriptionName}).

%% @doc Acknowledge receipt of an event by a subscriber.
%%
%% Returns `ok' on success, `{error, {subscription_not_found, _}}' if
%% the subscription has been removed (acking a dead subscription is a
%% caller error; surface it). Non-retriable.
-spec ack_event(atom(), binary(), pid(), map()) -> ok | {error, term()}.
ack_event(StoreId, SubscriptionName, SubscriberPid, Event) ->
    route_call(StoreId, {ack_event, StoreId, SubscriptionName, SubscriberPid, Event}).

%%====================================================================
%% 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 reckon_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 reckon_gater_worker_registry:get_all_workers() of
        {ok, Workers} ->
            StoreStats = maps:fold(fun count_store_entries/3, #{}, Workers),
            {ok, #{
                status => healthy,
                stores => StoreStats,
                total_workers => maps:fold(fun sum_worker_counts/3, 0, StoreStats),
                node => node(),
                timestamp => erlang:system_time(millisecond)
            }};
        {error, Reason} ->
            {ok, #{
                status => degraded,
                reason => Reason,
                node => node(),
                timestamp => erlang:system_time(millisecond)
            }}
    end.

count_store_entries(StoreId, Entries, Acc) ->
    maps:put(StoreId, length(Entries), Acc).

sum_worker_counts(_StoreId, Count, Acc) ->
    Acc + Count.

%% @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}).

%%====================================================================
%% 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}).

%% @doc Get node-wide CPU + disk resource stats for the node hosting this store
%% (see reckon_db_resource_monitor). Requires the store node on reckon_db >= 5.10.
-spec get_resource_stats(atom()) -> {ok, map()} | {error, term()}.
get_resource_stats(StoreId) ->
    route_call(StoreId, {get_resource_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}).

%%====================================================================
%% Store Inspector Operations
%%====================================================================

%% @doc Aggregate statistics for a store.
-spec store_stats(atom()) -> {ok, map()} | {error, term()}.
store_stats(StoreId) ->
    route_call(StoreId, {store_stats, StoreId}).

%% @doc List all snapshots across all streams in a store.
-spec list_all_snapshots(atom()) -> {ok, [map()]} | {error, term()}.
list_all_snapshots(StoreId) ->
    route_call(StoreId, {list_all_snapshots, StoreId}).

%% @doc List all subscriptions for a store with checkpoint positions.
-spec list_store_subscriptions(atom()) -> {ok, [map()]} | {error, term()}.
list_store_subscriptions(StoreId) ->
    route_call(StoreId, {list_store_subscriptions, StoreId}).

%% @doc Calculate lag for a specific subscription.
-spec subscription_lag(atom(), binary()) -> {ok, map()} | {error, term()}.
subscription_lag(StoreId, SubscriptionName) ->
    route_call(StoreId, {subscription_lag, StoreId, SubscriptionName}).

%% @doc Census of event types in the store with counts.
-spec event_type_summary(atom()) -> {ok, [map()]} | {error, term()}.
event_type_summary(StoreId) ->
    route_call(StoreId, {event_type_summary, StoreId}).

%% @doc Detailed info for a single stream (timestamps, snapshot coverage).
-spec stream_info(atom(), binary()) -> {ok, map()} | {error, term()}.
stream_info(StoreId, StreamId) ->
    route_call(StoreId, {stream_info, StoreId, StreamId}).

%%====================================================================
%% 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 = reckon_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) ->
    route_to_selected(select_worker(StoreId), Request, Timeout).

route_to_selected({ok, #worker_entry{pid = Pid}}, Request, Timeout) ->
    call_worker(Pid, Request, Timeout);
route_to_selected({error, no_workers}, _Request, _Timeout) ->
    {error, no_workers}.

call_worker(Pid, Request, Timeout) ->
    try
        %% Don't double-wrap if result is already {ok, _} or {error, _}
        unwrap_call_result(gen_server:call(Pid, Request, Timeout))
    catch
        exit:{timeout, _} ->
            {error, timeout};
        exit:{noproc, _} ->
            {error, worker_down};
        exit:{{nodedown, _}, _} ->
            {error, node_down};
        Class:Reason ->
            {error, {Class, Reason}}
    end.

unwrap_call_result({ok, _} = Result) -> Result;
unwrap_call_result({error, _} = Result) -> Result;
unwrap_call_result(ok) -> {ok, ok};
unwrap_call_result(Result) -> {ok, Result}.

%% @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.

%% @doc Pure worker-selection policy behind {@link select_worker/1}.
%%
%% Prefers workers on the caller's own node when any exist; falls
%% back to cluster-wide round-robin only when no local worker is
%% present. Extracted so the policy can be exercised without a
%% running pg scope.
%%
%% == Why local-first ==
%%
%% The worker registry is pg-based, so a call to {@link get_workers/1}
%% returns PIDs from every BEAM-connected node. That's the right pool
%% for stores that form a shared Raft cluster — any worker writes
%% into the same Khepri state machine and reads are cluster-wide.
%%
%% For stores that stay local per node (autojoin=false), a write
%% routed to a remote worker persists in THAT node's private store
%% and is invisible to the caller's local store. Preferring the
%% caller's own node keeps each daemon's writes on its own disk.
%% Falling back to the remote pool preserves the cluster-wide
%% behaviour when the local worker isn't registered yet (boot race)
%% or when the store is genuinely clustered and the caller has no
%% local presence.
-spec pick_worker([worker_entry()], node(), non_neg_integer()) ->
    {ok, worker_entry(), non_neg_integer()} | {error, no_workers}.
pick_worker([], _Node, _Index) ->
    {error, no_workers};
pick_worker(Workers, Node, Index) ->
    Candidates = case [W || W <- Workers, W#worker_entry.node =:= Node] of
        []    -> Workers;
        Local -> Local
    end,
    Pick = lists:nth((Index rem length(Candidates)) + 1, Candidates),
    {ok, Pick, Index + 1}.

%% @private Select a worker using round-robin with local-node preference.
-spec select_worker(atom()) -> {ok, worker_entry()} | {error, no_workers}.
select_worker(StoreId) ->
    selected_worker(reckon_gater_worker_registry:get_workers(StoreId), StoreId).

selected_worker({ok, Workers}, StoreId) ->
    Index = current_worker_index(StoreId),
    picked_worker(pick_worker(Workers, node(), Index), StoreId);
selected_worker({error, _} = Error, _StoreId) ->
    Error.

current_worker_index(StoreId) ->
    case get({worker_index, StoreId}) of
        undefined -> 0;
        N -> N
    end.

picked_worker({ok, Worker, NextIndex}, StoreId) ->
    put({worker_index, StoreId}, NextIndex),
    {ok, Worker};
picked_worker({error, no_workers} = Err, _StoreId) ->
    Err.

%% @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}
    ).