reckon_db

%% @doc Cluster discovery for reckon-db
%%
%% Handles node discovery via UDP multicast (LAN) or Kubernetes DNS.
%% Ported from LibCluster's gossip strategy.
%%
%% Protocol:
%% 1. Broadcast {gossip, Node, ClusterSecret, Timestamp} every BROADCAST_INTERVAL
%% 2. On receive: verify secret, call net_kernel:connect_node/1
%% 3. On node up: trigger Khepri cluster join via StoreCoordinator
%%
%% @author rgfaber

-module(reckon_db_discovery).
-behaviour(gen_server).

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

%% API
-export([start_link/1]).
-export([trigger_discovery/1]).
-export([get_discovered_nodes/1]).

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

-define(DEFAULT_PORT, 45892).
-define(MULTICAST_ADDR, {239, 255, 0, 1}).
-define(BROADCAST_INTERVAL_MS, 5000).
-define(MULTICAST_TTL, 1).

-record(state, {
    store_id :: atom(),
    config :: store_config(),
    socket :: gen_udp:socket() | undefined,
    port :: non_neg_integer(),
    multicast_addr :: inet:ip4_address(),
    cluster_secret :: binary(),
    broadcast_interval :: non_neg_integer(),
    discovered_nodes :: [node()]
}).

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

-spec start_link(store_config()) -> {ok, pid()} | {error, term()}.
start_link(#store_config{store_id = StoreId} = Config) ->
    Name = reckon_db_naming:discovery_name(StoreId),
    gen_server:start_link({local, Name}, ?MODULE, Config, []).

%% @doc Trigger immediate discovery broadcast
-spec trigger_discovery(atom()) -> ok.
trigger_discovery(StoreId) ->
    Name = reckon_db_naming:discovery_name(StoreId),
    gen_server:cast(Name, trigger_discovery).

%% @doc Get list of discovered nodes
-spec get_discovered_nodes(atom()) -> [node()].
get_discovered_nodes(StoreId) ->
    Name = reckon_db_naming:discovery_name(StoreId),
    gen_server:call(Name, get_discovered_nodes).

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

init(#store_config{store_id = StoreId, mode = Mode} = Config) ->
    case Mode of
        cluster ->
            init_cluster_mode(StoreId, Config);
        single ->
            %% In single mode, discovery is a no-op
            logger:info("Discovery disabled in single-node mode (store: ~p)", [StoreId]),
            {ok, #state{
                store_id = StoreId,
                config = Config,
                socket = undefined,
                discovered_nodes = []
            }}
    end.

init_cluster_mode(StoreId, Config) ->
    Port = get_config_value(discovery_port, ?DEFAULT_PORT),
    MulticastAddr = get_config_value(multicast_addr, ?MULTICAST_ADDR),
    ClusterSecret = get_cluster_secret(),
    BroadcastInterval = get_config_value(broadcast_interval, ?BROADCAST_INTERVAL_MS),

    State = #state{
        store_id = StoreId,
        config = Config,
        port = Port,
        multicast_addr = MulticastAddr,
        cluster_secret = ClusterSecret,
        broadcast_interval = BroadcastInterval,
        discovered_nodes = []
    },

    case open_multicast_socket(Port, MulticastAddr) of
        {ok, Socket} ->
            logger:info("Discovery started on port ~p (store: ~p)", [Port, StoreId]),
            %% Schedule first broadcast
            schedule_broadcast(BroadcastInterval),
            {ok, State#state{socket = Socket}};
        {error, Reason} ->
            logger:warning("Failed to open multicast socket: ~p, running in passive mode", [Reason]),
            %% Continue without socket - will use manual discovery
            {ok, State#state{socket = undefined}}
    end.

handle_call(get_discovered_nodes, _From, #state{discovered_nodes = Nodes} = State) ->
    {reply, Nodes, State};

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

handle_cast(trigger_discovery, #state{socket = undefined} = State) ->
    %% No socket, can't broadcast
    {noreply, State};

handle_cast(trigger_discovery, State) ->
    broadcast_presence(State),
    {noreply, State};

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

handle_info(broadcast, #state{socket = undefined} = State) ->
    %% No socket, schedule next attempt
    schedule_broadcast(State#state.broadcast_interval),
    {noreply, State};

handle_info(broadcast, #state{broadcast_interval = Interval} = State) ->
    broadcast_presence(State),
    schedule_broadcast(Interval),
    {noreply, State};

handle_info({udp, _Socket, _IP, _Port, Data}, State) ->
    NewState = handle_gossip_message(Data, State),
    {noreply, NewState};

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

terminate(_Reason, #state{socket = undefined}) ->
    ok;
terminate(_Reason, #state{socket = Socket}) ->
    gen_udp:close(Socket),
    ok.

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

%% @private Open UDP multicast socket
-spec open_multicast_socket(non_neg_integer(), inet:ip4_address()) ->
    {ok, gen_udp:socket()} | {error, term()}.
open_multicast_socket(Port, MulticastAddr) ->
    Options = [
        binary,
        {active, true},
        {reuseaddr, true},
        {multicast_ttl, ?MULTICAST_TTL},
        {multicast_loop, false},
        {add_membership, {MulticastAddr, {0, 0, 0, 0}}}
    ],
    gen_udp:open(Port, Options).

%% @private Broadcast presence to multicast group
-spec broadcast_presence(#state{}) -> ok.
broadcast_presence(#state{socket = Socket, port = Port, multicast_addr = Addr,
                          cluster_secret = Secret, store_id = StoreId}) ->
    Message = encode_gossip_message(node(), Secret),
    case gen_udp:send(Socket, Addr, Port, Message) of
        ok ->
            ok;
        {error, Reason} ->
            logger:warning("Failed to broadcast discovery message (store: ~p): ~p",
                          [StoreId, Reason])
    end.

%% @private Encode gossip message
-spec encode_gossip_message(node(), binary()) -> binary().
encode_gossip_message(Node, Secret) ->
    Timestamp = erlang:system_time(millisecond),
    term_to_binary({gossip, Node, Secret, Timestamp}).

%% @private Handle incoming gossip message
-spec handle_gossip_message(binary(), #state{}) -> #state{}.
handle_gossip_message(Data, #state{cluster_secret = OurSecret, store_id = StoreId,
                                   discovered_nodes = KnownNodes} = State) ->
    try binary_to_term(Data) of
        {gossip, Node, Secret, _Timestamp} when Secret =:= OurSecret, Node =/= node() ->
            handle_discovered_node(Node, StoreId, KnownNodes, State);
        {gossip, Node, _Secret, _Timestamp} when Node =/= node() ->
            %% Wrong secret, ignore
            logger:debug("Ignoring gossip from ~p with invalid secret", [Node]),
            State;
        _ ->
            State
    catch
        _:_ ->
            %% Invalid message format
            State
    end.

%% @private Handle a newly discovered node
-spec handle_discovered_node(node(), atom(), [node()], #state{}) -> #state{}.
handle_discovered_node(Node, StoreId, KnownNodes, State) ->
    case lists:member(Node, KnownNodes) of
        true ->
            State;
        false ->
            logger:info("Discovered new node: ~p (store: ~p)", [Node, StoreId]),
            connect_discovered_node(Node, StoreId, KnownNodes, State)
    end.

connect_discovered_node(Node, StoreId, KnownNodes, State) ->
    case net_kernel:connect_node(Node) of
        true ->
            logger:info("Connected to discovered node: ~p", [Node]),
            telemetry:execute(
                ?CLUSTER_NODE_UP,
                #{system_time => erlang:system_time(millisecond)},
                #{store_id => StoreId, node => Node,
                  member_count => length(nodes()) + 1}
            ),
            trigger_cluster_join(StoreId),
            State#state{discovered_nodes = [Node | KnownNodes]};
        false ->
            logger:warning("Failed to connect to discovered node: ~p", [Node]),
            State
    end.

%% @private Trigger cluster join via store coordinator
-spec trigger_cluster_join(atom()) -> ok.
trigger_cluster_join(StoreId) ->
    %% Use spawn to avoid blocking discovery
    spawn(fun() ->
        try
            reckon_db_store_coordinator:join_cluster(StoreId)
        catch
            _:Reason ->
                logger:warning("Failed to trigger cluster join: ~p", [Reason])
        end
    end),
    ok.

%% @private Schedule next broadcast
-spec schedule_broadcast(non_neg_integer()) -> reference().
schedule_broadcast(Interval) ->
    erlang:send_after(Interval, self(), broadcast).

%% @private Get cluster secret from environment or config
-spec get_cluster_secret() -> binary().
get_cluster_secret() ->
    case os:getenv("RECKON_DB_CLUSTER_SECRET") of
        false ->
            %% Use default secret (not recommended for production)
            <<"reckon_db_default_secret">>;
        Secret ->
            list_to_binary(Secret)
    end.

%% @private Get config value with default
-spec get_config_value(atom(), term()) -> term().
get_config_value(Key, Default) ->
    case application:get_env(reckon_db, Key) of
        {ok, Value} -> Value;
        undefined -> Default
    end.