reckon_db

%% @doc Khepri store lifecycle management for reckon-db
%%
%% Manages the Khepri store instance, including:
%% - Starting and stopping the store
%% - Cluster formation (in cluster mode)
%% - Health checks
%%
%% @author rgfaber

-module(reckon_db_store).
-behaviour(gen_server).

-include("reckon_db.hrl").
-include("reckon_db_telemetry.hrl").

%% API
-export([start_link/1]).
-export([get_store/1, is_ready/1, get_leader/1]).

%% gen_server callbacks
-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2]).

-record(state, {
    store_id :: atom(),
    config :: store_config(),
    started_at :: integer()
}).

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

%% @doc Start the store worker
%% IMPORTANT: We use store_worker_name/1 for gen_server registration to avoid
%% conflicting with Khepri's internal naming. Khepri uses the StoreId for
%% its Ra cluster and process registration.
-spec start_link(store_config()) -> {ok, pid()} | {error, term()}.
start_link(#store_config{store_id = StoreId} = Config) ->
    WorkerName = reckon_db_naming:store_worker_name(StoreId),
    gen_server:start_link({local, WorkerName}, ?MODULE, Config, []).

%% @doc Get the store name (for use with khepri operations)
-spec get_store(atom()) -> atom().
get_store(StoreId) ->
    StoreId.

%% @doc Check if the store is ready
-spec is_ready(atom()) -> boolean().
is_ready(StoreId) ->
    try
        khepri:exists(StoreId, [])
    catch
        _:_ -> false
    end.

%% @doc Get the current leader node for the store
-spec get_leader(atom()) -> {ok, node()} | {error, term()}.
get_leader(StoreId) ->
    case khepri_cluster:get_store_ids() of
        [] ->
            {error, not_started};
        _ ->
            try
                case ra:members({StoreId, node()}) of
                    {ok, _Members, Leader} when is_tuple(Leader) ->
                        {_LeaderName, LeaderNode} = Leader,
                        {ok, LeaderNode};
                    {ok, _, _} ->
                        {error, no_leader};
                    Error ->
                        Error
                end
            catch
                _:Reason -> {error, Reason}
            end
    end.

%%====================================================================
%% gen_server callbacks
%%====================================================================

%% @private
init(#store_config{store_id = StoreId, data_dir = DataDir, mode = Mode} = Config) ->
    process_flag(trap_exit, true),

    StartTime = erlang:system_time(millisecond),

    %% Ensure data directory exists
    ok = filelib:ensure_dir(filename:join(DataDir, "dummy")),

    %% Start Khepri store
    case start_khepri_store(StoreId, DataDir, Mode) of
        ok ->
            %% Initialize store paths
            ok = init_store_paths(StoreId),

            %% Emit telemetry
            telemetry:execute(
                ?STORE_STARTED,
                #{system_time => StartTime},
                #{store_id => StoreId, mode => Mode, data_dir => DataDir}
            ),

            logger:info("Khepri store ~p started in ~p mode", [StoreId, Mode]),

            %% Announce store to the distributed registry
            ok = reckon_db_store_registry:announce_store(StoreId, Config),

            State = #state{
                store_id = StoreId,
                config = Config,
                started_at = StartTime
            },
            {ok, State};
        {error, Reason} = Error ->
            logger:error("Failed to start Khepri store ~p: ~p", [StoreId, Reason]),
            {stop, Error}
    end.

%% @private
handle_call(get_state, _From, State) ->
    {reply, State, State};

handle_call(_Request, _From, State) ->
    {reply, {error, unknown_request}, State}.

%% @private
handle_cast(_Msg, State) ->
    {noreply, State}.

%% @private
handle_info(_Info, State) ->
    {noreply, State}.

%% @private
terminate(Reason, #state{store_id = StoreId, started_at = StartedAt}) ->
    Uptime = erlang:system_time(millisecond) - StartedAt,

    %% Unannounce store from the distributed registry
    catch reckon_db_store_registry:unannounce_store(StoreId),

    %% Emit telemetry
    telemetry:execute(
        ?STORE_STOPPED,
        #{system_time => erlang:system_time(millisecond), uptime_ms => Uptime},
        #{store_id => StoreId, reason => Reason}
    ),

    %% Stop Khepri store
    case khepri:stop(StoreId) of
        ok ->
            logger:info("Khepri store ~p stopped (uptime: ~pms)", [StoreId, Uptime]);
        {error, StopReason} ->
            logger:warning("Error stopping Khepri store ~p: ~p", [StoreId, StopReason])
    end,
    ok.

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

%% @private
-spec start_khepri_store(atom(), string(), single | cluster) -> ok | {error, term()}.
start_khepri_store(StoreId, DataDir, single) ->
    %% Single node mode - simple khepri start
    %% IMPORTANT: First argument MUST be DataDir (string/binary), not StoreId (atom).
    %% When khepri:start/2 receives a path as first arg, it auto-creates the Ra system.
    %% When it receives an atom, it assumes that Ra system already exists and fails
    %% with :system_not_started if it doesn't.
    Timeout = application:get_env(reckon_db, default_timeout, ?DEFAULT_TIMEOUT),
    case khepri:start(DataDir, StoreId, Timeout) of
        {ok, _} -> ok;
        {error, {already_started, _}} -> ok;
        Error -> Error
    end;

start_khepri_store(StoreId, DataDir, cluster) ->
    %% Cluster mode - start with Ra cluster configuration
    %% IMPORTANT: First argument MUST be DataDir (string/binary), not StoreId (atom).
    %% When khepri:start/2 receives a path as first arg, it auto-creates the Ra system.
    %% When it receives an atom, it assumes that Ra system already exists and fails
    %% with :system_not_started if it doesn't.
    RaServerConfig = #{
        cluster_name => StoreId,
        id => {StoreId, node()},
        uid => atom_to_binary(StoreId, utf8),
        initial_members => [{StoreId, node()}],
        log_init_args => #{uid => atom_to_binary(StoreId, utf8)},
        machine => {module, khepri_machine, #{store_id => StoreId}}
    },
    KhepriOpts = #{
        store_id => StoreId,
        ra_server_config => RaServerConfig
    },
    case khepri:start(DataDir, KhepriOpts) of
        {ok, _} -> ok;
        {error, {already_started, _}} -> ok;
        Error -> Error
    end.

%% @private
-spec init_store_paths(atom()) -> ok.
init_store_paths(StoreId) ->
    %% Ensure base paths exist
    Paths = [
        ?STREAMS_PATH,
        ?SNAPSHOTS_PATH,
        ?SUBSCRIPTIONS_PATH,
        ?METADATA_PATH
    ],
    lists:foreach(
        fun(Path) ->
            case khepri:exists(StoreId, Path) of
                false ->
                    khepri:put(StoreId, Path, #{});
                true ->
                    ok
            end
        end,
        Paths
    ),
    ok.