gen_cluster

-module(gen_cluster).
-moduledoc "
Process that manages calling the necessary discovery and connect/disconnect
functions at a configurable interval.
".

-behaviour(gen_statem).

-export([start_link/1]).

-export([init/1,
         active/3,
         inactive/3,
         callback_mode/0,
         terminate/3,
         code_change/4]).

-export([nodes_to_peers/1]).

-type builtin_discovery() :: {dns, gc_discover_dns_ip:options() | gc_discover_dns_srv:options()} |
                             {static, gc_discover_static:options()} |
                             {epmd_all, gc_discover_epmd_all:options()}.

-type option() :: {discovery,
                   builtin_discovery() |
                   {module(), term()}} |
                  {dist, module()} |
                  {refresh_interval_ms, integer() | undefined}.

-type config() :: [option()].

-type peer()   :: #{node := atom(),
                    port => inet:port_number()}.

-type peers() :: sets:set(peer()).

%%

-type cb_state() :: term().

-export_type([config/0,
              peer/0,
              peers/0]).

-record(data, {discovery           :: {module(), cb_state()},
               dist                :: {module(), cb_state()},
               refresh_interval_ms :: integer() | infinity}).

-define(DEFAULT_REFRESH_INTERVAL_MS, 5000).

-spec start_link(config()) -> {ok, pid()} | ignore | {error, term()}.
start_link(Configuration) ->
    gen_statem:start_link(?MODULE, [Configuration], []).

callback_mode() ->
    [state_functions].

-spec init([config()]) -> gen_statem:init_result(inactive | active, #data{}).
init([Configuration]) ->
    Data = data_from_config(Configuration),
    {ok, inactive, Data, [{next_event, internal, refresh}]}.

inactive(internal, refresh, Data=#data{refresh_interval_ms=RefreshIntervalMs}) ->
    Data1 = handle_refresh(Data),
    {next_state, active, Data1, [{timeout, RefreshIntervalMs, refresh}]};
inactive(_EventType, _EventContent, _Data) ->
    keep_state_and_data.

active(timeout, refresh, Data=#data{refresh_interval_ms=RefreshIntervalMs}) ->
    Data1 = handle_refresh(Data),
    {keep_state, Data1, [{timeout, RefreshIntervalMs, refresh}]};
active(_EventType, _EventContent, _Data) ->
    keep_state_and_data.

terminate(_Reason, _State, _Data) ->
    ok.

code_change(_, _OldState, Data, _) ->
    {ok, Data}.

%% Internal functions

-spec data_from_config(config()) -> #data{}.
data_from_config(Configuration) ->
    Discovery = init_callback(proplists:get_value(discovery,
                                                  Configuration,
                                                  {static, []})),
    Dist = init_callback(proplists:get_value(dist,
                                             Configuration,
                                             {gc_dist_erl, []})),
    RefreshIntervalMs =
        proplists:get_value(refresh_interval_ms,
                            Configuration,
                            ?DEFAULT_REFRESH_INTERVAL_MS),
    #data{discovery=Discovery,
          dist=Dist,
          refresh_interval_ms=RefreshIntervalMs}.

init_callback({dns, Config}) ->
    Module = to_dns_module(maps:get(record_type, Config, ip)),
    init_callback_({Module, Config});
init_callback({static, Config}) ->
    init_callback_({gc_discover_static, Config});
init_callback({epmd_all, Config}) ->
    init_callback_({gc_discover_epmd_all, Config});
init_callback({Module, Config}) ->
    %% use a separate function here so the `dns' case can't accidentally cause an
    %% infinite loop if `record_type' were set to `dns'
    init_callback_({Module, Config}).

init_callback_({Module, Config}) ->
    {ok, State} = run_callback({Module, Config}, init, []),
    {Module, State}.

handle_refresh(Data=#data{discovery=Discovery,
                          dist=Dist}) ->
    Members = run_callback(Dist, members, []),
    FoundPeers = run_callback(Discovery, peers, []),
    ToConnect = sets:subtract(FoundPeers, Members),
    _ = sets:fold(fun(Peer, _) ->
                          maybe_connect(Peer, Members, Dist)
                  end, [], ToConnect),
    Data.

maybe_connect(Peer, Members, Dist) ->
    case sets:is_element(Peer, Members) of
        false ->
            run_callback(Dist, connect, [Peer]);
        true ->
            false
    end.

-spec nodes_to_peers([node()]) -> peers().
nodes_to_peers(Nodes) ->
    lists:foldl(fun(Name, Acc) ->
                      sets:add_element(#{node => Name}, Acc)
                end, sets:new([{version, 2}]), Nodes).

-spec run_callback({module(), cb_state()}, atom(), list()) -> dynamic().
run_callback({CallbackMod, State}, Fun, Args) ->
    erlang:apply(CallbackMod, Fun, Args ++ [State]).

to_dns_module(ip) ->
    gc_discover_dns_ip;
to_dns_module(srv) ->
    gc_discover_dns_srv.