temporal_sdk

-module(temporal_sdk_api_awaitable).

% elp:ignore W0012 W0040
-moduledoc false.

-export([
    cast_key/2,
    init_match/2,
    match_test/2,
    match_test/3,
    is_ready/1,
    update_match/6,
    gen_idx/7,
    unblock_awaitable/4
]).

-spec cast_key(
    AwaitPattern :: temporal_sdk_workflow:await_pattern() | [temporal_sdk_workflow:await_pattern()],
    ApiCtx :: temporal_sdk_api:context()
) -> tuple().
cast_key({Operator, [P | TPattern]}, ApiContext) when Operator =:= all; Operator =:= any ->
    {H1, H2} = cast_key(P, ApiContext),
    {T1, T2} = cast_key(TPattern, ApiContext),
    {{Operator, [H1 | T1]}, {Operator, [H2 | T2]}};
cast_key([P | TPattern], ApiContext) ->
    {H1, H2} = cast_key(P, ApiContext),
    {T1, T2} = cast_key(TPattern, ApiContext),
    {[H1 | T1], [H2 | T2]};
cast_key([], _ApiContext) ->
    {[], []};
cast_key(#{state := _} = AwaitableData, _ApiContext) ->
    {awaitable_data, AwaitableData};
cast_key(Pattern, ApiContext) ->
    temporal_sdk_api_awaitable_index:cast_key(Pattern, ApiContext).

-spec init_match(AwaitPattern :: temporal_sdk_workflow:await_pattern(), {
    HistoryTable :: ets:table(), IndexTable :: ets:table()
}) -> temporal_sdk_workflow:await_match().
init_match({Operator, Pattern}, Tables) when Operator =:= all; Operator =:= any ->
    {Operator, [init_match(P, Tables) || P <- Pattern]};
init_match({event, HistoryEventPattern}, {HistoryTable, _IndexTable}) ->
    temporal_sdk_api_awaitable_history_table:fetch(HistoryTable, HistoryEventPattern);
init_match(#{state := _} = AwaitableData, {_HistoryTable, _IndexTable}) ->
    AwaitableData;
init_match(IndexKey, {_HistoryTable, IndexTable}) ->
    % eqwalizer:ignore
    temporal_sdk_api_awaitable_index_table:fetch(IndexTable, IndexKey).

match_test({Operator, Pattern}, {Operator, Match}) when Operator =:= all; Operator =:= any ->
    {Operator, lists:zipwith(fun match_test/2, Pattern, Match)};
match_test(awaitable_data, _AwaitableData) ->
    {true, true};
match_test(Pattern, Match) ->
    A = element(1, Pattern),
    {
        temporal_sdk_api_awaitable_index:is_ready(A, Match),
        temporal_sdk_api_awaitable_index:is_closed(A, Match)
    }.

match_test({Operator, Pattern}, {Operator, Match}, EId) when Operator =:= all; Operator =:= any ->
    {Operator, lists:zipwith(fun(P, M) -> match_test(P, M, EId) end, Pattern, Match)};
match_test(awaitable_data, _AwaitableData, _EId) ->
    {true, true};
match_test(Pattern, Match, EId) ->
    A = element(1, Pattern),
    {
        temporal_sdk_api_awaitable_index:is_ready(A, Match, EId),
        temporal_sdk_api_awaitable_index:is_closed(A, Match)
    }.

is_ready({any, [M | TMatches] = Matches}) ->
    case
        lists:member({true, false}, Matches) orelse lists:member({true, true}, Matches) orelse
            lists:member(true, Matches)
    of
        true ->
            true;
        false ->
            case M of
                {O, _} when O =:= any; O =:= all -> is_ready({any, [is_ready(M) | TMatches]});
                {_, _} -> is_ready({any, TMatches});
                false -> is_ready({any, TMatches})
            end
    end;
is_ready({any, []}) ->
    false;
is_ready({all, [M | TMatches] = Matches}) ->
    case lists:member({false, false}, Matches) orelse lists:member(false, Matches) of
        true ->
            false;
        false ->
            case M of
                {O, _} when O =:= any; O =:= all -> is_ready({all, [is_ready(M) | TMatches]});
                {_, _} -> is_ready({all, TMatches});
                true -> is_ready({all, TMatches})
            end
    end;
is_ready({all, []}) ->
    true;
is_ready({true, _}) ->
    true;
is_ready({false, _}) ->
    false.

update_match(Pattern, PatternKey, Match, Test, IndexTable, HistoryTable) ->
    case do_update(PatternKey, Match, Test, IndexTable, HistoryTable) of
        Match ->
            noop;
        NewMatch ->
            NewTest = match_test(Pattern, NewMatch),
            case is_ready(NewTest) of
                true -> {ready, NewMatch};
                false -> {update, NewMatch, NewTest}
            end
    end.

do_update({O, Pattern}, {O, Match}, {O, Test}, IT, HT) when O =:= all; O =:= any ->
    {O, lists:zipwith3(fun(P, M, T) -> do_update(P, M, T, IT, HT) end, Pattern, Match, Test)};
do_update(_Pattern, Match, {true, true}, _IndexTable, _HistoryTable) ->
    Match;
do_update({event, Pattern}, noevent, {false, false}, _IndexTable, HistoryTable) ->
    temporal_sdk_api_awaitable_history_table:fetch(HistoryTable, Pattern);
do_update(IndexKey, _Match, {_, false}, IndexTable, _HistoryTable) ->
    temporal_sdk_api_awaitable_index_table:fetch(IndexTable, IndexKey);
do_update(_Pattern, Match, {_, _}, _IndexTable, _HistoryTable) ->
    Match.

-spec gen_idx(
    OptsAttr :: map(),
    IndexKeyList :: list(),
    ExecutionId :: temporal_sdk_workflow:execution_id(),
    ACounter :: pos_integer(),
    Cmds :: [temporal_sdk_workflow:index_command()],
    ApiCtx :: temporal_sdk_api:context(),
    MsgName :: temporal_sdk_client:msg_name() | atom()
) ->
    {
        temporal_sdk_workflow:awaitable(),
        temporal_sdk_workflow:awaitable_index_key()
    }.
gen_idx(#{activity_id := Id} = OA, IndexKeyList, _EId, _ACounter, _Cmds, ApiCtx, _MName) when
    Id =/= "", Id =/= ~"", Id =/= ''
->
    do_gen_idx(Id, IndexKeyList, ApiCtx, OA);
gen_idx(#{marker_name := Id} = OA, IndexKeyList, _EId, _ACounter, _Cmds, ApiCtx, _MName) when
    Id =/= "", Id =/= ~"", Id =/= ''
->
    do_gen_idx(Id, IndexKeyList, ApiCtx, OA);
gen_idx(#{timer_id := Id} = OA, IndexKeyList, _EId, _ACounter, _Cmds, ApiCtx, _MName) when
    Id =/= "", Id =/= ~"", Id =/= ''
->
    do_gen_idx(Id, IndexKeyList, ApiCtx, OA);
gen_idx(#{workflow_id := Id} = OA, IndexKeyList, _EId, _ACounter, _Cmds, ApiCtx, _MName) when
    Id =/= "", Id =/= ~"", Id =/= ''
->
    do_gen_idx(Id, IndexKeyList, ApiCtx, OA);
% SDK
gen_idx(#{execution_id := Id} = OA, IndexKeyList, _EId, _ACounter, _Cmds, ApiCtx, _MName) when
    Id =/= "", Id =/= ~"", Id =/= ''
->
    do_gen_idx(Id, IndexKeyList, ApiCtx, OA);
gen_idx(#{info_id := Id} = OA, IndexKeyList, _EId, _ACounter, _Cmds, ApiCtx, _MName) when
    Id =/= "", Id =/= ~"", Id =/= ''
->
    do_gen_idx(Id, IndexKeyList, ApiCtx, OA);
gen_idx(#{awaitable_id := Opts} = OA, IndexKeyList, ExecutionId, ACounter, Cmds, ApiCtx, MName) when
    is_map(Opts)
->
    N1 =
        case Opts of
            #{id := Id} when Id =/= "", Id =/= ~"", Id =/= '' ->
                [temporal_sdk_api:serialize(ApiCtx, MName, id, Id)];
            #{} ->
                []
        end,
    N2 =
        case Opts of
            #{prefix := true} ->
                SerializedExecutionId =
                    temporal_sdk_api:serialize(ApiCtx, MName, id, ExecutionId),
                [SerializedExecutionId | [ACounter | N1]];
            #{prefix := false} ->
                N1;
            #{prefix := Pr} when Pr =/= "", Pr =/= ~"", Pr =/= '' -> [Pr | N1];
            #{} ->
                N1
        end,
    case N2 of
        [] ->
            erlang:error("Void awaitable id/name.", [
                OA,
                IndexKeyList,
                ExecutionId,
                ACounter,
                Cmds,
                ApiCtx,
                MName
            ]);
        _ ->
            ok
    end,
    case Opts of
        #{postfix := true} ->
            N = temporal_sdk_utils_path:string_path(N2) ++ "/",
            NCasted = temporal_sdk_api_awaitable_index:cast_value(hd(IndexKeyList), N, ApiCtx),
            IndexKeyListCasted = temporal_sdk_api_awaitable_index:cast_list(IndexKeyList, ApiCtx),
            NId = new_id(IndexKeyListCasted, NCasted, Cmds, ApiCtx),
            Idx = list_to_tuple(do_set_idx_event(OA, IndexKeyList) ++ [NId]),
            IdxCasted = list_to_tuple(IndexKeyListCasted ++ [NId]),
            {Idx, IdxCasted};
        #{postfix := Po} when Po =/= false, Po =/= "", Po =/= ~"", Po =/= '' ->
            NId = temporal_sdk_utils_path:string_path(
                N2 ++ [temporal_sdk_api:serialize(ApiCtx, MName, id, Po)]
            ),
            do_gen_idx(NId, IndexKeyList, ApiCtx, OA);
        #{} ->
            NId = temporal_sdk_utils_path:string_path(N2),
            do_gen_idx(NId, IndexKeyList, ApiCtx, OA)
    end;
gen_idx(#{awaitable_id := Id} = Opts, IndexKeyList, ExeId, ACounter, Cmds, ApiCtx, MName) ->
    gen_idx(
        Opts#{awaitable_id := #{prefix => true, postfix => true, id => Id}},
        IndexKeyList,
        ExeId,
        ACounter,
        Cmds,
        ApiCtx,
        MName
    ).

do_gen_idx(Id, IndexKeyList, ApiCtx, OptsAttr) ->
    IdCasted = temporal_sdk_api_awaitable_index:cast_value(hd(IndexKeyList), Id, ApiCtx),
    IndexKeyListCasted = temporal_sdk_api_awaitable_index:cast_list(IndexKeyList, ApiCtx),
    Idx = list_to_tuple(do_set_idx_event(OptsAttr, IndexKeyList) ++ [Id]),
    IdxCasted = list_to_tuple(IndexKeyListCasted ++ [IdCasted]),
    {Idx, IdxCasted}.

do_set_idx_event(#{awaitable_event := AE}, [A | TIdxKeyList]) ->
    AwaitableEvent = temporal_sdk_api_awaitable_index:to_event(A, AE),
    [AwaitableEvent | TIdxKeyList];
do_set_idx_event(#{}, IdxKeyList) ->
    IdxKeyList.

new_id(IKPrefix, NamePattern, Cmds, ApiCtx) ->
    Count = do_count_match(IKPrefix, NamePattern, Cmds),
    do_new_id(IKPrefix, NamePattern, Cmds, ApiCtx, Count).

do_new_id(IKPrefix, NamePattern, Cmds, ApiCtx, Count) ->
    NewName =
        case is_binary(NamePattern) of
            true ->
                C = integer_to_binary(Count),
                <<NamePattern/binary, C/binary>>;
            false ->
                NamePattern ++ integer_to_list(Count)
        end,
    case do_count_match(IKPrefix, NewName, Cmds) of
        0 -> NewName;
        _ -> do_new_id(IKPrefix, NamePattern, Cmds, ApiCtx, Count + 1)
    end.

do_count_match(IKPrefix, NamePattern, Cmds) ->
    lists:foldl(
        fun({{IK, _IV}, _C}, Acc) when is_integer(Acc) ->
            do_match_idx(IKPrefix, NamePattern, IK) + Acc
        end,
        0,
        Cmds
    ).

do_match_idx([A, T], P, {A, T, N}) -> do_match_name(P, N);
do_match_idx([A], P, {A, N}) -> do_match_name(P, N);
do_match_idx(_, _, _) -> 0.

do_match_name(P, N) when is_binary(P), is_binary(N) ->
    PSize = byte_size(P),
    case N of
        <<P:PSize/binary, _/binary>> -> 1;
        _ -> 0
    end;
do_match_name(P, N) when is_list(P), is_list(N) ->
    case lists:prefix(P, N) of
        true -> 1;
        false -> 0
    end.

-spec unblock_awaitable(
    Index :: {
        temporal_sdk_workflow:activity_index_key(), temporal_sdk_workflow:activity_data()
    },
    HistoryEvents :: [temporal_sdk_workflow:history_event()],
    ApiCtx :: temporal_sdk_api:context(),
    IndexTable :: ets:table()
) -> ok | noevent | {error, Reason :: map()}.
unblock_awaitable(
    {{activity, _} = IdxKey, #{direct_result := true, event_id := EId}},
    HistoryEvents,
    ApiCtx,
    IndexTable
) ->
    Fn = fun
        ({_, 'EVENT_TYPE_ACTIVITY_TASK_COMPLETED', #{scheduled_event_id := SEId}, _}) when
            SEId =:= EId
        ->
            true;
        ({_, 'EVENT_TYPE_ACTIVITY_TASK_CANCELED', #{scheduled_event_id := SEId}, _}) when
            SEId =:= EId
        ->
            true;
        (_) ->
            false
    end,
    case lists:search(Fn, HistoryEvents) of
        {value, {_, 'EVENT_TYPE_ACTIVITY_TASK_COMPLETED', #{result := Result}, _}} ->
            R = temporal_sdk_api:map_from_payloads(
                ApiCtx,
                'temporal.api.history.v1.ActivityTaskCompletedEventAttributes',
                result,
                Result
            ),
            Index = {IdxKey, #{result => R}},
            % eqwalizer:ignore
            temporal_sdk_api_awaitable_index_table:upsert_cmd(IndexTable, Index);
        {value, {_, 'EVENT_TYPE_ACTIVITY_TASK_COMPLETED', #{}, _}} ->
            R = temporal_sdk_api:map_from_payloads(
                ApiCtx,
                'temporal.api.history.v1.ActivityTaskCompletedEventAttributes',
                result,
                #{}
            ),
            Index = {IdxKey, #{result => R}},
            % eqwalizer:ignore
            temporal_sdk_api_awaitable_index_table:upsert_cmd(IndexTable, Index);
        {value, {_, 'EVENT_TYPE_ACTIVITY_TASK_CANCELED', #{}, _}} ->
            Index = {IdxKey, #{result => 'ACTIVITY_TASK_CANCELED'}},
            % eqwalizer:ignore
            temporal_sdk_api_awaitable_index_table:upsert_cmd(IndexTable, Index);
        _ ->
            noevent
    end;
unblock_awaitable(UsupportedAwaitable, _HistoryEvents, _ApiCtx, _IndexTable) ->
    {error, #{
        reason => "Expected blocked awaitable, got unsupported awaitable.",
        unsupported_awaitable => UsupportedAwaitable
    }}.