evoq

%% @doc ETS-based command idempotency store.
%%
%% Stores command results by command_id for deduplication.
%% Duplicate commands return the cached result instead of
%% re-executing against the aggregate.
%%
%% Uses ETS with automatic TTL-based expiration.
%%
%% @author rgfaber
-module(evoq_idempotency).
-behaviour(gen_server).

-include("evoq.hrl").
-include("evoq_telemetry.hrl").

%% API
-export([start_link/0]).
-export([store/3, lookup/1, delete/1]).
-export([check_and_store/3]).

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

-define(TABLE, evoq_idempotency_table).
-define(DEFAULT_TTL, 3600000).  %% 1 hour
-define(CLEANUP_INTERVAL, 60000).  %% 1 minute

-record(state, {
    cleanup_ref :: reference() | undefined
}).

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

%% @doc Start the idempotency store.
-spec start_link() -> {ok, pid()} | {error, term()}.
start_link() ->
    gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).

%% @doc Store a command result.
-spec store(binary(), term(), pos_integer()) -> ok.
store(CommandId, Result, TTL) ->
    ExpiresAt = erlang:system_time(millisecond) + TTL,
    ets:insert(?TABLE, {CommandId, Result, ExpiresAt}),
    ok.

%% @doc Lookup a command result.
-spec lookup(binary()) -> {ok, term()} | not_found.
lookup(CommandId) ->
    Now = erlang:system_time(millisecond),
    case ets:lookup(?TABLE, CommandId) of
        [{CommandId, Result, ExpiresAt}] when ExpiresAt > Now ->
            telemetry:execute(?TELEMETRY_IDEMPOTENCY_HIT, #{}, #{
                command_id => CommandId
            }),
            {ok, Result};
        [{CommandId, _Result, _ExpiresAt}] ->
            %% Expired, delete it
            ets:delete(?TABLE, CommandId),
            not_found;
        [] ->
            not_found
    end.

%% @doc Delete a command result.
-spec delete(binary()) -> ok.
delete(CommandId) ->
    ets:delete(?TABLE, CommandId),
    ok.

%% @doc Check for existing result or execute and store.
%% This is the main entry point for idempotent command execution.
-spec check_and_store(binary(), fun(() -> term()), pos_integer()) -> term().
check_and_store(CommandId, ExecuteFun, TTL) ->
    case lookup(CommandId) of
        {ok, CachedResult} ->
            CachedResult;
        not_found ->
            telemetry:execute(?TELEMETRY_IDEMPOTENCY_MISS, #{}, #{
                command_id => CommandId
            }),
            Result = ExecuteFun(),
            %% Only cache successful results
            case Result of
                {ok, _, _} -> store(CommandId, Result, TTL);
                {ok, _} -> store(CommandId, Result, TTL);
                ok -> store(CommandId, Result, TTL);
                _ -> ok
            end,
            Result
    end.

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

%% @private
init([]) ->
    %% Create ETS table (delete existing if any from crashed run)
    case ets:whereis(?TABLE) of
        undefined -> ok;
        _Tid -> ets:delete(?TABLE)
    end,

    ?TABLE = ets:new(?TABLE, [
        named_table,
        public,
        set,
        {read_concurrency, true},
        {write_concurrency, true}
    ]),

    %% Start cleanup timer
    Ref = erlang:send_after(?CLEANUP_INTERVAL, self(), cleanup),

    {ok, #state{cleanup_ref = Ref}}.

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

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

%% @private
handle_info(cleanup, State) ->
    cleanup_expired(),
    Ref = erlang:send_after(?CLEANUP_INTERVAL, self(), cleanup),
    {noreply, State#state{cleanup_ref = Ref}};

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

%% @private
terminate(_Reason, _State) ->
    ok.

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

%% @private
cleanup_expired() ->
    Now = erlang:system_time(millisecond),
    %% Use match_delete for efficiency
    ets:select_delete(?TABLE, [
        {{'_', '_', '$1'}, [{'<', '$1', Now}], [true]}
    ]).