-module(temporal_sdk_api_common). % elp:ignore W0012 W0040 -moduledoc false. -export([ retry_policy/1, get_by_casted_key/4, mutation_reset_reason/1, marker_encode_value/2, marker_decode_value/2, run_request/5, format_response/4, t2e/3 ]). -include("proto.hrl"). -spec retry_policy(RetryPolicy :: temporal_sdk:retry_policy()) -> ?TEMPORAL_SPEC:'temporal.api.common.v1.RetryPolicy'(). retry_policy(RetryPolicy) -> RP1 = case RetryPolicy of #{initial_interval := II} -> RetryPolicy#{initial_interval := temporal_sdk_utils_time:msec_to_protobuf(II)}; #{} -> RetryPolicy end, case RP1 of #{maximum_interval := MI} -> RP1#{maximum_interval := temporal_sdk_utils_time:msec_to_protobuf(MI)}; #{} -> RP1 end. -spec get_by_casted_key( Key :: term(), Map :: map(), Default :: term(), ApiCtx :: temporal_sdk_api:context() ) -> {CastedKey :: unicode:chardata(), Value :: term()}. get_by_casted_key(Key, Map, Default, #{client_opts := #{grpc_opts := #{codec := {Codec, _, _}}}}) -> CastedKey = Codec:cast(Key), case Map of #{CastedKey := V} -> {CastedKey, V}; #{} -> {CastedKey, Default} end. -spec mutation_reset_reason(Name :: unicode:chardata() | atom()) -> {string(), binary()}. mutation_reset_reason(Name) -> Str = temporal_sdk_utils_path:string_path([?MUTATION_RESET_PREFIX, Name], ":"), {Str, temporal_sdk_utils_unicode:characters_to_binary1(Str)}. -spec marker_encode_value( Codec :: temporal_sdk_workflow:marker_value_codec(), Value :: temporal_sdk:term_to_payloads() | term() ) -> {ok, EncodedValue :: temporal_sdk:term_to_payloads(), Decoder :: none | list | term | fun((temporal_sdk:term_from_payloads()) -> term()) | {Module :: module(), Function :: atom()}} | {error, Reason :: map()}. marker_encode_value(list, Value) -> {ok, [Value], list}; marker_encode_value(term, Value) -> {ok, [erlang:term_to_binary(Value)], term}; marker_encode_value(none, Value) when is_list(Value) -> {ok, Value, none}; marker_encode_value({none, none}, Value) when is_list(Value) -> {ok, Value, none}; marker_encode_value({none, DF}, Value) when is_list(Value), is_function(DF, 1) -> {ok, Value, DF}; marker_encode_value({none, {DM, DF}}, Value) when is_list(Value), is_atom(DM), is_atom(DF) -> {ok, Value, {DM, DF}}; marker_encode_value({EF, none}, Value) when is_function(EF, 1) -> case EF(Value) of EV when is_list(EV) -> {ok, EV, none}; EV -> do_encode_type_err(EF, Value, EV) end; marker_encode_value({{EM, EF}, none}, Value) when is_atom(EM), is_atom(EF) -> case EM:EF(Value) of EV when is_list(EV) -> {ok, EV, none}; EV -> do_encode_type_err({EM, EF}, Value, EV) end; marker_encode_value({EF, none}, Value) when is_function(EF, 1) -> case EF(Value) of EV when is_list(EV) -> {ok, EV, none}; EV -> do_encode_type_err(EF, Value, EV) end; marker_encode_value({EF, DF}, Value) when is_function(EF, 1), is_function(DF, 1) -> case EF(Value) of EV when is_list(EV) -> {ok, EV, none}; EV -> do_encode_type_err(EF, Value, EV) end; marker_encode_value({EF, {DM, DF}}, Value) when is_function(EF, 1), is_atom(DM), is_atom(DF) -> case EF(Value) of EV when is_list(EV) -> {ok, EV, {DM, DF}}; EV -> do_encode_type_err(EF, Value, EV) end; marker_encode_value(Codec, Value) -> {error, #{ reason => "Invalid marker value type or invalid marker configuration.", marker_value => Value, marker_value_codec => Codec }}. do_encode_type_err(Encoder, Value, EncodedValue) -> {error, #{ reason => "Invalid encoded marker value type. Expected a list() type.", marker_value => Value, invalid_encoded_marker_value => EncodedValue, encoder => Encoder }}. -spec marker_decode_value( Decoder :: none | list | term | fun((temporal_sdk:term_from_payloads()) -> term()) | {Module :: module(), Function :: atom()}, Value :: temporal_sdk:term_from_payloads() ) -> DecodedValue :: term(). marker_decode_value(list, [Value]) -> Value; marker_decode_value(term, [Value]) -> erlang:binary_to_term(Value); marker_decode_value(none, Value) -> Value; marker_decode_value(DF, Value) when is_function(DF, 1) -> DF(Value); marker_decode_value({DM, DF}, Value) when is_atom(DM), is_atom(DF) -> DM:DF(Value). -spec run_request( Cluster :: temporal_sdk_cluster:cluster_name(), Opts :: proplists:proplist(), DefaultOpts :: temporal_sdk_utils_opts:defaults(), ServiceName :: temporal_sdk_api:temporal_service(), MessageName :: temporal_msg_name() ) -> temporal_sdk:response(). run_request(Cluster, Opts, DefaultOpts, ServiceName, MessageName) -> maybe {ok, ApiCtx} ?= temporal_sdk_api_context:build(Cluster), {ok, FullOpts} ?= temporal_sdk_utils_opts:build(DefaultOpts, Opts, ApiCtx), {RawRequest, O1} = maps:take(raw_request, FullOpts), {ResponseType, O2} = maps:take(response_type, O1), Req = maps:merge(O2, RawRequest), Response = temporal_sdk_api:request(ServiceName, Cluster, Req, ResponseType, #{}), format_response(MessageName, ResponseType, Response, ApiCtx) end. -spec format_response( MessageName :: temporal_msg_name(), ResponseType :: temporal_sdk:response_type(), Response :: {ok, temporal_msg()} | {error, term()}, ApiCtx :: temporal_sdk_api:context() ) -> temporal_sdk:response(). format_response(MessageName, call_formatted, {ok, Response}, ApiCtx) -> {ok, t2e(MessageName, Response, ApiCtx)}; format_response( _MessageName, call_formatted, {error, #{grpc_response_headers := H}}, _ApiCtx ) when is_list(H) -> Err = case proplists:get_value(~"grpc-message", H, '$_undefined') of '$_undefined' -> proplists:get_value("grpc-message", H, undefined); E -> E end, {error, Err}; format_response(_MessageName, _Opts, Response, _ApiCtx) -> Response. -spec t2e( MessageName :: temporal_msg_name(), Message :: temporal_msg(), ApiCtx :: temporal_sdk_api:context() ) -> ConvertedMessage :: map(). t2e(MessageName, Message, ApiCtx) when is_map(Message) -> maps:from_list(te(MessageName, maps:to_list(Message), ApiCtx, [], [])). te(MN, [{K, #{payloads := _} = V} | TM], ApiCtx, KAcc, Acc) -> te(MN, TM, ApiCtx, KAcc, [ {K, temporal_sdk_api:map_from_payloads(ApiCtx, MN, KAcc ++ [K], V)} | Acc ]); te(MN, [{K, #{data := _, metadata := _} = V} | TM], ApiCtx, KAcc, Acc) -> te(MN, TM, ApiCtx, KAcc, [ {K, temporal_sdk_api:map_from_payload(ApiCtx, MN, KAcc ++ [K], V)} | Acc ]); te(MN, [{header = K, #{fields := V}} | TM], ApiCtx, KAcc, Acc) -> te(MN, TM, ApiCtx, KAcc, [ {K, temporal_sdk_api:map_from_mapstring_payload(ApiCtx, MN, KAcc ++ [K, fields], V)} | Acc ]); te(MN, [{memo = K, #{fields := V}} | TM], ApiCtx, KAcc, Acc) -> te(MN, TM, ApiCtx, KAcc, [ {K, temporal_sdk_api:map_from_mapstring_payload(ApiCtx, MN, KAcc ++ [K, fields], V)} | Acc ]); te(MN, [{search_attributes = K, #{indexed_fields := V}} | TM], ApiCtx, KAcc, Acc) -> te(MN, TM, ApiCtx, KAcc, [ {K, temporal_sdk_api:map_from_mapstring_payload(ApiCtx, MN, KAcc ++ [K, fields], V)} | Acc ]); te(MN, [{workflow_type = K, #{name := N}} | TM], ApiCtx, KAcc, Acc) -> te(MN, TM, ApiCtx, KAcc, [{K, N} | Acc]); te(MN, [{task_queue = K, #{name := N}} | TM], ApiCtx, KAcc, Acc) -> te(MN, TM, ApiCtx, KAcc, [{K, N} | Acc]); te(MN, [{type = K, #{name := N}} | TM], ApiCtx, KAcc, Acc) -> te(MN, TM, ApiCtx, KAcc, [{K, N} | Acc]); te(MN, [{K, #{seconds := _, nanos := _} = T} | TM], ApiCtx, KAcc, Acc) -> te(MN, TM, ApiCtx, KAcc, [{K, temporal_sdk_utils_time:protobuf_to_msec(T)} | Acc]); te(MN, [{K, [#{} | _] = Li} | TM], ApiCtx, KAcc, Acc) -> V = lists:map( fun(M) when is_map(M) -> maps:from_list(te(MN, maps:to_list(M), ApiCtx, KAcc ++ [K], [])) end, Li ), te(MN, TM, ApiCtx, KAcc, [{K, V} | Acc]); te(MN, [{K, #{} = Ma} | TM], ApiCtx, KAcc, Acc) -> V = maps:from_list(te(MN, maps:to_list(Ma), ApiCtx, KAcc ++ [K], [])), te(MN, TM, ApiCtx, KAcc, [{K, V} | Acc]); te(MN, [{K, {KT, #{} = Ma}} | TM], ApiCtx, KAcc, Acc) -> V = maps:from_list(te(MN, maps:to_list(Ma), ApiCtx, KAcc ++ [K], [])), te(MN, TM, ApiCtx, KAcc, [{K, {KT, V}} | Acc]); te(MN, [KV | TM], ApiCtx, KAcc, Acc) -> te(MN, TM, ApiCtx, KAcc, [KV | Acc]); te(_MN, [], _ApiCtx, _KAcc, Acc) -> Acc.