% This file is licensed to you under the Apache License,
% Version 2.0 (the "License"); you may not use this file
% except in compliance with the License.  You may obtain
% a copy of the License at
%
% https://www.apache.org/licenses/LICENSE-2.0
%
% Unless required by applicable law or agreed to in writing,
% software distributed under the License is distributed on an
% "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
% KIND, either express or implied.  See the License for the
% specific language governing permissions and limitations
% under the License.
%%% @private
%%% @doc Decorator over `gen_server' that lets `wpool_pool'
%%%      control certain aspects of the execution
-module(wpool_process).

-behaviour(gen_server).

%% Taken from gen_server OTP
-record(callback_cache, {
    module :: module(),
    handle_call ::
        fun(
            (Request :: term(), From :: gen_server:from(), State :: term()) ->
                {reply, Reply :: term(), NewState :: term()}
                | {reply, Reply :: term(), NewState :: term(),
                    timeout() | hibernate | {continue, term()}}
                | {noreply, NewState :: term()}
                | {noreply, NewState :: term(), timeout() | hibernate | {continue, term()}}
                | {stop, Reason :: term(), Reply :: term(), NewState :: term()}
                | {stop, Reason :: term(), NewState :: term()}
        ),
    handle_cast ::
        fun(
            (Request :: term(), State :: term()) ->
                {noreply, NewState :: term()}
                | {noreply, NewState :: term(), timeout() | hibernate | {continue, term()}}
                | {stop, Reason :: term(), NewState :: term()}
        ),
    handle_info ::
        fun(
            (Info :: timeout | term(), State :: term()) ->
                {noreply, NewState :: term()}
                | {noreply, NewState :: term(), timeout() | hibernate | {continue, term()}}
                | {stop, Reason :: term(), NewState :: term()}
        )
}).
-record(state, {
    name :: atom(),
    mod :: #callback_cache{},
    state :: term(),
    options :: opts()
}).

-type opts() :: #{
    time_checker := atom(),
    queue_manager := atom(),
    event_manager := atom(),
    overrun_warning := timeout(),
    _ => _
}.

-opaque state() :: #state{}.

-export_type([state/0]).

-type next_step() :: timeout() | hibernate | {continue, term()}.

-export_type([next_step/0]).

%% api
-export([start_link/4, run/3, call/3, cast/2, send_request/2]).

-ifdef(TEST).

-export([get_state/1]).

-endif.

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

%%%===================================================================
%%% API
%%%===================================================================
%% @doc Starts a named process
-spec start_link(wpool:name(), module(), term(), wpool:options()) ->
    {ok, pid()} | ignore | {error, {already_started, pid()} | term()}.
start_link(Name, Module, InitArgs, Options) ->
    FullOpts = wpool_utils:add_defaults(Options),
    WorkerOpt = maps:get(worker_opt, FullOpts, []),
    gen_server:start_link(
        {local, Name},
        ?MODULE,
        {Name, Module, InitArgs, FullOpts},
        WorkerOpt
    ).

%% @doc Runs a function that takes as a parameter the given process
-spec run(wpool:name() | pid(), wpool:run(Result), timeout()) -> Result.
run(Process, Run, Timeout) ->
    Run(Process, Timeout).

%% @equiv gen_server:call(Process, Call, Timeout)
-spec call(wpool:name() | pid(), term(), timeout()) -> term().
call(Process, Call, Timeout) ->
    gen_server:call(Process, Call, Timeout).

%% @equiv gen_server:cast(Process, {cast, Cast})
-spec cast(wpool:name() | pid(), term()) -> ok.
cast(Process, Cast) ->
    gen_server:cast(Process, Cast).

%% @equiv gen_server:send_request(Process, Request)
-spec send_request(wpool:name() | pid(), term()) -> gen_server:request_id().
send_request(Name, Request) ->
    gen_server:send_request(Name, Request).

-ifdef(TEST).

-spec get_state(state()) -> term().
get_state(#state{state = State}) ->
    State.

-endif.

%%%===================================================================
%%% init, terminate, code_change, info callbacks
%%%===================================================================
%% @private
-spec init({atom(), atom(), term(), opts()}) ->
    {ok, state()} | {ok, state(), next_step()} | {stop, can_not_ignore} | {stop, term()}.
init({Name, Mod, InitArgs, Options}) ->
    #{event_manager := EventManager, queue_manager := QueueManager} = Options,
    wpool_process_callbacks:notify(handle_init_start, EventManager, [Name]),
    CbCache = create_callback_cache(Mod),
    case Mod:init(InitArgs) of
        {ok, ModState} ->
            ok = notify_queue_manager(new_worker, Name, QueueManager),
            wpool_process_callbacks:notify(handle_worker_creation, EventManager, [Name]),
            {ok, #state{
                name = Name,
                mod = CbCache,
                state = ModState,
                options = Options
            }};
        {ok, ModState, NextStep} ->
            ok = notify_queue_manager(new_worker, Name, QueueManager),
            wpool_process_callbacks:notify(handle_worker_creation, EventManager, [Name]),
            {ok,
                #state{
                    name = Name,
                    mod = CbCache,
                    state = ModState,
                    options = Options
                },
                NextStep};
        ignore ->
            {stop, can_not_ignore};
        Error ->
            Error
    end.

%% @private
-spec terminate(atom(), state()) -> term().
terminate(Reason, State) ->
    #state{
        mod = #callback_cache{module = Mod},
        state = ModState,
        name = Name,
        options = #{event_manager := EventManager, queue_manager := QueueManager}
    } = State,
    ok = notify_queue_manager(worker_dead, Name, QueueManager),
    wpool_process_callbacks:notify(handle_worker_death, EventManager, [Name, Reason]),
    case erlang:function_exported(Mod, terminate, 2) of
        true ->
            Mod:terminate(Reason, ModState);
        _ ->
            ok
    end.

%% @private
-spec code_change(string() | {down, string()}, state(), any()) ->
    {ok, state()} | {error, term()}.
code_change(OldVsn, #state{mod = #callback_cache{module = Mod}} = State, Extra) ->
    case erlang:function_exported(Mod, code_change, 3) of
        true ->
            case Mod:code_change(OldVsn, State#state.state, Extra) of
                {ok, NewState} ->
                    {ok, State#state{state = NewState}};
                {error, Error} ->
                    {error, Error}
            end;
        _ ->
            {ok, State}
    end.

%% @private
-spec handle_info(any(), state()) ->
    {noreply, state()} | {noreply, state(), next_step()} | {stop, term(), state()}.
handle_info(Info, #state{mod = CbCache} = State) ->
    #callback_cache{module = Mod, handle_info = HandleInfo} = CbCache,
    try HandleInfo(Info, State#state.state) of
        {noreply, NewState} ->
            {noreply, State#state{state = NewState}};
        {noreply, NewState, NextStep} ->
            {noreply, State#state{state = NewState}, NextStep};
        {stop, Reason, NewState} ->
            {stop, Reason, State#state{state = NewState}}
    catch
        error:undef:Stacktrace ->
            case erlang:function_exported(Mod, handle_info, 2) of
                false ->
                    {noreply, State};
                true ->
                    erlang:raise(error, undef, Stacktrace)
            end;
        _:{noreply, NewState} ->
            {noreply, State#state{state = NewState}};
        _:{noreply, NewState, NextStep} ->
            {noreply, State#state{state = NewState}, NextStep};
        _:{stop, Reason, NewState} ->
            {stop, Reason, State#state{state = NewState}}
    end.

%% @private
-spec handle_continue(any(), state()) ->
    {noreply, state()}
    | {noreply, state(), next_step()}
    | {stop, term(), state()}.
handle_continue(Continue, #state{mod = #callback_cache{module = Mod}} = State) ->
    try Mod:handle_continue(Continue, State#state.state) of
        {noreply, NewState} ->
            {noreply, State#state{state = NewState}};
        {noreply, NewState, NextStep} ->
            {noreply, State#state{state = NewState}, NextStep};
        {stop, Reason, NewState} ->
            {stop, Reason, State#state{state = NewState}}
    catch
        error:undef:Stacktrace ->
            case erlang:function_exported(Mod, handle_continue, 2) of
                false ->
                    {noreply, State};
                true ->
                    erlang:raise(error, undef, Stacktrace)
            end;
        _:{noreply, NewState} ->
            {noreply, State#state{state = NewState}};
        _:{noreply, NewState, NextStep} ->
            {noreply, State#state{state = NewState}, NextStep};
        _:{stop, Reason, NewState} ->
            {stop, Reason, State#state{state = NewState}}
    end.

%% @private
-spec format_status(gen_server:format_status()) -> gen_server:format_status().
format_status(#{state := #state{mod = #callback_cache{module = Mod}}} = Status) ->
    case erlang:function_exported(Mod, format_status, 1) of
        false ->
            Status;
        true ->
            Mod:format_status(Status)
    end.

%%%===================================================================
%%% real (i.e. interesting) callbacks
%%%===================================================================
%% @private
-spec handle_cast(term(), state()) ->
    {noreply, state()} | {noreply, state(), next_step()} | {stop, term(), state()}.
handle_cast(Cast, #state{mod = CbCache, options = Options} = State) ->
    #callback_cache{handle_cast = HandleCast} = CbCache,
    #{overrun_warning := OverrunWarning, queue_manager := QueueManager} = Options,
    Task = task_init(OverrunWarning, {cast, Cast}, Options),
    ok = notify_queue_manager(worker_busy, State#state.name, QueueManager),
    Reply =
        try HandleCast(Cast, State#state.state) of
            {noreply, NewState} ->
                {noreply, State#state{state = NewState}};
            {noreply, NewState, NextStep} ->
                {noreply, State#state{state = NewState}, NextStep};
            {stop, Reason, NewState} ->
                {stop, Reason, State#state{state = NewState}}
        catch
            _:{noreply, NewState} ->
                {noreply, State#state{state = NewState}};
            _:{noreply, NewState, NextStep} ->
                {noreply, State#state{state = NewState}, NextStep};
            _:{stop, Reason, NewState} ->
                {stop, Reason, State#state{state = NewState}}
        end,
    task_end(Task),
    ok = notify_queue_manager(worker_ready, State#state.name, QueueManager),
    Reply.

%% @private
-spec handle_call(term(), gen_server:from(), state()) ->
    {reply, term(), state()}
    | {reply, term(), state(), next_step()}
    | {noreply, state()}
    | {noreply, state(), next_step()}
    | {stop, term(), term(), state()}
    | {stop, term(), state()}.
handle_call(Call, From, #state{mod = CbCache, options = Options} = State) ->
    #callback_cache{handle_call = HandleCall} = CbCache,
    #{overrun_warning := OverrunWarning, queue_manager := QueueManager} = Options,
    Task = task_init(OverrunWarning, {call, Call}, Options),
    ok = notify_queue_manager(worker_busy, State#state.name, QueueManager),
    Reply =
        try HandleCall(Call, From, State#state.state) of
            {noreply, NewState} ->
                {noreply, State#state{state = NewState}};
            {noreply, NewState, NextStep} ->
                {noreply, State#state{state = NewState}, NextStep};
            {reply, Response, NewState} ->
                {reply, Response, State#state{state = NewState}};
            {reply, Response, NewState, NextStep} ->
                {reply, Response, State#state{state = NewState}, NextStep};
            {stop, Reason, NewState} ->
                {stop, Reason, State#state{state = NewState}};
            {stop, Reason, Response, NewState} ->
                {stop, Reason, Response, State#state{state = NewState}}
        catch
            _:{noreply, NewState} ->
                {noreply, State#state{state = NewState}};
            _:{noreply, NewState, NextStep} ->
                {noreply, State#state{state = NewState}, NextStep};
            _:{reply, Response, NewState} ->
                {reply, Response, State#state{state = NewState}};
            _:{reply, Response, NewState, NextStep} ->
                {reply, Response, State#state{state = NewState}, NextStep};
            _:{stop, Reason, NewState} ->
                {stop, Reason, State#state{state = NewState}};
            _:{stop, Reason, Response, NewState} ->
                {stop, Reason, Response, State#state{state = NewState}}
        end,
    task_end(Task),
    ok = notify_queue_manager(worker_ready, State#state.name, QueueManager),
    Reply.

notify_queue_manager(_, _, undefined) ->
    ok;
notify_queue_manager(worker_busy, Name, QueueManager) ->
    wpool_queue_manager:worker_busy(QueueManager, Name);
notify_queue_manager(worker_ready, Name, QueueManager) ->
    wpool_queue_manager:worker_ready(QueueManager, Name);
notify_queue_manager(worker_dead, Name, QueueManager) ->
    wpool_queue_manager:worker_dead(QueueManager, Name);
notify_queue_manager(new_worker, Name, QueueManager) ->
    wpool_queue_manager:new_worker(QueueManager, Name).

task_init(infinity, Task, _) ->
    Time = erlang:system_time(),
    erlang:put(wpool_task, {undefined, Time, Task}),
    undefined;
task_init(OverrunTime, Task, #{time_checker := TimeChecker, max_overrun_warnings := MaxWarnings}) ->
    TaskId = erlang:make_ref(),
    Time = erlang:system_time(),
    erlang:put(wpool_task, {TaskId, Time, Task}),
    erlang:send_after(
        OverrunTime,
        TimeChecker,
        {check, self(), TaskId, OverrunTime, MaxWarnings}
    ).

task_end(undefined) ->
    erlang:put(wpool_task, undefined);
task_end(TimerRef) ->
    _ = erlang:cancel_timer(TimerRef, [{async, true}, {info, false}]),
    erlang:put(wpool_task, undefined).

create_callback_cache(Mod) ->
    #callback_cache{
        module = Mod,
        handle_call = fun Mod:handle_call/3,
        handle_cast = fun Mod:handle_cast/2,
        handle_info = fun Mod:handle_info/2
    }.