temporal_sdk

-module(temporal_sdk_api_context).

% elp:ignore W0012 W0040
-moduledoc false.

-export([
    build/1,
    activity_from_workflow/1,
    update/2,
    take_next_page_token/1,
    add_activity_opts/3,
    add_activity_opts/4,
    add_nexus_task_opts/2,
    add_workflow_opts/3,
    restart_workflow_task_opts/2,
    add_limiter_counters/3
]).

-spec build(Cluster :: temporal_sdk_cluster:cluster_name()) ->
    {ok, temporal_sdk_api:context()} | {error, invalid_cluster}.
build(Cluster) ->
    case temporal_sdk_cluster_sup:get_context(Cluster) of
        {ok, ApiContext} -> {ok, ApiContext#{worker_identity => self()}};
        Err -> Err
    end.

-spec activity_from_workflow(WFApiContext :: temporal_sdk_api:context()) ->
    ActivityApiContext :: temporal_sdk_api:context().
activity_from_workflow(WFApiContext) ->
    #{worker_opts := WFWorkerOpts} = WFApiContext,
    #{task_settings := #{eager_execution_settings := EagerExecutionSettings}} = WFWorkerOpts,
    WO = maps:with([task_queue, failure_info, worker_version], WFWorkerOpts),
    WorkerOpts = WO#{
        worker_id => eager_execution_worker,
        namespace => <<>>,
        task_settings => EagerExecutionSettings,
        allowed_temporal_names => all,
        allowed_erlang_modules => all,
        temporal_name_to_erlang => fun temporal_sdk_api:temporal_name_to_erlang/2
    },
    C = maps:with([cluster, client_opts, worker_identity], WFApiContext),
    C#{worker_opts => WorkerOpts}.

-spec update(
    ApiContext :: temporal_sdk_api:context(),
    Task ::
        temporal_sdk_workflow:task()
        | temporal_sdk_activity:task()
        | #{task_token := undefined | unicode:chardata()}
) -> UpdatedApiContext :: temporal_sdk_api:context().
update(#{task_opts := TaskOpts} = ApiContext, #{task_token := TaskToken}) ->
    ApiContext#{task_opts := TaskOpts#{token := TaskToken}};
update(#{task_opts := _} = ApiContext, #{}) ->
    ApiContext.

-spec take_next_page_token(ApiContext :: temporal_sdk_api:context()) ->
    {NextPageToken :: unicode:chardata(), temporal_sdk_api:context()}.
take_next_page_token(#{task_opts := #{next_page_token := NextPageToken} = TaskOpts} = ApiContext) ->
    {NextPageToken, ApiContext#{task_opts := maps:without([next_page_token], TaskOpts)}}.

-spec add_activity_opts(
    ApiContext :: temporal_sdk_api:context(),
    Task :: temporal_sdk_activity:task() | #{task_token := undefined},
    ExecutionModule :: module()
) ->
    ApiContextWithOpts :: temporal_sdk_api:context().
add_activity_opts(ApiContext, #{task_token := TaskToken}, ExecutionModule) ->
    ApiContext#{task_opts => #{token => TaskToken}, execution_module => ExecutionModule}.

-spec add_activity_opts(
    ApiContext :: temporal_sdk_api:context(),
    Task :: temporal_sdk_activity:task() | #{task_token := undefined},
    IndexKey :: temporal_sdk_workflow:activity(),
    ExecutionModule :: module()
) ->
    ApiContextWithOpts :: temporal_sdk_api:context().
add_activity_opts(ApiContext, #{task_token := TaskToken}, IndexKey, ExecutionModule) ->
    ApiContext#{
        task_opts => #{token => TaskToken, index_key => IndexKey},
        execution_module => ExecutionModule
    }.

-spec add_nexus_task_opts(
    ApiContext :: temporal_sdk_api:context(),
    Task :: temporal_sdk_nexus:task()
) ->
    ApiContextWithTaskOpts :: temporal_sdk_api:context().
add_nexus_task_opts(ApiContext, #{task_token := TaskToken}) ->
    ApiContext#{task_opts => #{token => TaskToken}}.

-spec add_workflow_opts(
    ApiContext :: temporal_sdk_api:context(),
    Task :: temporal_sdk_workflow:task(),
    ExecutionModule :: module()
) ->
    ApiContextWithTaskOpts :: temporal_sdk_api:context().
add_workflow_opts(
    #{worker_opts := #{task_queue := TaskQueue}} = ApiContext,
    #{
        task_token := TaskToken,
        next_page_token := NextPageToken,
        workflow_type := #{name := WorkflowTypeName},
        workflow_execution := #{workflow_id := WorkflowId, run_id := RunId}
    } =
        Task,
    ExecutionModule
) when not is_function(TaskQueue) ->
    StickyName = temporal_sdk_utils_path:string_path([
        node(), pid_to_list(self()), temporal_sdk_utils:uuid4()
    ]),
    ApiContext#{
        execution_module => ExecutionModule,
        task_opts => #{
            token => TaskToken,
            next_page_token => NextPageToken,
            workflow_type => WorkflowTypeName,
            workflow_id => WorkflowId,
            run_id => RunId,
            sticky_attributes => #{
                worker_task_queue => #{
                    name => StickyName,
                    kind => 'TASK_QUEUE_KIND_STICKY',
                    normal_name => TaskQueue
                },
                schedule_to_start_timeout => fetch_sticky_execution_schedule_to_start_timeout(
                    ApiContext, Task
                )
            }
        }
    }.

fetch_sticky_execution_schedule_to_start_timeout(
    #{worker_opts := #{task_settings := #{sticky_execution_schedule_to_start_ratio := Ratio}}}, Task
) ->
    case temporal_sdk_api_workflow_task:workflow_task_timeout_msec(Task) of
        T when is_integer(T) -> temporal_sdk_utils_time:msec_to_protobuf(round(T * Ratio));
        _ -> temporal_sdk_utils_time:msec_to_protobuf(5_000)
    end.

-spec restart_workflow_task_opts(
    ApiContext :: temporal_sdk_api:context(),
    Task :: temporal_sdk_workflow:task()
) ->
    ApiContextWithTaskOpts :: temporal_sdk_api:context().
restart_workflow_task_opts(
    #{task_opts := TaskOpts} = ApiContext,
    #{
        task_token := TaskToken,
        next_page_token := NextPageToken,
        workflow_execution := #{run_id := RunId}
    }
) ->
    % eqwalizer:ignore
    ApiContext#{
        task_opts := TaskOpts#{
            token := TaskToken, next_page_token => NextPageToken, run_id := RunId
        }
    }.

-spec add_limiter_counters(
    ApiContext :: temporal_sdk_api:context(),
    WorkerType :: temporal_sdk_worker:worker_type(),
    LimiterCounters :: temporal_sdk_limiter:counters()
) ->
    ApiContextWithLimiters ::
        temporal_sdk_api:context()
        | {invalid_opts, Reason :: map()}.
add_limiter_counters(ApiContext, activity, LimiterCounters) ->
    do_add_limiter_counters(ApiContext, [activity_regular], LimiterCounters, []);
add_limiter_counters(ApiContext, session, LimiterCounters) ->
    do_add_limiter_counters(ApiContext, [activity_session], LimiterCounters, []);
add_limiter_counters(ApiContext, nexus, LimiterCounters) ->
    do_add_limiter_counters(ApiContext, [nexus], LimiterCounters, []);
add_limiter_counters(ApiContext, workflow, LimiterCounters) ->
    do_add_limiter_counters(
        ApiContext, [workflow, activity_eager, activity_direct], LimiterCounters, []
    ).

do_add_limiter_counters(ApiContext, [L | TLimitables], LimiterCounters, Acc) ->
    case temporal_sdk_limiter:build_counters(L, LimiterCounters) of
        {ok, C} -> do_add_limiter_counters(ApiContext, TLimitables, LimiterCounters, [C | Acc]);
        Err -> Err
    end;
do_add_limiter_counters(ApiContext, [], _LimiterCounters, Acc) ->
    ApiContext#{limiter_counters => lists:reverse(Acc)}.