khepri

%% This Source Code Form is subject to the terms of the Mozilla Public
%% License, v. 2.0. If a copy of the MPL was not distributed with this
%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
%%
%% Copyright © 2022-2026 Broadcom. All Rights Reserved. The term "Broadcom"
%% refers to Broadcom Inc. and/or its subsidiaries.
%%

%% @hidden

-module(khepri_event_handler).
-behaviour(gen_server).

-include_lib("kernel/include/logger.hrl").
-include_lib("stdlib/include/assert.hrl").

-include("include/khepri.hrl").
-include("src/khepri_evf.hrl").
-include("src/khepri_machine.hrl").

-export([start_link/0,
         handle_triggered_actions/2]).

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

-type trigger_action() :: {sproc, khepri_path:native_path()}.
%% The type of action associated with a trigger.
%%
%% It must be the path to a stored procedure.
%%
%% The action will get a trigger desccriptor as argument to describe a specific
%% execution of that trigger. See {@link trigger_descriptor/0}.

-type triggered_action() :: {sproc, horus:horus_fun()}.
%% The action associated with a trigger once it is triggered.
%%
%% It is the stored procedure pointed to by the path in the triggered action's
%% arguments.

-type trigger_exec_loc() :: leader.
%% Where to execute the triggered action.
%%
%% It supports the following locations:
%% <ul>
%% <li>`leader': the action is executed on the leader node at the time the
%% action is triggered.</li>
%% </ul>

-type trigger_descriptor() :: #khepri_trigger{}.
%% Descriptor of the trigger used as an argument to the trigger action.
%%
%% The descriptor contains all the properties of a specific instance of that
%% trigger.
%%
%% When the action is a stored procedure or an MFA tuple, the descriptor is
%% appended to the list of arguments.
%%
%% When the action is a PID, the descriptor is send to the PID as a standalone
%% message. The `action' properties map will contain the `Priv' element of the
%% `send' tuple under the `priv' field.

-export_type([trigger_action/0,
              triggered_action/0,
              trigger_exec_loc/0,
              trigger_descriptor/0]).

-define(SERVER, ?MODULE).

-define(SILENCE_ERROR_FOR, 10000).

-record(?MODULE, {trigger_crashes = #{}}).

start_link() ->
    gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).

handle_triggered_actions(_StoreId, []) ->
    ok;
handle_triggered_actions(StoreId, TriggeredActions) ->
    gen_server:cast(?SERVER, {?FUNCTION_NAME, StoreId, TriggeredActions}).

init(_) ->
    erlang:process_flag(trap_exit, true),
    State = #?MODULE{},
    {ok, State}.

handle_call(Request, From, State) ->
    ?LOG_WARNING(
       "Unhandled handle_call request from ~0p: ~p",
       [From, Request]),
    {State1, Timeout} = log_accumulated_trigger_crashes(State),
    {reply, ok, State1, Timeout}.

handle_cast(
  {handle_triggered_actions, StoreId, TriggeredActions}, State) ->
    State1 =
    lists:foldl(
      fun(TriggeredAction, S) ->
              ActionArg = prepare_action_arg(StoreId, TriggeredAction),
              run_triggered_action(StoreId, TriggeredAction, ActionArg, S)
      end, State, TriggeredActions),
    _ = khepri_machine:ack_triggers_execution(StoreId, TriggeredActions),
    {State2, Timeout} = log_accumulated_trigger_crashes(State1),
    {noreply, State2, Timeout};
handle_cast(Request, State) ->
    ?LOG_WARNING("Unhandled handle_cast request: ~p", [Request]),
    {State1, Timeout} = log_accumulated_trigger_crashes(State),
    {noreply, State1, Timeout}.

handle_info(timeout, State) ->
    {State1, Timeout} = log_accumulated_trigger_crashes(State),
    {noreply, State1, Timeout};
handle_info(Msg, State) ->
    ?LOG_WARNING("Unhandled handle_info message: ~p", [Msg]),
    {State1, Timeout} = log_accumulated_trigger_crashes(State),
    {noreply, State1, Timeout}.

terminate(_Reason, _State) ->
    ok.

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

prepare_action_arg(_StoreId, #triggered{props = Props}) ->
    Arg = Props,
    Arg;
prepare_action_arg(
  StoreId,
  #triggered_v2{id = TriggerId,
                event = Event,
                action = Action}) ->
    EventType = event_type(Event),
    EventProps = event_to_props(Event),
    ActionProps = action_to_props(Action),
    Arg = #khepri_trigger{type = EventType,
                          store_id = StoreId,
                          trigger_id = TriggerId,
                          event = EventProps,
                          action = ActionProps},
    Arg.

event_type(#ev_tree{}) ->
    tree.

event_to_props(#ev_tree{path = Path, change = Change}) ->
    #{path => Path,
     change => Change}.

action_to_props({sproc, _StoredProc}) ->
    #{}.

run_triggered_action(
  StoreId,
  #triggered{sproc = StoredProc} = TriggeredAction,
  ActionArg, State) ->
    run_triggered_sproc(
      StoreId, TriggeredAction, StoredProc, ActionArg, State);
run_triggered_action(
  StoreId,
  #triggered_v2{action = {sproc, StoredProc}} = TriggeredAction,
  ActionArg, State) ->
    run_triggered_sproc(
      StoreId, TriggeredAction, StoredProc, ActionArg, State).

run_triggered_sproc(StoreId, TriggeredAction, StoredProc, ActionArg, State) ->
    Args = [ActionArg],
    try
        %% TODO: Be flexible and accept a function with an arity of 0.
        _ = khepri_sproc:run(StoredProc, Args),
        State
    catch
        Class:Reason:Stacktrace ->
            handle_action_crash(
              StoreId, TriggeredAction, ActionArg,
              Class, Reason, Stacktrace, State)
    end.

handle_action_crash(
  StoreId, TriggeredAction, ActionArg, Class, Reason, Stacktrace,
  #?MODULE{trigger_crashes = Crashes} = State) ->
    Key = {Class, Reason, Stacktrace},
    case Crashes of
        #{Key := {Timestamp, Count, Msg}} ->
            Crashes1 = Crashes#{Key => {Timestamp, Count + 1, Msg}},
            State#?MODULE{trigger_crashes = Crashes1};
        _ ->
            TriggerId = case TriggeredAction of
                            #triggered{id = Id}    -> Id;
                            #triggered_v2{id = Id} -> Id
                        end,
            EventFilter = case TriggeredAction of
                              #triggered{event_filter = EF}    -> EF;
                              #triggered_v2{event_filter = EF} -> EF
                          end,
            Msg = io_lib:format(
                    "Triggered stored procedure crash~n"
                    "  Store ID: ~s~n"
                    "  Trigger ID: ~s~n"
                    "  Event filter:~n"
                    "    ~p~n"
                    "  Action arg:~n"
                    "    ~p~n"
                    "  Crash:~n"
                    "    ~ts",
                    [StoreId, TriggerId, EventFilter, ActionArg,
                     khepri_utils:format_exception(
                       Class, Reason, Stacktrace,
                       #{column => 4})]),
            ?LOG_ERROR(Msg, []),

            Timestamp = erlang:monotonic_time(millisecond),
            Crashes1 = Crashes#{Key => {Timestamp, 1, Msg}},
            State#?MODULE{trigger_crashes = Crashes1}
    end.

log_accumulated_trigger_crashes(
  #?MODULE{trigger_crashes = Crashes} = State)
  when Crashes =:= #{} ->
    {State, infinity};
log_accumulated_trigger_crashes(
  #?MODULE{trigger_crashes = Crashes} = State) ->
    Now = erlang:monotonic_time(millisecond),
    Crashes1 = maps:filter(
                 fun
                     (_Key, {Timestamp, Count, Msg})
                       when Now - Timestamp >= ?SILENCE_ERROR_FOR andalso
                            Msg > 1 ->
                         ?LOG_ERROR(
                            "~ts~n"
                            "(this crash occurred ~b times in the last ~b "
                            "seconds)",
                            [Msg, Count, (Now - Timestamp) div 1000]),
                         false;
                     (_Key, _Value) ->
                         true
                 end, Crashes),
    State1 = State#?MODULE{trigger_crashes = Crashes1},
    case Crashes =:= #{} of
        true  -> {State1, infinity};
        false -> {State1, ?SILENCE_ERROR_FOR}
    end.