eredis_streams

-module(eredis_streams_worker).

-behaviour(gen_server).

%% API

-export([start_link/1]).

%% 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
%% @end
%%--------------------------------------------------------------------

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

%%%===================================================================
%%% 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),
      worker_num => worker_num(Params),
      xgroup => xgroup(Params),
      process_name => xgroup_process_name(xgroup(Params)),
      prefetch_count => prefetch_count(Params),
      fetch_interval => fetch_interval_ms(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,
    worker_num := WorkerNum,
    opts := #{workers := #{module := Handler}}
  } = State
) ->
  ?EXEC_TELEMETRY(?EREDIS_STREAMS_WORKER_INIT, #{}, #{stream => Stream, worker_num => WorkerNum}),
  case eredis_streams_db:init_xgroup(Pool, Stream, Xgroup) of
    ok ->
      HandlerModule = maps:get(name, Handler),
      HandlerArgs = maps:get(args, Handler, #{}),
      HandlerModule:init(Handler#{args => HandlerArgs#{worker_num => WorkerNum}}),
      erlang:send(self(), bootstrap_cluster),
      {noreply, State#{status => wait_for_cluster}};

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

handle_info(
  bootstrap_cluster,
  #{
    status := wait_for_cluster,
    stream := Stream,
    worker_num := WorkerNum,
    opts := #{workers := #{singleton := true}}
  } = State
) ->
  ?EXEC_TELEMETRY(
    ?EREDIS_STREAMS_WORKER_READY,
    #{},
    #{stream => Stream, worker_num => WorkerNum, sigleton => true}
  ),
  {noreply, update_master(State)};

handle_info(bootstrap_cluster, #{stream := Stream, worker_num := WorkerNum} = State) ->
  ?EXEC_TELEMETRY(
    ?EREDIS_STREAMS_WORKER_READY,
    #{},
    #{stream => Stream, worker_num => WorkerNum, sigleton => false}
  ),
  erlang:send(self(), tick),
  {noreply, State#{master => true, status => ready}};

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

handle_info(
  tick,
  #{
    status := ready,
    master := true,
    worker_num := WorkerNum,
    stream := Stream,
    pool := Pool,
    xgroup := Group,
    process_name := Process,
    prefetch_count := Count,
    fetch_interval := FetchInterval,
    opts := #{workers := #{module := Handler}}
  } = State
) ->
  HandlerModule = maps:get(name, Handler),
  HandlerArgs = maps:get(args, Handler, #{}),
  StartTime = eredis_streams_tools:now_mcs(),
  Stats =
    case
    catch
    eredis_streams_db:stream_fetch_messages(Pool, Stream, Group, Process, Count, FetchInterval) of
      [] -> #{};

      [_ | _] = MsgList ->
        HandlerModule:handle_messages(
          eredis_streams_tools:build_conn(Pool, Stream, Group, #{}),
          MsgList,
          HandlerArgs#{worker_num => WorkerNum}
        );

      Err ->
        ?EXEC_TELEMETRY(?EREDIS_STREAMS_ERROR, #{}, #{reason => Err}),
        #{}
    end,
  ?EXEC_TELEMETRY(
    ?EREDIS_STREAMS_WORKER_TICK,
    #{total_msc => eredis_streams_tools:now_mcs() - StartTime, stats => Stats},
    #{stream => Stream, worker_num => WorkerNum}
  ),
  erlang:send_after(FetchInterval, self(), tick),
  {noreply, State};

handle_info(tick, State) ->
  % не генерируем новое событие tick, так как tick перезапустится в update_master
  {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) ->
  ?EXEC_TELEMETRY(?EREDIS_STREAMS_WORKER_TERMINATE, #{}, #{}),
  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,
    opts := #{workers := WorkersOpts},
    master := IsCurMaster
  } = State
) ->
  MasterTimeoutMs = maps:get(singleton_timeout_ms, WorkersOpts, 3000),
  case eredis_streams_db:is_master(Pool, Stream, Wid, MasterTimeoutMs) of
    false ->
      erlang:send_after(300, self(), update_master),
      State#{master => false, status => ready};

    true ->
      case IsCurMaster of
        Nope when Nope =:= false orelse Nope =:= undefined -> erlang:send(self(), tick);
        _ -> skip
      end,
      erlang:send_after(300, self(), update_master),
      State#{master => true, status => ready}
  end.


process_name(StreamName, Num) ->
  binary_to_atom(
    iolist_to_binary([["eredis_streams_worker_", StreamName, "_", integer_to_list(Num), "_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()))]).

prefetch_count(Params) ->
  Workers = maps:get(workers, Params, #{}),
  maps:get(prefetch_count, Workers, 1).


fetch_interval_ms(Params) ->
  Workers = maps:get(workers, Params, #{}),
  maps:get(fetch_interval_ms, Workers, 450).


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

worker_num(#{worker_num := Num} = _Params) -> Num;
worker_num(_) -> 1.