-module(lashup_hyparview_membership).
-author("sdhillon").
-behaviour(gen_server).

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

-ifdef(TEST).
-include_lib("eunit/include/eunit.hrl").
-endif.

%% API
-export([
  start_link/0,
  get_active_view/0,
  get_passive_view/0,
  do_probe/1,
  ping_failed/1,
  recognize_pong/1,
  recommend_neighbor/1,
  update_masters/1,
  init_metrics/0
]).

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

-record(monitor, {
    monitor_ref :: reference(),
    node :: node()
}).
-type monitor() :: #monitor{}.

-record(state, {
  active_view = ordsets:new() :: ordsets:ordset(),
  passive_view = ordsets:new() :: ordsets:ordset(),
  monitors = [] :: [monitor()],
  fixed_seed :: rand:state(),
  idx = 1 :: pos_integer(),
  unfilled_active_set_count = 0, %% Number of times I've tried to neighbor and I've not seen an active set
  join_window,
  ping_idx = 1 :: pos_integer(),
  joined = false :: boolean(),
  shuffle_interval = lashup_config:shuffle_interval() :: non_neg_integer(),
  join_interval = lashup_config:join_interval() :: non_neg_integer(),
  active_view_size = lashup_config:active_view_size() :: non_neg_integer(),
  passive_view_size = lashup_config:passive_view_size() :: non_neg_integer()
}).

-type state() :: state().

%% TODO: Make map types better defined
%% We probably want to define a partial of the map

-type join_success() :: map().
-type join() :: map().
-type join_deny() :: map().
-type forward_join() :: map().
-type disconnect() :: map().
-type neighbor() :: map().
-type neighbor_deny() :: map().
-type neighbor_accept() :: map().
-type shuffle() :: map().
-type shuffle_reply() :: map().
-type hyparview_message() :: join_success() | join() | forward_join() | disconnect() | neighbor() | neighbor_deny() |
                             neighbor_accept() | shuffle() | shuffle_reply() | join_deny().

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

%% The recognize ping function tells the hyparview membership server that
%% this node has us in their active view
recognize_pong(Pong) ->
  gen_server:cast(?MODULE, {recognize_pong, Pong}).

%% This is a way for lashup_gm_probe to recommend a neighbor
%% if it finds a neighbor that's outside of our reachability graph
recommend_neighbor(Node) ->
  gen_server:cast(?MODULE, {recommend_neighbor, Node}).

%% Pings the node immediately.
do_probe(Node) ->
  gen_server:cast(?MODULE, {do_probe, Node}).

ping_failed(Node) ->
  gen_server:cast(?MODULE, {ping_failed, Node}).

get_active_view() ->
  gen_server:call(?MODULE, get_active_view).

get_passive_view() ->
  gen_server:call(?MODULE, get_passive_view).

-spec update_masters([node()]) -> ok.
update_masters(Nodes) ->
  lashup_config:update_contact_nodes(Nodes).

-spec(start_link() ->
  {ok, Pid :: pid()} | ignore | {error, Reason :: term()}).
start_link() ->
  gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).

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

init([]) ->
  rand:seed(exsplus),
  %% This seed is a fixed seed used for shuffling the list
  FixedSeed = lashup_utils:seed(),
  MyPid = self(),
  spawn_link(fun() -> shuffle_backoff_loop(5000, MyPid) end),
  reschedule_join(5100),
  %% Schedule the maybe_neighbor
  reschedule_maybe_neighbor(30000),
  Window = lashup_utils:new_window(1000),
  reschedule_ping(60000),

  State = #state{
    passive_view = lashup_config:contact_nodes(),
    fixed_seed = FixedSeed, join_window = Window},
  {ok, State}.

handle_call(stop, _From, State) ->
  {stop, normal, State};
handle_call({do_connect, Node}, _From, State) when is_atom(Node) ->
  send_neighbor_to(5000, Node, high),
  {reply, ok, State};
handle_call(get_active_view, _From, State = #state{active_view = ActiveView}) ->
  {reply, ActiveView, State};
handle_call(get_passive_view, _From, State = #state{passive_view = PassiveView}) ->
  {reply, PassiveView, State};
handle_call(try_shuffle, _From, State) ->
  NewDelay = try_shuffle(State),
  {reply, NewDelay, State}.
%% No handler here, because no one should ever call us off-node, and it's indicative of a bug

%% We generated a timer (with a ref) when we did the join with a timeout
%% We get passed back that ref, and now we need to delete it
handle_cast({do_probe, Node}, State) ->
  State1 = handle_do_probe(Node, State),
  {noreply, check_state(State1)};
handle_cast(Message = #{message := Type}, State) ->
  {message_queue_len, MsgQueueLen} =
    erlang:process_info(self(), message_queue_len),
  prometheus_gauge:set(lashup, hyparview_message_queue_length, [], MsgQueueLen),
  State1 =
    prometheus_summary:observe_duration(
      lashup, hyparview_incoming_message_processing_seconds, [Type],
      fun () -> handle_message_cast(Message, State) end),
  {noreply, State1};
handle_cast({recognize_pong, Pong}, State) ->
  handle_recognize_pong(Pong, State),
  {noreply, State};
handle_cast({ping_failed, Node}, State) ->
  State1 = handle_ping_failed(Node, State),
  push_state(State1),
  {noreply, check_state(State1)};
handle_cast({recommend_neighbor, Node}, State) ->
  handle_recommend_neighbor(Node, State),
  {noreply, State};
handle_cast(Request, State) ->
  ?LOG_DEBUG("Received unknown cast: ~p", [Request]),
  {noreply, check_state(State)}.

%% It's likely just that someone connected to us, and that's okay
handle_info(DownMessage = {'DOWN', _, _, _, _}, State) ->
  State1 = handle_down_message(DownMessage, State),
  push_state(State1),
  {noreply, check_state(State1)};
%% We don't need a full active view
%% In fact, the network can get into (healthy) cases where it's not possible
%% Like running fewer than the active view size count nodes
%% So, only neighbor if more than 25% of our active view is open.
handle_info(maybe_neighbor, State0) when length(State0#state.active_view) >= State0#state.active_view_size * 0.75 ->
  %% Ignore this, because my active view is full
  reschedule_maybe_neighbor(),
  State1 = State0#state{unfilled_active_set_count = 0},
  {noreply, check_state(State1)};
handle_info(maybe_neighbor, State) ->
  ?LOG_DEBUG("Maybe neighbor triggered"),
  State1 = maybe_neighbor(State),
  push_state(State1),
  {noreply, check_state(State1)};
%% Stop trying to join is somehow someone connects to us
handle_info(join_failed, State = #state{active_view = ActiveView}) when length(ActiveView) > 0 ->
  {noreply, State};
handle_info(join_failed, State) ->
  ?LOG_DEBUG("Attempt to join timed out, rescheduling"),
  reschedule_join(),
  {noreply, State};
handle_info(try_join, State = #state{joined = true}) ->
  {noreply, State};
handle_info(try_join, State) ->
  try_do_join(),
  {noreply, State};
handle_info({tried_neighbor, Node}, State = #state{passive_view = PassiveView}) ->
  PassiveView1 = ordsets:del_element(Node, PassiveView),
  State1 = State#state{passive_view = PassiveView1},
  push_state(State1),
  {noreply, check_state(State1)};
handle_info(ping_rq, State) ->
  State1 = handle_ping_rq(State),
  State2 = check_state(State1),
  {noreply, State2, lashup_utils:hibernate()};
handle_info({maybe_disconnect, Node}, State) ->
  State1 = handle_maybe_disconnect(Node, State),
  {noreply, check_state(State1)};
handle_info(Info, State) ->
  ?LOG_DEBUG("Received unknown info: ~p", [Info]),
  {noreply, State}.

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

%% Message Dispatch functions
-spec(handle_message_cast(hyparview_message(), State :: state()) -> state()).
handle_message_cast(JoinSuccess = #{message := join_success}, State) ->
  check_state(handle_join_success(JoinSuccess, State));
handle_message_cast(JoinDeny = #{message := join_deny}, State) ->
  check_state(handle_join_deny(JoinDeny, State));
handle_message_cast(_Join = #{message := join, sender := Node, ref := Ref}, State) ->
  check_state(handle_join(Node, State, Ref));
handle_message_cast(ForwardJoin = #{message := forward_join}, State) ->
  State1 = handle_forward_join(ForwardJoin, State),
  push_state(State1),
  check_state(State1);
handle_message_cast(Disconnect = #{message := disconnect}, State) ->
  State1 = handle_disconnect(Disconnect, State),
  push_state(State1),
  check_state(State1);
handle_message_cast(Neighbor = #{message := neighbor}, State) ->
  State1 = handle_neighbor(Neighbor, State),
  push_state(State1),
  check_state(State1);
handle_message_cast(NeighborDeny = #{message := neighbor_deny}, State) ->
  check_state(handle_neighbor_deny(NeighborDeny, State));
handle_message_cast(NeighborAccept = #{message := neighbor_accept}, State) ->
  check_state(handle_neighbor_accept(NeighborAccept, State));
handle_message_cast(Shuffle = #{message := shuffle}, State) ->
  check_state(handle_shuffle(Shuffle, State));
handle_message_cast(ShuffleReply = #{message := shuffle_reply}, State) ->
  check_state(handle_shuffle_reply(ShuffleReply, State));
handle_message_cast(UnknownMessage, State) ->
  ?LOG_WARNING("Received unknown lashup message: ~p", [UnknownMessage]),
  State.

%% RESCHEDULING Functions
%% Ping every one of my neighbors at least every second
reschedule_ping() ->
  reschedule_ping(100).
reschedule_ping(Time) ->
  RandFloat = rand:uniform(),
  Multipler = 1 + round(RandFloat),
  Delay = Multipler * Time,
  timer:send_after(Delay, ping_rq).

reschedule_maybe_neighbor() ->
  reschedule_maybe_neighbor(lashup_config:neighbor_interval()).
reschedule_maybe_neighbor(Time) ->
  RandFloat = rand:uniform(),
  Multipler = 1 + round(RandFloat),
  Delay = Multipler * Time,
  timer:send_after(Delay, maybe_neighbor).

reschedule_join() ->
  reschedule_join(lashup_config:join_interval()).
reschedule_join(BaseTime) ->
  RandFloat = rand:uniform(),
  Multipler = 1 + round(RandFloat),
  timer:send_after(Multipler * BaseTime, try_join).

%%% Rescheduling functions

choose_node(Nodes) when length(Nodes) > 0 ->
  Length = erlang:length(Nodes),
  NodeIdx = rand:uniform(Length),
  lists:nth(NodeIdx, Nodes).

%% JOIN CODE
-spec try_do_join() -> error | ok.
try_do_join() ->
  case do_join() of
    ok ->
      ok;
    _ ->
      reschedule_join(),
      error
  end.

-spec(do_join() -> ok | {error, Reason :: term()}).
do_join() ->
  ContactNodes = lashup_config:contact_nodes(),
  case ContactNodes of
    [] ->
      {error, no_contact_nodes};
    _ ->
      Node = choose_node(ContactNodes),
      try_connect_then_join(Node)
  end.

-spec(try_connect_then_join(node()) -> ok | {error, Reason :: term()}).
try_connect_then_join(Node) ->
  Timeout = lashup_config:join_timeout(),
  case ping_with_timeout(Node, Timeout) of
    pong ->
      join(Node);
    %% This is the timeout case
    maybe_pong ->
      join(Node);
    _ ->
      {error, could_not_connect}
  end.

-spec(ping_with_timeout(Node :: node(), Timeout:: non_neg_integer()) -> ok | maybe_pong | pong | pang).
ping_with_timeout(Node, Timeout) ->
  Ref = make_ref(),
  Self = self(),
  spawn_link(fun() -> Self ! {Ref, net_adm:ping(Node)} end),
  receive
    {Ref, Response} ->
      Response
  after Timeout ->
    maybe_pong
  end.

-spec(join(node()) -> ok).
join(Node) ->
  {ok, Ref} = timer:send_after(1000, join_failed),
  cast(Node, #{message => join, sender => node(), ref => Ref}).

%% This is ridiculously inefficient on the order of O(N!),
%% but I'm prototyping
%% And this input shouldn't get much bigger than 30.

-spec(trim_ordset_to(ordsets:ordset(), non_neg_integer()) -> {ordsets:ordset(), ordsets:ordset()}).
trim_ordset_to(Ordset, Size) when Size > 0 ->
  trim_ordset_to(Ordset, Size, []).

trim_ordset_to(Ordset, Size, DroppedItems) when length(Ordset) > Size ->
  Idx = rand:uniform(length(Ordset)),
  ItemToRemove = lists:nth(Idx, Ordset),
  Ordset1 = ordsets:del_element(ItemToRemove, Ordset),
  DroppedItems1 = ordsets:add_element(ItemToRemove, DroppedItems),
  trim_ordset_to(Ordset1, Size, DroppedItems1);
trim_ordset_to(Ordset, _Size, DroppedItems) ->
  {Ordset, DroppedItems}.

-spec(handle_join(Node :: node(), State :: state(), Ref :: reference()) -> state()).
handle_join(Node, State = #state{join_window = JoinWindow, active_view = ActiveView, active_view_size = AVS}, Ref)
  when Node =/= node() ->
  ?LOG_DEBUG("Saw join from ~p", [Node]),
  State1 =
  case lashup_utils:count_ticks(JoinWindow) of
    %% Limit it to 25 joins/sec if the active view is full
    Num when length(ActiveView) == AVS andalso Num < 25 ->
      really_handle_join(Node, State, Ref);
    %% If the active view is less than that, throttle to 1 / sec
    Num when Num < 1 ->
      really_handle_join(Node, State, Ref);
    %% Else, drop it
    WindowSize ->
      ?LOG_WARNING("Throttling joins, window size: ~p, active view size: ~p", [WindowSize, length(ActiveView)]),
      deny_join(State, Node, Ref),
      State
  end,
  push_state(State1),
  State1;

handle_join(_, State, _) -> State.

-spec(deny_join(State :: state(), Node :: node(), Ref :: reference()) -> ok).
deny_join(_State = #state{active_view = ActiveView}, Node, Ref) ->
  Reply = #{message => join_deny, sender => node(), ref => Ref, active_view => ActiveView},
  cast(Node, Reply).

-spec(really_handle_join(Node :: node(), State :: state(), Ref :: reference()) -> state()).
really_handle_join(Node, State = #state{join_window = JoinWindow}, Ref) ->
  case try_add_node_to_active_view(Node, State) of
    {ok, NewState = #state{active_view = ActiveView, passive_view = PassiveView}} ->
      ARWL = lashup_config:arwl(),
      Fanout = ordsets:del_element(Node, ActiveView),
      ForwardJoinMessage = #{message => forward_join, node => Node, ttl => ARWL, sender => node(), seen => [node()]},
      abcast(Fanout, ForwardJoinMessage),
      Reply = #{message => join_success, sender => node(), ref => Ref, passive_view => PassiveView},
      cast(Node, Reply),
      JoinWindow1 = lashup_utils:add_tick(JoinWindow),
      NewState#state{join_window = JoinWindow1};
    {error, NewState} ->
      NewState
  end.

-spec(handle_join_deny(join_deny(), state()) -> state()).
handle_join_deny(_JoinDeny =  #{message := join_deny, sender := Sender, ref := Ref, active_view := RemoteActiveView},
    State = #state{active_view = []})->
  ContactNodes = lashup_config:contact_nodes(),
  ProhibitedNodes = ordsets:add_element(Sender, ContactNodes),
  case ordsets:subtract(RemoteActiveView, ProhibitedNodes) of
    [] ->
      State;
    RemoteActiveView1 ->
      timer:cancel(Ref),
      Node = choose_node(RemoteActiveView1),
      join(Node),
      State
  end;
handle_join_deny(_JoinDeny, State) ->
  State.

-spec(handle_join_success(join_success(), state()) -> state()).
handle_join_success(_JoinSuccess =
  #{message := join_success, sender := Sender, ref := Ref, passive_view := RemotePassiveView}, State) ->
  timer:cancel(Ref),
  State1 = lists:foldl(fun maybe_add_node_passive_view/2, State, RemotePassiveView),
  State3 =
  case try_add_node_to_active_view(Sender, State1) of
    {ok, State2} ->
      State2#state{joined = true};
    {error, State2} ->
      State2
  end,
  push_state(State3),
  State3.

-spec(trim_active_view(Size :: non_neg_integer(), State :: state()) -> state()).
trim_active_view(Size, State = #state{active_view = ActiveView}) ->
  {_, DroppedNodes} = trim_ordset_to(ActiveView, Size),
  State1 = lists:foldl(fun disconnect_node/2, State, DroppedNodes),
  case DroppedNodes of
    [] ->
      ok;
    _ ->
      ?LOG_DEBUG("Removing ~p from active view", [DroppedNodes])
  end,
  State1.

-spec(try_add_node_to_active_view(node(), state()) -> {ok, state()} | {error, state()}).
try_add_node_to_active_view(Node, State = #state{active_view_size = ActiveViewSize}) when Node =/= node() ->
  %% There is one critical component here -
  %% We have to return a non-error code once we decide to trim from the active view
  %% If we don't, it could end with an asymmetrical graph
  case net_kernel:connect_node(Node) of
    true ->
      State1 = trim_active_view(ActiveViewSize - 1, State),
      Monitors1 = State1#state.monitors,
      ActiveView1 = State1#state.active_view,
      PassiveView1 = State1#state.passive_view,
      Monitors2 = ensure_monitor(Node, Monitors1),
      ActiveView2 = ordsets:add_element(Node, ActiveView1),
      PassiveView2 = ordsets:del_element(Node, PassiveView1),
      State2 = State1#state{active_view = ActiveView2, passive_view = PassiveView2, monitors = Monitors2},
      {ok, State2};
    _ ->
      ?LOG_WARNING("Received join from node ~p, but could not connect back to it", [Node]),
      {error, State}
  end;

try_add_node_to_active_view(_Node, State) ->
  {error, State}.

maybe_add_node_passive_view(Node,
    State = #state{active_view = ActiveView, passive_view = PassiveView, passive_view_size = PassiveViewSize})
    when Node =/= node() ->
  {ActiveViewNew, PassiveViewNew} = case {ordsets:is_element(Node, ActiveView), ordsets:is_element(Node,
    PassiveView)} of
                                      {false, false} ->
                                        {PassiveView1, _} = trim_ordset_to(PassiveView, PassiveViewSize - 1),
                                        PassiveView2 = ordsets:add_element(Node, PassiveView1),
                                        {ActiveView, PassiveView2};
                                      _ ->
                                        {ActiveView, PassiveView}
                                    end,
  State#state{active_view = ActiveViewNew, passive_view = PassiveViewNew};
maybe_add_node_passive_view(_Node, State) ->
  State.

handle_forward_join(_ForwardJoin = #{ttl := 0, node := Node}, State) ->
  {_, State1} = try_add_node_to_active_view(Node, State),
  State1;
handle_forward_join(_ForwardJoin = #{node := Node}, State = #state{active_view = []}) ->
  {_, State1} = try_add_node_to_active_view(Node, State),
  State1;
handle_forward_join(ForwardJoin = #{node := Node, ttl := TTL}, State) ->
  PRWL = lashup_config:prwl(),
  State1 = case TTL of
             PRWL ->
               maybe_add_node_passive_view(Node, State);
             _ ->
               State
           end,
  forward_forward_join(ForwardJoin, State1),
  State1.

forward_forward_join(ForwardJoin = #{ttl := TTL, node := Node, seen := Seen},
    #state{active_view = ActiveView}) ->
  Seen1 = ordsets:add_element(node(), Seen),
  ForwardJoin1 = ForwardJoin#{ttl => TTL - 1, sender := node(), seen := Seen1},
  case ordsets:subtract(ActiveView, Seen) of
    [] ->
      ?LOG_WARNING("Forwarding join original node ~p dropped", [Node]);
    Nodes ->
      forward_forward_join1(ForwardJoin1, Nodes)
  end;
forward_forward_join(ForwardJoin = #{ttl := TTL, sender := Sender, node := Node}, #state{active_view = ActiveView}) ->
  ForwardJoin1 = ForwardJoin#{ttl => TTL - 1, sender := node()},
  case ordsets:del_element(Sender, ActiveView) of
    [] ->
      ?LOG_WARNING("Forwarding join original node ~p dropped", [Node]);
    Nodes ->
      forward_forward_join1(ForwardJoin1, Nodes)
  end.

forward_forward_join1(ForwardJoin, Nodes) when is_list(Nodes) ->
  Idx = rand:uniform(length(Nodes)),
  TargetNode = lists:nth(Idx, Nodes),
  cast(TargetNode, ForwardJoin).

%% TODO:
%% Maybe we should disconnect from the node at this point?
%% I'm unsure, because if another service is using hyparview for peer sampling
%% We could break the TCP connection before it's ready
-spec(handle_disconnect(disconnect(), state()) -> state()).
handle_disconnect(_Disconnect = #{sender := Sender},
    State = #state{active_view = ActiveView, passive_view = PassiveView, monitors = Monitors}) ->
  ?LOG_INFO("Node ~p received disconnect from ~p", [node(), Sender]),
  MonitorRef = node_to_monitor_ref(Sender, Monitors),
  Monitors1 = remove_monitor(MonitorRef, Monitors),
  ActiveView1 = ordsets:del_element(Sender, ActiveView),
  PassiveView1 = ordsets:add_element(Sender, PassiveView),
  schedule_disconnect(Sender),
  State#state{active_view = ActiveView1, passive_view = PassiveView1, monitors = Monitors1}.

handle_down_message(_DownMessage = {'DOWN', MonitorRef, process, _Info, Reason},
    State = #state{active_view = ActiveView, passive_view = PassiveView, monitors = Monitors}) ->
  case monitor_ref_to_node(MonitorRef, Monitors) of
    false ->
      State;
    Node ->
      ?LOG_INFO("Lost active neighbor: ~p because: ~p", [Node, Reason]),
      Monitors1 = remove_monitor(MonitorRef, Monitors),
      ActiveView1 = ordsets:del_element(Node, ActiveView),
      PassiveView1 = ordsets:add_element(Node, PassiveView),
      State#state{active_view = ActiveView1, passive_view = PassiveView1, monitors = Monitors1}
  end.

%% Maybe neighbor should only be called if we have a non-full active view
%% We filter for a full active view in the handle_info callback.
maybe_neighbor(State = #state{joined = false, active_view = []}) ->
  reschedule_maybe_neighbor(10000),
  State;
maybe_neighbor(State = #state{joined = false}) ->
  %% The active view has someone in it
  %% Probably better to mark myself as joined
  reschedule_maybe_neighbor(500),
  State#state{joined = true};
maybe_neighbor(State = #state{fixed_seed = FixedSeed, idx = Idx, unfilled_active_set_count = Count}) ->
  case {State#state.active_view, State#state.passive_view} of
    %% Both views are empty
    %% This shouldn't happen
    %% Hydrate the passive view with contact nodes, and reschedule immediately
    {[], []} ->
      reschedule_maybe_neighbor(500),
      ContactNodes = lashup_config:contact_nodes(),
      State#state{passive_view = ContactNodes};
    %% We have nodes in the active view, but none in the passive view
    %% This is concerning, but not necessarily bad
    %% Let's try to reconnect to the contact nodes
    {ActiveView, []} ->
      reschedule_maybe_neighbor(10000),
      ?LOG_DEBUG("Trying to connect to node from passive view, but passive view empty"),
      ContactNodes = lashup_config:contact_nodes(),
      UnconnectedContactNodes = ordsets:subtract(ContactNodes, ActiveView),
      State#state{passive_view = UnconnectedContactNodes};
    %% If we have nodes in the passive view, let's try to connect to them
    %% One difference between our implementation and the paper is that it evicts nodes from the passive view
    %% after they fail to be connected to
    %% Given our PassiveView is bounded, we just circle through that list
    {[], PassiveView} ->
      reschedule_maybe_neighbor(500),
      send_neighbor(500, PassiveView, high, Idx, FixedSeed),
      State#state{idx = Idx + 1, unfilled_active_set_count = Count + 1};
    {_ActiveView, PassiveView} ->
      reschedule_maybe_neighbor(2000),
      maybe_gm_neighbor(2000, State),
      send_neighbor(2000, PassiveView, low, Idx, FixedSeed),
      State#state{idx = Idx + 1, unfilled_active_set_count = Count + 1}
  end.

%% Timeout is actually a minimum time
maybe_gm_neighbor(Timeout, _State = #state{unfilled_active_set_count = Count, active_view_size = ActiveViewSize})
  when Count rem 5 == 0 andalso Count > 3 ->
  %% This is to ensure that there isn't a dependency loop between us and gm
  case catch lashup_gm:get_neighbor_recommendations(ActiveViewSize) of
    {ok, Node} ->
      ?LOG_INFO("Get Neighbors Successful: ~p", [Node]),
      send_neighbor_to(Timeout, Node, low),
      Node;
    Error ->
      ?LOG_INFO("Get Neighbors Unsuccessful: ~p", [Error]),
      ok
  end;
maybe_gm_neighbor(_Timeout, _State) ->
  ok.

send_neighbor(Timeout, PassiveView, Priority, Idx, FixedSeed) when length(PassiveView) > 0 ->
  ShuffledPassiveView = lashup_utils:shuffle_list(PassiveView, FixedSeed),
  RealIdx = Idx rem length(ShuffledPassiveView) + 1,
  Node = lists:nth(RealIdx, ShuffledPassiveView),
  send_neighbor_to(Timeout, Node, Priority),
  Node.

-spec(send_neighbor_to(Timeout :: non_neg_integer(), Node :: node(), Priority :: high | low) -> ok).
send_neighbor_to(Timeout, Node, Priority) when is_integer(Timeout) ->
  ?LOG_DEBUG("Sending neighbor to: ~p", [Node]),
  Ref = timer:send_after(Timeout * 3, {tried_neighbor, Node}),
  cast(Node, #{message => neighbor, ref => Ref, priority => Priority, sender => node()}).

-spec(handle_neighbor(neighbor(), state()) -> state()).
handle_neighbor(_Neighbor = #{priority := low, sender := Sender, ref := Ref},
    State = #state{active_view = ActiveView, active_view_size = ActiveViewSize})
    when length(ActiveView) == ActiveViewSize ->
  %% The Active neighbor list is full
  ?LOG_INFO("Denied neighbor request from ~p because active view full", [Sender]),
  PassiveView = State#state.passive_view,
  cast(Sender, #{message => neighbor_deny, sender => node(), ref => Ref, passive_view => PassiveView}),
  State;

%% Either this is a high priority request
%% Or I have an empty slot in my active view list
handle_neighbor(_Neighbor = #{sender := Sender, ref := Ref}, State) ->
  case try_add_node_to_active_view(Sender, State) of
    {ok, NewState} ->
      cast(Sender, #{message => neighbor_accept, sender => node(), ref => Ref}),
      NewState;
    {error, NewState = #state{passive_view = PassiveView}} ->
      ?LOG_WARNING("Failed to add neighbor ~p to active view on neighbor message", [Sender]),
      NeighborDeny = #{message => neighbor_deny, sender => node(), ref => Ref, passive_view => PassiveView},
      cast(Sender, NeighborDeny),
      NewState
  end.

-spec(handle_neighbor_accept(neighbor_accept(), state()) -> state()).
handle_neighbor_accept(_NeighborAccept = #{message := neighbor_accept, sender := Sender, ref := Ref}, State) ->
  timer:cancel(Ref),
  {_, State1} = try_add_node_to_active_view(Sender, State),
  push_state(State1),
  State1.

-spec(handle_neighbor_deny(neighbor_deny(), state()) -> state()).
handle_neighbor_deny(NeighborDeny = #{message := neighbor_deny, sender := Sender, ref := Ref}, State) ->
  timer:cancel(Ref),
  ActiveView = State#state.active_view,
  PassiveView = State#state.passive_view,
  ?LOG_DEBUG("Denied from joining ~p, while active view ~p, passive view: ~p", [Sender, ActiveView, PassiveView]),
  schedule_disconnect(Sender),
  State1 = neighbor_deny_combine_passive_view(NeighborDeny, State),
  push_state(State1),
  State1.

neighbor_deny_combine_passive_view(#{passive_view := RemotePassiveView}, State = #state{passive_view_size = PVS}) ->
  PassiveView = State#state.passive_view,
  ActiveView = State#state.active_view,
  LargeCombinedView = ordsets:union(RemotePassiveView, PassiveView),
  LargeCombinedView1 = ordsets:subtract(LargeCombinedView, ActiveView),
  LargeCombinedView2 = ordsets:del_element(node(), LargeCombinedView1),
  {NewPassiveView, _} = trim_ordset_to(LargeCombinedView2, PVS),
  State#state{passive_view = NewPassiveView};
neighbor_deny_combine_passive_view(_, State) ->
  State.

%% This is triggered by lashup_gm_probe to handle a permanently sectioned graph
%% We give it a one minute timeout, because there's no point in making it a small number.
%% We don't want to make it too high
-spec(handle_recommend_neighbor(Node :: node(), State :: state()) -> ok).
handle_recommend_neighbor(Node, _State) ->
  send_neighbor_to(60000, Node, high),
  ok.

%%%%%%%%%% End Neighbor management

%% Check State function is mostly there during testing
%% We should rip it out / disable it in prod
-spec(check_state(state()) -> state()).
check_state(State = #state{}) ->
  ok = check_views(State),
  ok = check_monitors(State),
  State.

-spec(check_monitors(state()) -> ok).
check_monitors(#state{active_view = ActiveView, monitors = Monitors}) ->
  %% Ensure we have a monitor_ref for every node in ActiveView
  CheckFun1 =
    fun(Node) ->
      case node_to_monitor_ref(Node, Monitors) of
        false ->
          ?LOG_ERROR("Node ~p lacks of a monitor reference: ~p", [Node, Monitors]),
          error({no_monitor_ref, Node});
        Ref when is_reference(Ref) ->
          ok
      end
    end,
  lists:foreach(CheckFun1, ActiveView),
  %% Ensure there aren't duplicate monitor refs
  MonitoredNodes = [Monitor#monitor.node || Monitor <- Monitors],
  SortedMonitoredNodes = lists:sort(MonitoredNodes),
  MonitoredNodesSet = ordsets:from_list(MonitoredNodes),
  case MonitoredNodesSet of
    SortedMonitoredNodes ->
      ok;
    _Other ->
      ?LOG_ERROR("Duplicate monitored nodes: ~p", [MonitoredNodes]),
      error(duplicate_monitored_nodes)
  end,
  %% Ensure there aren't excess monitors
  case MonitoredNodesSet == ActiveView of
    true ->
      ok;
    false ->
      error(mismatched_monitors)
  end,
  ok.

-spec(check_views(state()) -> ok).
check_views(_State = #state{active_view = ActiveView, passive_view = PassiveView}) ->
  case ordsets:intersection(PassiveView, ActiveView) of
    [] ->
      ok;
    Else ->
      error({overlapping, Else})
  end,
  case ordsets:is_element(node(), ActiveView) of
    true ->
      error(self_in_active_view);
    _ -> ok
  end,
  case ordsets:is_element(node(), PassiveView) of
    true ->
      error(self_in_passive_view);
    _ -> ok
  end,
  prometheus_gauge:set(
    lashup, hyparview_active_view_size, [],
    length(ActiveView)),
  prometheus_gauge:set(
    lashup, hyparview_passive_view_size, [],
    length(PassiveView)).

-spec(handle_shuffle(Shuffle :: shuffle(), state()) -> state()).
handle_shuffle(Shuffle = #{ttl := 0}, State) ->
  State1 = do_shuffle(Shuffle, State),
  push_state(State1),
  State1;
handle_shuffle(Shuffle = #{node := Node, sender := Sender, ttl := TTL}, State = #state{active_view = ActiveView})
  when Sender =/= node() andalso Node =/= node() ->
  PotentialNodes = ordsets:del_element(Node, ActiveView),
  PotentialNodes1 = ordsets:del_element(Sender, PotentialNodes),
  case PotentialNodes1 of
    [] ->
      ?LOG_DEBUG("Handling shuffle early, because no nodes to send it to"),
      State1 = do_shuffle(Shuffle, State),
      push_state(State1),
      State1;
    Else ->
      NewShuffle = Shuffle#{sender := node(), ttl := TTL - 1},
      NextNode = choose_node(Else),
      cast(NextNode, NewShuffle),
      State
  end.
do_shuffle(_Shuffle = #{active_view := RemoteActiveView, passive_view := RemotePassiveView, node := Node},
  State = #state{passive_view = MyPassiveView, active_view = MyActiveView, passive_view_size = PassiveViewSize}) ->
  ReplyNodes1 = ordsets:subtract(MyPassiveView, RemoteActiveView),
  ReplyNodes2 = ordsets:subtract(ReplyNodes1, RemotePassiveView),
  ShuffleReply = #{message => shuffle_reply, node => node(), combined_view => ReplyNodes2},
  cast(Node, ShuffleReply),
  schedule_disconnect(Node),

  LargeCombinedView = ordsets:union([MyPassiveView, RemoteActiveView, RemotePassiveView]),
  LargeCombinedView1 = ordsets:subtract(LargeCombinedView, MyActiveView),
  LargeCombinedView2 = ordsets:del_element(node(), LargeCombinedView1),
  {CombinedView, _} = trim_ordset_to(LargeCombinedView2, PassiveViewSize),
  State#state{passive_view = CombinedView}.


%% It returns how long to wait until to shuffle again
try_shuffle(_State = #state{active_view = []}) ->
  ?LOG_INFO("Could not shuffle because active view empty"),
  10000;
try_shuffle(_State = #state{active_view = ActiveView, passive_view = PassiveView, passive_view_size = PVS}) ->
  %% TODO:
  %% -Make TTL Configurable
  %% -Allow for limiting view sizes further
  Shuffle = #{
    message => shuffle,
    sender => node(),
    node => node(),
    active_view => ActiveView,
    passive_view => PassiveView,
    ttl => 5
  },
  Node = choose_node(ActiveView),
  cast(Node, Shuffle),
  case length(PassiveView) of
    Size when Size < 0.5 * PVS ->
      5000;
    _ ->
      60000
  end.

-spec(handle_shuffle_reply(shuffle_reply(), state()) -> state()).
handle_shuffle_reply(_ShuffleReply = #{combined_view := CombinedView},
    State = #state{passive_view = MyPassiveView, active_view = MyActiveView, passive_view_size = PassiveViewSize}) ->
  LargeCombinedView = ordsets:union(CombinedView, MyPassiveView),
  LargeCombinedView1 = ordsets:subtract(LargeCombinedView, MyActiveView),
  LargeCombinedView2 = ordsets:del_element(node(), LargeCombinedView1),
  {NewPassiveView, _} = trim_ordset_to(LargeCombinedView2, PassiveViewSize),
  State1 = State#state{passive_view = NewPassiveView},
  push_state(State1),
  State1.

ensure_monitor(Node, Monitors) ->
  case has_monitor(Node, Monitors) of
    false ->
      MonitorRef = monitor(process, {?MODULE, Node}),
      true = is_reference(MonitorRef),
      Monitor = #monitor{node = Node, monitor_ref = MonitorRef},
      lists:keystore(Node, #monitor.node, Monitors, Monitor);
    true ->
      Monitors
  end.

remove_monitor(MonitorRef, Monitors) when is_reference(MonitorRef) andalso is_list(Monitors) ->
  demonitor(MonitorRef),
  lists:keydelete(MonitorRef, #monitor.monitor_ref, Monitors);
remove_monitor(Node, Monitors) when is_atom(Node) andalso is_list(Monitors) ->
  case node_to_monitor_ref(Node, Monitors) of
    false ->
      Monitors;
    MonitorRef ->
      demonitor(MonitorRef),
      lists:keydelete(MonitorRef, #monitor.monitor_ref, Monitors)
  end.
node_to_monitor_ref(Node, Monitors) ->
  case lists:keyfind(Node, #monitor.node, Monitors) of
    false ->
      false;
    Monitor ->
      Monitor#monitor.monitor_ref
  end.

monitor_ref_to_node(Node, Monitors) ->
  case lists:keyfind(Node, #monitor.monitor_ref, Monitors) of
    false ->
      false;
    Monitor ->
      Monitor#monitor.node
  end.


has_monitor(Node, Monitors) ->
  lists:keymember(Node, #monitor.node, Monitors).


-spec(push_state(state()) -> ok).
push_state(#state{passive_view = PV, active_view = AV}) ->
  lashup_hyparview_events:ingest(AV, PV),
  ok.

shuffle_backoff_loop(Delay, Pid) ->
  timer:sleep(Delay),
  case catch gen_server:call(?MODULE, try_shuffle) of
    Backoff when is_integer(Backoff) ->
      shuffle_backoff_loop(Backoff, Pid);
    _ ->
      shuffle_backoff_loop(Delay, Pid)
  end.

handle_do_probe(Node, State = #state{active_view = ActiveViews}) ->
  case lists:member(Node, ActiveViews) of
    true ->
      lashup_hyparview_ping_handler:ping(Node),
      State;
    false ->
      State
  end.

handle_ping_rq(State = #state{active_view = ActiveViews}) when ActiveViews == [] ->
  reschedule_ping(),
  State;

handle_ping_rq(State = #state{active_view = ActiveViews, ping_idx = PingIdx}) ->
  reschedule_ping(),
  Idx = PingIdx rem length(ActiveViews) + 1,
  Node = lists:nth(Idx, ActiveViews),
  lashup_hyparview_ping_handler:ping(Node),
  State#state{ping_idx = PingIdx + 1}.

%% TODO: Determine whether we should disconnect_node
%% Or do something more 'clever'
-spec(handle_ping_failed(node(), state()) -> state()).
handle_ping_failed(Node, State) ->
  disconnect_node(Node, State).

%% Schedule two disconnects
%% The purpose of this is so that if we wrote some Erlang code that subscribes to the view
%% and doesn't only use erlang:send(..., [noconnect]), it'll result in reconnecting
%% This is fairly safe anyways, because we (try to) ensure that the node is not in our active view
%% when connecting

%% The only condition that this could be problematic is if there was something non-hyparview running
%% between the two nodes
%% TODO:
%% -Add a way to make connections exempt from disconnect
%% -Run a regular maintenance to make sure our connection count is small

%% The problem with lots of disterl connections is that:
%% (1) Net Ticks are awful, esp. if we get into a situation where we make a full mesh
%% (2) Run a regular maintenance and try to prune the nodes() in case something weird (tm) is happening
schedule_disconnect(Node) ->
  schedule_disconnect(Node, 25000),
  schedule_disconnect(Node, 50000).
schedule_disconnect(Node, Time) ->
  RandFloat = rand:uniform(),
  Multipler = 1 + round(RandFloat),
  Delay = Multipler * Time,
  timer:send_after(Delay, {maybe_disconnect, Node}).

-spec(disconnect_node(node(), state()) -> state()).
disconnect_node(Node, State = #state{monitors = Monitors, active_view = ActiveView, passive_view = PassiveView}) ->
  case lists:member(Node, ActiveView) of
    true ->
      Monitors1 = remove_monitor(Node, Monitors),
      DisconnectMessage = #{message => disconnect, sender => node()},
      case lists:member(Node, nodes()) of
        true ->
          cast(Node, DisconnectMessage);
        false ->
          ok
      end,
      PassiveView1 = ordsets:add_element(Node, PassiveView),
      ActiveView1 = ordsets:del_element(Node, ActiveView),
      schedule_disconnect(Node),
      State#state{active_view = ActiveView1, passive_view = PassiveView1, monitors = Monitors1};
    false ->
      State
  end.

handle_maybe_disconnect(Node, State) ->
  handle_maybe_disconnect1(Node, State).

%% Is this node part of exempt nodes?
handle_maybe_disconnect1(Node, State) ->
  ExemptNodes = application:get_env(lashup, exempt_nodes, []),
  case lists:member(Node, ExemptNodes) of
    true ->
      State;
    false ->
      handle_maybe_disconnect2(Node, State)
  end.

%% Is this node not part of my active view
handle_maybe_disconnect2(Node, State = #state{active_view = ActiveView}) ->
  case lists:member(Node, ActiveView) of
    true ->
      State;
    false ->
      handle_maybe_disconnect3(Node, State)
  end.

%% Is this node even connected?
handle_maybe_disconnect3(Node, State) ->
  case lists:member(Node, nodes()) of
    true ->
      erlang:disconnect_node(Node),
      State;
    false ->
      State
  end.

handle_recognize_pong(_Pong = #{now := _Now, receiving_node := Node},
  _State = #state{active_view = ActiveView, passive_view = PassiveView}) ->
  case {lists:member(Node, ActiveView), lists:member(Node, PassiveView)} of
    {true, false} ->
      %% TODO:
      %% Record successful roundtrip metric
      ok;
    {false, true} ->
      %% Something kinda weird has happened
      %% This can reasonably happen from late pongs
      %% TODO: Record it
      ?LOG_INFO("Late Pong from Node: ~p, already moved to passive view", [Node]);
    {false, false} ->
      %% This is bad
      ?LOG_WARNING("Pong from unknown node: ~p, not in active nor passive veiws", [Node])
  end.
  %% {true, true} should never happen. If it does, we _should_ crash.

-spec(cast(node(), hyparview_message()) -> ok).
cast(Node, #{message := Type} = Message) ->
  prometheus_counter:inc(lashup, hyparview_outgoing_messages_total, [Type], 1),
  gen_server:cast({?MODULE, Node}, Message).

-spec(abcast([node()], hyparview_message()) -> ok).
abcast(Nodes, Message) ->
  lists:foreach(fun (Node) -> cast(Node, Message) end, Nodes).

%%%===================================================================
%%% Metrics functions
%%%===================================================================

-spec(init_metrics() -> ok).
init_metrics() ->
  prometheus_gauge:new([
    {registry, lashup},
    {name, hyparview_active_view_size},
    {help, "The size of HyParView Active View."}
  ]),
  prometheus_gauge:new([
    {registry, lashup},
    {name, hyparview_passive_view_size},
    {help, "The size of HyParView Passive View."}
  ]),
  prometheus_counter:new([
    {registry, lashup},
    {name, hyparview_outgoing_messages_total},
    {labels, [type]},
    {help, "Total number of HyParView messages sent to other nodes."}
  ]),
  prometheus_summary:new([
    {registry, lashup},
    {name, hyparview_incoming_message_processing_seconds},
    {labels, [type]},
    {duration_unit, seconds},
    {help, "The time spent processing HyParView messages from other nodes."}
  ]),
  prometheus_gauge:new([
    {registry, lashup},
    {name, hyparview_message_queue_length},
    {help, "The length of HyParView process message box."}
  ]).

%%%===================================================================
%%% Test functions
%%%===================================================================

-ifdef(TEST).
trim_test() ->
  Ordset = ordsets:from_list([1, 3, 4, 5]),
  {Ordset1, _} = trim_ordset_to(Ordset, 3),
  ?assertEqual(3, length(Ordset1)).
-endif.