%%%-------------------------------------------------------------------
%%% @author anlicheng
%%% @copyright (C) 2026, <COMPANY>
%%% @doc
%%%
%%% @end
%%% Created : 20. 4月 2026
%%%-------------------------------------------------------------------
-module(efka_task_reporter).
-author("anlicheng").

-behaviour(gen_server).

%% API
-export([start_link/0]).
-export([stream/3, close/2]).

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

-define(SERVER, ?MODULE).
-define(FLUSH_INTERVAL, 1000).

-record(state, {
    pending = queue:new(),
    flush_ref = undefined
}).

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

-spec stream(TaskId :: integer(), Type :: binary(), Stream :: binary()) -> ok.
stream(TaskId, Type, Stream) when is_integer(TaskId), is_binary(Type), is_binary(Stream) ->
    gen_server:cast(?SERVER, {stream, TaskId, Type, Stream}).

-spec close(TaskId :: integer(), Reason :: binary()) -> ok.
close(TaskId, Reason) when is_integer(TaskId), is_binary(Reason) ->
    gen_server:cast(?SERVER, {close, TaskId, Reason}).

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

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

init([]) ->
    {ok, #state{}}.

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

handle_cast({stream, TaskId, Type, Stream}, State0 = #state{pending = Pending0}) ->
    Pending = queue:in({stream, TaskId, Type, Stream}, Pending0),
    {noreply, ensure_flush_timer(State0#state{pending = Pending}, 0)};
handle_cast({close, TaskId, Reason}, State0 = #state{pending = Pending0}) ->
    Pending = queue:in({close, TaskId, Reason}, Pending0),
    {noreply, ensure_flush_timer(State0#state{pending = Pending}, 0)};
handle_cast(_Request, State) ->
    {noreply, State}.

handle_info(flush, State0 = #state{}) ->
    State1 = State0#state{flush_ref = undefined},
    {noreply, flush_pending(State1)};
handle_info(_Info, State) ->
    {noreply, State}.

terminate(_Reason, _State) ->
    ok.

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

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

ensure_flush_timer(State = #state{flush_ref = undefined}, Delay) when is_integer(Delay), Delay >= 0 ->
    Ref = erlang:send_after(Delay, self(), flush),
    State#state{flush_ref = Ref};
ensure_flush_timer(State = #state{}, _Delay) ->
    State.

flush_pending(State = #state{pending = Pending0}) ->
    case queue:out(Pending0) of
        {empty, _} ->
            State;
        {{value, Event}, Pending1} ->
            case send_event(Event) of
                ok ->
                    flush_pending(State#state{pending = Pending1});
                not_ready ->
                    ensure_flush_timer(State#state{pending = Pending0}, ?FLUSH_INTERVAL)
            end
    end.

send_event({stream, TaskId, Type, Stream}) ->
    send_result(catch efka_client:send_task_event_stream(TaskId, Type, Stream));
send_event({close, TaskId, Reason}) ->
    send_result(catch efka_client:send_close_task_event_stream(TaskId, Reason)).

send_result(ok) ->
    ok;
send_result(_) ->
    not_ready.