eredis_streams

-module(eredis_streams_failover).

-behaviour(gen_server).

-define(EMPTY_QUEUE_SLEEP, 750).

%% API

-export([start_link/1, failover_timeout/0]).

-define(FAILOVER_TIMEOUT_MS, 10000).

%% gen_server callbacks

-include_lib("eredis_streams/include/telemetry.hrl").

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

%%--------------------------------------------------------------------
%% @doc
%% Starts producer failover
%% @end
%%--------------------------------------------------------------------

start_link(#{name := StreamName} = Params) ->
  gen_server:start_link({local, process_name(StreamName)}, ?MODULE, Params, []).

%%--------------------------------------------------------------------
%% @doc
%% Failover timeout
%% @end
%%--------------------------------------------------------------------

failover_timeout() -> ?FAILOVER_TIMEOUT_MS.

%%%===================================================================
%%% gen_server callbacks
%%%===================================================================
%%--------------------------------------------------------------------
%% @private
%% @doc
%% Initializes the server
%% @end
%%--------------------------------------------------------------------

init(Params) ->
  process_flag(trap_exit, true),
  erlang:send(self(), init),
  {
    ok,
    #{
      wid => new_wid(),
      stream => stream_name(Params),
      pool => stream_pool(Params),
      xgroup => xgroup(Params),
      process_name => xgroup_process_name(xgroup(Params)),
      master => undefined,
      status => wait_for_init,
      opts => Params
    }
  }.

%%--------------------------------------------------------------------
%% @private
%% @doc
%% Handling call messages
%% @end
%%--------------------------------------------------------------------

handle_call(_Request, _From, State) ->
  Reply = ok,
  {reply, Reply, State}.

%%--------------------------------------------------------------------
%% @private
%% @doc
%% Handling cast messages
%% @end
%%--------------------------------------------------------------------

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

%%--------------------------------------------------------------------
%% @private
%% @doc
%% Handling all non call/cast messages
%% @end
%%--------------------------------------------------------------------

handle_info(
  init,
  #{status := wait_for_init, xgroup := Xgroup, pool := Pool, stream := Stream} = State
) ->
  ?EXEC_TELEMETRY(?EREDIS_STREAMS_FAILOVER_INIT, #{}, #{stream => Stream}),
  case eredis_streams_db:init_xgroup(Pool, Stream, Xgroup) of
    ok ->
      ?EXEC_TELEMETRY(?EREDIS_STREAMS_FAILOVER_READY, #{}, #{stream => Stream}),
      erlang:send(self(), tick),
      erlang:send(self(), update_master),
      {noreply, State#{status => ready}};

    Err ->
      erlang:send_after(100, self(), init),
      ?EXEC_TELEMETRY(?EREDIS_STREAMS_ERROR, #{}, #{error => Err}),
      {noreply, State}
  end;

handle_info(update_master, State) -> {noreply, update_master(State)};

handle_info(
  tick,
  #{status := ready, master := true, pool := Pool, stream := Stream, xgroup := Group} = State
) ->
  StartTime = eredis_streams_tools:now_mcs(),
  Stats =
    case catch eredis_streams_db:stream_fetch_pending_messages(Pool, Stream, Group, 1000) of
      [] -> #{};
      [_ | _] = MsgList -> do_failover(State, MsgList, #{processed => 0, skipped => 0});

      Err ->
        ?EXEC_TELEMETRY(?EREDIS_STREAMS_ERROR, #{}, #{reason => Err}),
        #{}
    end,
  Processed = maps:get(processed, Stats, 0),
  case Processed of
    0 -> erlang:send_after(?EMPTY_QUEUE_SLEEP, self(), tick);
    _ -> erlang:send(self(), tick)
  end,
  ?EXEC_TELEMETRY(
    ?EREDIS_STREAMS_FAILOVER_TICK,
    #{total_msc => eredis_streams_tools:now_mcs() - StartTime, stats => Stats},
    #{stream => Stream}
  ),
  {noreply, State};

handle_info(tick, State) ->
  erlang:send_after(?EMPTY_QUEUE_SLEEP, self(), tick),
  {noreply, State};

handle_info(Info, #{opts := #{workers := #{module := #{name := Handler}}}} = State) ->
  Handler:handle_info(Info, State);

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

%%--------------------------------------------------------------------
%% @private
%% @doc
%% This function is called by a gen_server when it is about to
%% terminate. It should be the opposite of Module:init/1 and do any
%% necessary cleaning up. When it returns, the gen_server terminates
%% with Reason. The return value is ignored.
%% @end
%%--------------------------------------------------------------------

terminate(_Reason, _State) -> ok.

%%--------------------------------------------------------------------
%% @private
%% @doc
%% Convert process state when code is changed
%% @end
%%--------------------------------------------------------------------

code_change(_OldVsn, State, _Extra) -> {ok, State}.

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

update_master(#{pool := Pool, stream := Stream, wid := Wid} = State) ->
  MasterTimeoutMs = 3000,
  NewState =
    case eredis_streams_db:is_failover_master(Pool, Stream, Wid, MasterTimeoutMs) of
      false -> State#{master => false, status => ready};
      true -> State#{master => true, status => ready}
    end,
  erlang:send_after(300, self(), update_master),
  NewState.


do_failover(_State, [], Acc) -> Acc;

do_failover(
  #{
    pool := Pool,
    stream := Stream,
    xgroup := Group,
    process_name := ProcessName,
    opts := #{workers := #{module := Handler}} = Params
  } = State,
  [#{msg_id := MsgId, time_ms := TimeInQueueMs} = Msg | MsgList],
  #{processed := Processed, skipped := Skipped} = Acc
) ->
  PendingTimeout = eredis_streams_tools:failover_pending_timeout(Params),
  NewAcc =
    case PendingTimeout =< TimeInQueueMs of
      true ->
        case
        eredis_streams_db:xclaim_message(Pool, Stream, Group, MsgId, ProcessName, PendingTimeout) of
          [XClaimedMsg | _] ->
            HandlerModule = maps:get(name, Handler),
            HandlerArgs = maps:get(args, Handler, #{}),
            HandlerModule:handle_failover(
              eredis_streams_tools:build_conn(Pool, Stream, Group, #{}),
              [XClaimedMsg],
              HandlerArgs#{failover_meta => Msg, worker_num => 1}
            ),
            ?EXEC_TELEMETRY(
              ?EREDIS_STREAMS_FAILOVER_HANDLE_MESSAGE,
              #{},
              #{stream => Stream, group => Group, message => XClaimedMsg, args => HandlerArgs}
            ),
            Acc#{processed => Processed + 1};

          _Else -> Acc#{skipped => Skipped + 1}
        end;

      false -> Acc#{skipped => Skipped + 1}
    end,
  do_failover(State, MsgList, NewAcc).


process_name(StreamName) ->
  binary_to_atom(iolist_to_binary([["eredis_streams_failover_", StreamName, "_sup"]]), utf8).

stream_name(#{name := Name} = _Params) -> Name.

new_wid() -> integer_to_binary(erlang:phash2({self(), eredis_streams_tools:now_mcs()})).

xgroup(#{xgroup := Xgroup} = _Config) -> Xgroup;
xgroup(_) -> "g_default".

xgroup_process_name(Xgroup) -> lists:flatten([Xgroup, "-", integer_to_list(erlang:phash2(node()))]).

stream_pool(#{pool := Pool} = _Params) -> Pool;
stream_pool(_Params) -> {error, undefined_pool}.