macula

%%%-------------------------------------------------------------------
%%% @doc
%%% GenServer managing Kademlia DHT routing table and operations.
%%% Integrates all routing components: table, DHT algorithms, protocol.
%%%
%%% Pure functional helpers are in macula_routing_storage.
%%% Network transport and escalation are in macula_routing_network.
%%% @end
%%%-------------------------------------------------------------------
-module(macula_routing_server).
-behaviour(gen_server).

-include_lib("kernel/include/logger.hrl").

%% API
-export([
    start_link/2,
    add_node/2,
    remove_node/2,
    find_closest/3,
    store_local/3,
    store/3,
    get_local/2,
    get_all_keys/1,
    delete_local/3,
    find_value/3,
    find_value_local/2,
    get_routing_table/1,
    size/1,
    handle_message/2,
    handle_message_async/2
]).

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

%% Stale peer eviction interval (60 seconds)
-define(EVICT_INTERVAL_MS, 60000).
%% Peers not seen in 5 minutes are considered stale
-define(STALE_THRESHOLD_MS, 300000).
%% DHT value expiry interval (30 seconds)
-define(VALUE_EXPIRY_INTERVAL_MS, 30000).

%% State
-record(state, {
    local_node_id :: binary(),
    routing_table :: macula_routing_table:routing_table(),
    storage :: #{binary() => term()},  % Local key-value storage
    config :: #{
        k => pos_integer(),
        alpha => pos_integer(),
        escalation_enabled => boolean(),
        escalation_timeout => pos_integer()
    }
}).

%%%===================================================================
%%% API Functions
%%%===================================================================

%% @doc Start routing server with registered name macula_routing_server.
-spec start_link(binary(), map()) -> {ok, pid()} | {error, term()}.
start_link(LocalNodeId, Config) ->
    gen_server:start_link({local, macula_routing_server}, ?MODULE, {LocalNodeId, Config}, []).

%% @doc Add node to routing table (async - does not block caller).
-spec add_node(pid(), macula_routing_bucket:node_info()) -> ok.
add_node(Pid, NodeInfo) ->
    gen_server:cast(Pid, {add_node, NodeInfo}).

%% @doc Remove node from routing table (async - does not block caller).
-spec remove_node(pid(), binary()) -> ok.
remove_node(Pid, NodeId) ->
    gen_server:cast(Pid, {remove_node, NodeId}).

%% @doc Find k closest nodes to target.
-spec find_closest(pid(), binary(), pos_integer()) -> [macula_routing_bucket:node_info()].
find_closest(Pid, Target, K) ->
    gen_server:call(Pid, {find_closest, Target, K}).

%% @doc Store value locally.
-spec store_local(pid(), binary(), term()) -> ok.
store_local(Pid, Key, Value) ->
    gen_server:call(Pid, {store_local, Key, Value}).

%% @doc Store value in DHT by propagating to k closest nodes.
%% Stores locally first, then sends STORE messages to k closest peers.
-spec store(pid(), binary(), term()) -> ok.
store(Pid, Key, Value) ->
    gen_server:call(Pid, {store, Key, Value}, 10000).

%% @doc Get value from local storage.
-spec get_local(pid(), binary()) -> {ok, term()} | not_found.
get_local(Pid, Key) ->
    gen_server:call(Pid, {get_local, Key}).

%% @doc Get all keys from local storage.
-spec get_all_keys(pid()) -> {ok, [binary()]} | {error, term()}.
get_all_keys(Pid) ->
    gen_server:call(Pid, get_all_keys).

%% @doc Delete value from local storage.
-spec delete_local(pid(), binary(), binary()) -> ok.
delete_local(Pid, Key, NodeId) ->
    gen_server:call(Pid, {delete_local, Key, NodeId}).

%% @doc Find value in DHT -- goes through gen_server for full iterative lookup.
%% For local-only reads (answering peer queries), use find_value_local/2.
-spec find_value(pid(), binary(), pos_integer()) ->
    {ok, term()} | {nodes, [macula_routing_bucket:node_info()]} | {error, term()}.
find_value(Pid, Key, K) ->
    gen_server:call(Pid, {find_value, Key, K}, 10000).

%% @doc Fast local-only lookup via ETS -- for answering incoming FIND_VALUE from peers.
%% Does NOT query the network. Returns only locally stored values.
-spec find_value_local(binary(), pos_integer()) ->
    {ok, term()} | {error, not_found}.
find_value_local(Key, _K) ->
    case ets:info(macula_dht_storage) of
        undefined -> {error, not_found};
        _ ->
            case ets:lookup(macula_dht_storage, Key) of
                [{_, Values}] when is_list(Values), Values =/= [] -> {ok, Values};
                _ -> {error, not_found}
            end
    end.

%% @doc Get routing table snapshot.
-spec get_routing_table(pid()) -> macula_routing_table:routing_table().
get_routing_table(Pid) ->
    gen_server:call(Pid, get_routing_table).

%% @doc Get number of nodes in routing table.
-spec size(pid()) -> non_neg_integer().
size(Pid) ->
    gen_server:call(Pid, size).

%% @doc Handle incoming DHT message and return reply.
-spec handle_message(pid(), map()) -> map().
handle_message(Pid, Message) ->
    gen_server:call(Pid, {handle_message, Message}).

%% @doc Handle incoming DHT message asynchronously (fire-and-forget).
%% Use for STORE messages where no reply is needed.
-spec handle_message_async(pid(), map()) -> ok.
handle_message_async(Pid, Message) ->
    gen_server:cast(Pid, {handle_message_async, Message}).

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

%% @private
init({LocalNodeId, Config}) ->
    K = maps:get(k, Config, 20),
    Alpha = maps:get(alpha, Config, 3),
    EscalationEnabled = maps:get(escalation_enabled, Config, true),
    EscalationTimeout = maps:get(escalation_timeout, Config, 5000),

    %% Named ETS table for concurrent DHT storage reads (bypasses gen_server mailbox)
    StorageEts = ets:new(macula_dht_storage, [named_table, set, public, {read_concurrency, true}]),

    State = #state{
        local_node_id = LocalNodeId,
        routing_table = macula_routing_table:new(LocalNodeId, K),
        storage = #{},
        config = #{
            k => K,
            alpha => Alpha,
            escalation_enabled => EscalationEnabled,
            escalation_timeout => EscalationTimeout,
            storage_ets => StorageEts
        }
    },

    erlang:send_after(?EVICT_INTERVAL_MS, self(), evict_stale_peers),
    erlang:send_after(?VALUE_EXPIRY_INTERVAL_MS, self(), expire_stale_values),

    {ok, State}.

%% @private
handle_call({find_closest, Target, K}, _From, #state{routing_table = Table} = State) ->
    Closest = macula_routing_table:find_closest(Table, Target, K),
    {reply, Closest, State};

handle_call({store_local, Key, Value}, _From, #state{storage = Storage, config = Config} = State) ->
    ExistingProviders = maps:get(Key, Storage, []),
    ProviderList = macula_routing_storage:ensure_provider_list(ExistingProviders),
    NodeId = maps:get(node_id, Value, undefined),
    UpdatedProviders = macula_routing_storage:upsert_provider(NodeId, Value, ProviderList),
    NewStorage = Storage#{Key => UpdatedProviders},
    Ets = maps:get(storage_ets, Config, undefined),
    macula_routing_storage:mirror_to_ets(Ets, Key, UpdatedProviders),
    {reply, ok, State#state{storage = NewStorage}};

handle_call({store, Key, Value}, _From, #state{routing_table = Table, config = Config} = State) ->
    %% 1. Store locally first
    NewState = case handle_call({store_local, Key, Value}, _From, State) of
        {reply, ok, S} -> S;
        _ -> State
    end,

    %% 2. Find k closest nodes to Key
    K = maps:get(k, Config, 20),
    ClosestNodes = macula_routing_table:find_closest(Table, Key, K),

    %% 3. Send STORE message to each node (fire-and-forget, non-blocking)
    StoreMsg = macula_routing_protocol:encode_store(Key, Value),
    spawn(fun() -> macula_routing_network:propagate_store_to_peers(ClosestNodes, StoreMsg, Key, Value) end),

    {reply, ok, NewState};

handle_call({get_local, Key}, _From, #state{storage = Storage} = State) ->
    Reply = macula_routing_storage:format_storage_value(maps:get(Key, Storage, undefined)),
    {reply, Reply, State};

handle_call(get_all_keys, _From, #state{storage = Storage} = State) ->
    {reply, {ok, maps:keys(Storage)}, State};

handle_call(get_routing_table, _From, #state{routing_table = Table} = State) ->
    {reply, Table, State};

handle_call(size, _From, #state{routing_table = Table} = State) ->
    {reply, macula_routing_table:size(Table), State};

handle_call({delete_local, Key, NodeId}, _From, #state{storage = Storage, config = Config} = State) ->
    NewStorage = macula_routing_storage:delete_provider_from_storage(Key, NodeId, Storage),
    Ets = maps:get(storage_ets, Config, undefined),
    macula_routing_storage:mirror_delete_to_ets(Ets, Key, NewStorage),
    {reply, ok, State#state{storage = NewStorage}};

handle_call({find_value, Key, K}, From, #state{routing_table = Table, storage = Storage,
                                              config = Config} = State) ->
    ?LOG_DEBUG("find_value: key=~p, storage_size=~p", [Key, maps:size(Storage)]),
    LocalValues = case maps:get(Key, Storage, undefined) of
        undefined -> [];
        V when is_list(V) -> V;
        V -> [V]
    end,
    %% Always query the network -- local values may be incomplete (e.g. pubsub subscribers).
    %% Spawn to avoid blocking the routing server.
    spawn(fun() ->
        RemoteResult = macula_routing_network:find_value_with_escalation(Key, K, Storage, Table, Config),
        RemoteValues = case RemoteResult of
            {ok, R} when is_list(R) -> R;
            _ -> []
        end,
        AllValues = macula_routing_storage:deduplicate_providers(LocalValues ++ RemoteValues),
        Reply = case AllValues of
            [] -> {error, not_found};
            _ -> {ok, AllValues}
        end,
        gen_server:reply(From, Reply)
    end),
    {noreply, State};

handle_call({handle_message, Message}, _From, State) ->
    {Reply, NewState} = process_dht_message(Message, State),
    {reply, Reply, NewState};

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

%% @private
handle_cast({add_node, NodeInfo}, #state{routing_table = Table, storage = Storage,
                                         config = Config} = State) ->
    OldSize = macula_routing_table:size(Table),
    NewTable = macula_routing_table:add_node(Table, NodeInfo),
    NewSize = macula_routing_table:size(NewTable),
    NodeId = maps:get(node_id, NodeInfo, undefined),
    ?LOG_DEBUG("[RoutingServer] add_node: node_id=~s, table_size: ~p -> ~p",
              [case NodeId of B when is_binary(B), byte_size(B) =:= 32 -> binary:encode_hex(B);
                              B when is_binary(B) -> B; _ -> <<"?">> end,
               OldSize, NewSize]),
    case NewSize > OldSize of
        true ->
            spawn(fun() -> macula_routing_network:replicate_to_new_peer(NodeInfo, Storage, NewTable, Config) end);
        false ->
            ok
    end,
    {noreply, State#state{routing_table = NewTable}};

handle_cast({remove_node, NodeId}, #state{routing_table = Table} = State) ->
    OldSize = macula_routing_table:size(Table),
    NewTable = macula_routing_table:remove_node(Table, NodeId),
    NewSize = macula_routing_table:size(NewTable),
    ?LOG_DEBUG("[RoutingServer] remove_node: node_id=~s, table_size: ~p -> ~p",
              [binary:encode_hex(NodeId), OldSize, NewSize]),
    {noreply, State#state{routing_table = NewTable}};

%% @private
%% Handle async store (from _dht.store RPC) - stores locally only without propagation
handle_cast({store, Key, Value}, #state{storage = Storage, config = Config} = State) when is_binary(Key) ->
    ?LOG_DEBUG("Async STORE received - key hash prefix: ~p, value: ~p",
              [binary:part(Key, 0, min(8, byte_size(Key))), Value]),
    ExistingProviders = maps:get(Key, Storage, []),
    ProviderList = macula_routing_storage:ensure_provider_list(ExistingProviders),
    NodeId = macula_routing_storage:get_node_id_from_value(Value),
    UpdatedProviders = macula_routing_storage:upsert_provider(NodeId, Value, ProviderList),
    NewStorage = Storage#{Key => UpdatedProviders},
    Ets = maps:get(storage_ets, Config, undefined),
    macula_routing_storage:mirror_to_ets(Ets, Key, UpdatedProviders),
    ?LOG_DEBUG("Stored value for key, now have ~p provider(s)", [length(UpdatedProviders)]),
    {noreply, State#state{storage = NewStorage}};

handle_cast({store, Key, _Value}, State) ->
    ?LOG_ERROR("Async STORE received with non-binary key: ~p", [Key]),
    {noreply, State};

handle_cast({handle_message_async, Message}, State) ->
    {_Reply, NewState} = process_dht_message(Message, State),
    {noreply, NewState};

handle_cast(_Request, State) ->
    {noreply, State}.

%% @private
handle_info(evict_stale_peers, #state{routing_table = Table} = State) ->
    StaleThreshold = erlang:system_time(millisecond) - ?STALE_THRESHOLD_MS,
    OldSize = macula_routing_table:size(Table),
    NewTable = macula_routing_table:evict_stale(Table, StaleThreshold),
    NewSize = macula_routing_table:size(NewTable),
    case OldSize - NewSize of
        0 -> ok;
        Evicted ->
            ?LOG_INFO("[RoutingServer] Evicted ~p stale peer(s) (table: ~p -> ~p)",
                      [Evicted, OldSize, NewSize])
    end,
    erlang:send_after(?EVICT_INTERVAL_MS, self(), evict_stale_peers),
    {noreply, State#state{routing_table = NewTable}};

handle_info(expire_stale_values, #state{storage = Storage, config = Config} = State) ->
    Now = erlang:system_time(millisecond),
    {NewStorage, Expired} = maps:fold(fun(Key, Providers, {StorageAcc, ExpiredCount}) ->
        Fresh = [P || P <- Providers, not macula_routing_storage:is_provider_expired(P, Now)],
        case {Fresh, length(Providers) - length(Fresh)} of
            {[], Removed} ->
                {maps:remove(Key, StorageAcc), ExpiredCount + Removed};
            {Kept, 0} ->
                {StorageAcc#{Key => Kept}, ExpiredCount};
            {Kept, Removed} ->
                {StorageAcc#{Key => Kept}, ExpiredCount + Removed}
        end
    end, {Storage, 0}, Storage),
    case Expired of
        0 -> ok;
        N -> ?LOG_INFO("[RoutingServer] Expired ~p stale DHT value(s)", [N])
    end,
    %% Mirror to ETS
    case Expired > 0 of
        true ->
            case maps:get(storage_ets, Config, undefined) of
                undefined -> ok;
                Ets ->
                    ets:delete_all_objects(Ets),
                    maps:foreach(fun(K, V) -> ets:insert(Ets, {K, V}) end, NewStorage)
            end;
        false -> ok
    end,
    erlang:send_after(?VALUE_EXPIRY_INTERVAL_MS, self(), expire_stale_values),
    {noreply, State#state{storage = NewStorage}};

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

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

%%%===================================================================
%%% Internal Functions - DHT Message Processing
%%%===================================================================

%% @doc Process incoming DHT message and generate reply.
-spec process_dht_message(map(), #state{}) -> {map(), #state{}}.
process_dht_message(Message, State) ->
    MessageType = classify_dht_message(Message),
    dispatch_dht_message(MessageType, Message, State).

classify_dht_message(Message) ->
    case macula_routing_protocol:is_find_node(Message) of
        true -> find_node;
        false -> classify_non_find_node(Message)
    end.

classify_non_find_node(Message) ->
    case macula_routing_protocol:is_store(Message) of
        true -> store;
        false -> classify_find_value(Message)
    end.

classify_find_value(Message) ->
    case macula_routing_protocol:is_find_value(Message) of
        true -> find_value;
        false -> unknown
    end.

dispatch_dht_message(find_node, Message, State) ->
    handle_find_node(Message, State);
dispatch_dht_message(store, Message, State) ->
    handle_store(Message, State);
dispatch_dht_message(find_value, Message, State) ->
    handle_find_value(Message, State);
dispatch_dht_message(unknown, _Message, State) ->
    {#{type => error, reason => unknown_message}, State}.

%% @doc Handle FIND_NODE request.
-spec handle_find_node(map(), #state{}) -> {map(), #state{}}.
handle_find_node(Message, #state{routing_table = Table, config = Config} = State) ->
    {ok, Target} = macula_routing_protocol:decode_find_node(Message),
    K = maps:get(k, Config, 20),
    TableSize = macula_routing_table:size(Table),
    Closest = macula_routing_table:find_closest(Table, Target, K),
    ?LOG_DEBUG("[RoutingServer] FIND_NODE: target=~s, table_size=~p, found=~p",
              [binary:encode_hex(Target), TableSize, length(Closest)]),
    Reply = macula_routing_protocol:encode_find_node_reply(Closest),
    {Reply, State}.

%% @doc Handle STORE request.
-spec handle_store(map(), #state{}) -> {map(), #state{}}.
handle_store(Message, #state{storage = Storage} = State) ->
    {ok, Key, Value} = macula_routing_protocol:decode_store(Message),
    ?LOG_DEBUG("[RoutingServer] STORE received: key_prefix=~p, value_node_id=~p",
             [binary:part(Key, 0, min(8, byte_size(Key))), maps:get(node_id, Value, maps:get(<<"node_id">>, Value, unknown))]),
    ?LOG_DEBUG("STORE: key=~p, value=~p", [Key, Value]),
    ExistingProviders = maps:get(Key, Storage, []),
    ProviderList = macula_routing_storage:ensure_provider_list(ExistingProviders),
    NodeId = macula_routing_storage:get_node_id_from_value(Value),
    UpdatedProviders = macula_routing_storage:upsert_provider(NodeId, Value, ProviderList),
    NewStorage = Storage#{Key => UpdatedProviders},
    ?LOG_DEBUG("STORE complete: new_storage_size=~p, key_prefix=~p",
              [maps:size(NewStorage), binary:part(Key, 0, min(8, byte_size(Key)))]),
    Reply = #{type => store_reply, result => ok},
    {Reply, State#state{storage = NewStorage}}.

%% @doc Handle FIND_VALUE request.
-spec handle_find_value(map(), #state{}) -> {map(), #state{}}.
handle_find_value(Message, #state{storage = Storage, routing_table = Table, config = Config} = State) ->
    {ok, Key} = macula_routing_protocol:decode_find_value(Message),
    StorageValue = maps:get(Key, Storage, undefined),
    K = maps:get(k, Config, 20),
    ?LOG_DEBUG("[RoutingServer] FIND_VALUE: key_prefix=~p, storage_size=~p, found=~p",
             [binary:part(Key, 0, min(8, byte_size(Key))), maps:size(Storage), StorageValue =/= undefined]),
    Reply = encode_find_value_result(StorageValue, Key, K, Table),
    {Reply, State}.

encode_find_value_result(undefined, Key, K, Table) ->
    Closest = macula_routing_table:find_closest(Table, Key, K),
    macula_routing_protocol:encode_find_value_reply({nodes, Closest});
encode_find_value_result(Value, _Key, _K, _Table) when is_list(Value) ->
    macula_routing_protocol:encode_find_value_reply({value, Value});
encode_find_value_result(Value, _Key, _K, _Table) ->
    macula_routing_protocol:encode_find_value_reply({value, Value}).