reckon_db

%% @doc Subscriptions store for reckon-db
%%
%% Manages subscription persistence and retrieval directly via Khepri.
%% This is a facade module that provides direct access to the subscription
%% storage without going through a gen_server, since Khepri/Ra handles
%% concurrency internally.
%%
%% @author rgfaber

-module(reckon_db_subscriptions_store).

-include("reckon_db.hrl").
-include_lib("khepri/include/khepri.hrl").

%% API
-export([
    put/2,
    get/2,
    delete/2,
    exists/2,
    exists/3,
    list/1,
    key/1,
    key/3,
    update_checkpoint/3,
    find_by_name/2
]).

%%====================================================================
%% Types
%%====================================================================

-type store_id() :: atom().

%%====================================================================
%% API
%%====================================================================

%% @doc Generate a unique key for a subscription
%%
%% The key is a phash2 hash of {type, selector, subscription_name}
-spec key(subscription()) -> binary().
key(#subscription{type = Type, selector = Selector, subscription_name = Name}) ->
    key(Type, Selector, Name).

%% @doc Generate a unique key for a subscription from components
-spec key(subscription_type(), binary() | map(), binary()) -> binary().
key(Type, Selector, SubscriptionName) ->
    integer_to_binary(erlang:phash2({Type, Selector, SubscriptionName})).

%% @doc Store a subscription
-spec put(store_id(), subscription()) -> ok | {error, term()}.
put(StoreId, #subscription{} = Subscription) ->
    Key = key(Subscription),
    Path = ?SUBSCRIPTIONS_PATH ++ [Key],
    %% Add the id to the subscription if not set
    SubscriptionWithId = case Subscription#subscription.id of
        undefined -> Subscription#subscription{id = Key};
        _ -> Subscription
    end,
    case khepri:put(StoreId, Path, SubscriptionWithId) of
        ok -> ok;
        {error, _} = Error -> Error
    end.

%% @doc Get a subscription by key
-spec get(store_id(), binary()) -> subscription() | undefined.
get(StoreId, Key) when is_binary(Key) ->
    Path = ?SUBSCRIPTIONS_PATH ++ [Key],
    case khepri:get(StoreId, Path) of
        {ok, Subscription} when is_record(Subscription, subscription) ->
            Subscription;
        {ok, SubscriptionMap} when is_map(SubscriptionMap) ->
            map_to_subscription(SubscriptionMap);
        _ ->
            undefined
    end.

%% @doc Delete a subscription by key
-spec delete(store_id(), binary()) -> ok | {error, term()}.
delete(StoreId, Key) when is_binary(Key) ->
    Path = ?SUBSCRIPTIONS_PATH ++ [Key],
    case khepri:delete(StoreId, Path) of
        ok -> ok;
        {error, _} = Error -> Error
    end.

%% @doc Check if a subscription exists by key
-spec exists(store_id(), binary()) -> boolean().
exists(StoreId, Key) when is_binary(Key) ->
    Path = ?SUBSCRIPTIONS_PATH ++ [Key],
    case khepri:exists(StoreId, Path) of
        true -> true;
        false -> false;
        {error, _} -> false
    end.

%% @doc Check if a subscription exists by record
-spec exists(store_id(), subscription_type(), binary()) -> boolean().
exists(StoreId, Type, SubscriptionName) ->
    %% We need to search through subscriptions since we don't have
    %% the selector to generate the key
    case list(StoreId) of
        {ok, Subs} ->
            lists:any(
                fun(#subscription{type = T, subscription_name = N}) ->
                    T =:= Type andalso N =:= SubscriptionName
                end,
                Subs
            );
        _ ->
            false
    end.

%% @doc List all subscriptions in the store
-spec list(store_id()) -> {ok, [subscription()]} | {error, term()}.
list(StoreId) ->
    Path = ?SUBSCRIPTIONS_PATH ++ [?KHEPRI_WILDCARD_STAR],
    case khepri:get_many(StoreId, Path) of
        {ok, Results} when is_map(Results) ->
            Subs = [convert_to_subscription(V) || {_, V} <- maps:to_list(Results)],
            {ok, [S || S <- Subs, S =/= undefined]};
        {ok, _} ->
            {ok, []};
        {error, _} = Error ->
            Error
    end.

%% @doc Find a subscription by name
%%
%% Searches all subscriptions for one matching the given name.
%% Returns the subscription key and record if found.
-spec find_by_name(store_id(), binary()) -> {ok, binary(), subscription()} | {error, not_found}.
find_by_name(StoreId, SubscriptionName) ->
    case list(StoreId) of
        {ok, Subscriptions} ->
            find_matching_subscription(Subscriptions, SubscriptionName);
        {error, _} ->
            {error, not_found}
    end.

%% @doc Update the checkpoint for a subscription
%%
%% Parameters:
%%   StoreId - The store identifier
%%   Key     - The subscription key
%%   Position - The new checkpoint position
-spec update_checkpoint(store_id(), binary(), non_neg_integer()) -> ok | {error, term()}.
update_checkpoint(StoreId, Key, Position) ->
    case ?MODULE:get(StoreId, Key) of
        undefined ->
            {error, not_found};
        Subscription ->
            UpdatedSubscription = Subscription#subscription{checkpoint = Position},
            ?MODULE:put(StoreId, UpdatedSubscription)
    end.

%%====================================================================
%% Internal functions
%%====================================================================

%% @private Find a subscription by name in a list
-spec find_matching_subscription([subscription()], binary()) ->
    {ok, binary(), subscription()} | {error, not_found}.
find_matching_subscription([], _Name) ->
    {error, not_found};
find_matching_subscription([#subscription{subscription_name = Name} = Sub | _], Name) ->
    {ok, key(Sub), Sub};
find_matching_subscription([_ | Rest], Name) ->
    find_matching_subscription(Rest, Name).

%% @private Convert value to subscription record
-spec convert_to_subscription(term()) -> subscription() | undefined.
convert_to_subscription(V) when is_record(V, subscription) ->
    V;
convert_to_subscription(V) when is_map(V) ->
    map_to_subscription(V);
convert_to_subscription(_) ->
    undefined.

%% @private Convert map to subscription record
-spec map_to_subscription(map()) -> subscription().
map_to_subscription(Map) ->
    #subscription{
        id = maps:get(id, Map, undefined),
        type = maps:get(type, Map, stream),
        selector = maps:get(selector, Map, <<>>),
        subscription_name = maps:get(subscription_name, Map, <<>>),
        subscriber_pid = maps:get(subscriber_pid, Map, undefined),
        created_at = maps:get(created_at, Map, 0),
        pool_size = maps:get(pool_size, Map, 1),
        checkpoint = maps:get(checkpoint, Map, undefined),
        options = maps:get(options, Map, #{})
    }.