Adds WIP on grouped operations by finalization behaviour

apps/limiter/src/lim_config_machine.erl

@@ -60,12 +60,14 @@
     scope => limit_scope(),
     description => description(),
     op_behaviour => op_behaviour(),
-    currency_conversion => currency_conversion()
+    currency_conversion => currency_conversion(),
+    finalization_behaviour => finalization_behaviour()
 }.
 
 -type op_behaviour() :: #{operation_type() := addition | subtraction}.
 -type operation_type() :: invoice_payment_refund.
 -type currency_conversion() :: boolean().
+-type finalization_behaviour() :: normal | {invertable, session_presence}.
 
 -type lim_id() :: limproto_limiter_thrift:'LimitID'().
 -type lim_version() :: dmsl_domain_thrift:'DataRevision'().
@@ -74,6 +76,7 @@
 -type timestamp() :: dmsl_base_thrift:'Timestamp'().
 -type operation_id() :: limproto_limiter_thrift:'OperationID'().
 -type lim_changes() :: [lim_change()].
+-type changes_group() :: {context_type(), finalization_behaviour()}.
 
 -export_type([config/0]).
 -export_type([limit_type/0]).
@@ -134,6 +137,12 @@ scope(_) ->
 context_type(#{context_type := Value}) ->
     Value.
 
+-spec finalization_behaviour(config()) -> finalization_behaviour().
+finalization_behaviour(#{finalization_behaviour := Value}) ->
+    Value;
+finalization_behaviour(_) ->
+    normal.
+
 -spec currency_conversion(config()) -> currency_conversion().
 currency_conversion(#{currency_conversion := Value}) ->
     Value;
@@ -155,42 +164,116 @@ get_values(LimitChanges, LimitContext) ->
     {ok, [lim_liminator:limit_response()]} | {error, config_error() | {processor(), get_limit_error()}}.
 get_batch(OperationID, LimitChanges, LimitContext) ->
     do(fun() ->
-        unwrap(
-            OperationID,
-            lim_liminator:get(OperationID, unwrap(collect_changes(hold, LimitChanges, LimitContext)), LimitContext)
-        )
+        GroupedChanges = unwrap(collect_grouped_changes(hold, LimitChanges, LimitContext)),
+        F = fun(Group, Changes) ->
+            OperationIDForGroup = operation_id_for_group(OperationID, Group),
+            unwrap(OperationID, lim_liminator:get(OperationIDForGroup, Changes, LimitContext))
+        end,
+        lists:flatten(maps:values(maps:map(F, GroupedChanges)))
     end).
 
 -spec hold_batch(operation_id(), lim_changes(), lim_context()) ->
-    {ok, [lim_liminator:limit_response()]} | {error, config_error() | {processor(), hold_error()}}.
+    {ok, [lim_liminator:limit_response()]}
+    | {error, config_error() | {processor(), hold_error()} | {operation_id(), lim_liminator:invalid_request_error()}}.
 hold_batch(OperationID, LimitChanges, LimitContext) ->
     do(fun() ->
-        unwrap(
-            OperationID,
-            lim_liminator:hold(OperationID, unwrap(collect_changes(hold, LimitChanges, LimitContext)), LimitContext)
-        )
+        GroupedChanges = unwrap(collect_grouped_changes(hold, LimitChanges, LimitContext)),
+        F = fun(Group, Changes) ->
+            OperationIDForGroup = operation_id_for_group(OperationID, Group),
+            unwrap(OperationID, lim_liminator:hold(OperationIDForGroup, Changes, LimitContext))
+        end,
+        lists:flatten(maps:values(maps:map(F, GroupedChanges)))
     end).
 
 -spec commit_batch(operation_id(), lim_changes(), lim_context()) ->
-    ok | {error, config_error() | {processor(), commit_error()}}.
+    ok
+    | {error, config_error() | {processor(), commit_error()} | {operation_id(), lim_liminator:invalid_request_error()}}.
 commit_batch(OperationID, LimitChanges, LimitContext) ->
     do(fun() ->
-        unwrap(
-            OperationID,
-            lim_liminator:commit(OperationID, unwrap(collect_changes(commit, LimitChanges, LimitContext)), LimitContext)
-        )
+        GroupedChanges = unwrap(collect_grouped_changes(commit, LimitChanges, LimitContext)),
+        F = fun(Group, Changes) ->
+            OperationIDForGroup = operation_id_for_group(OperationID, Group),
+            Behaviour = resolve_group_finalization_behaviour(Group, LimitContext),
+            unwrap(OperationID, finalize(OperationIDForGroup, Changes, LimitContext, commit, Behaviour))
+        end,
+        maps:foreach(F, GroupedChanges)
     end).
 
 -spec rollback_batch(operation_id(), lim_changes(), lim_context()) ->
-    ok | {error, config_error() | {processor(), rollback_error()}}.
+    ok
+    | {error,
+        config_error() | {processor(), rollback_error()} | {operation_id(), lim_liminator:invalid_request_error()}}.
 rollback_batch(OperationID, LimitChanges, LimitContext) ->
     do(fun() ->
-        unwrap(
-            OperationID,
-            lim_liminator:rollback(OperationID, unwrap(collect_changes(hold, LimitChanges, LimitContext)), LimitContext)
-        )
+        GroupedChanges = unwrap(collect_grouped_changes(hold, LimitChanges, LimitContext)),
+        F = fun(Group, Changes) ->
+            OperationIDForGroup = operation_id_for_group(OperationID, Group),
+            Behaviour = resolve_group_finalization_behaviour(Group, LimitContext),
+            unwrap(OperationID, finalize(OperationIDForGroup, Changes, LimitContext, rollback, Behaviour))
+        end,
+        maps:foreach(F, GroupedChanges)
     end).
 
+finalize(OperationIDForGroup, Changes, LimitContext, commit, normal) ->
+    lim_liminator:commit(OperationIDForGroup, Changes, LimitContext);
+finalize(OperationIDForGroup, Changes, LimitContext, commit, inverted) ->
+    lim_liminator:rollback(OperationIDForGroup, Changes, LimitContext);
+finalize(OperationIDForGroup, Changes, LimitContext, rollback, normal) ->
+    lim_liminator:rollback(OperationIDForGroup, Changes, LimitContext);
+finalize(OperationIDForGroup, Changes, LimitContext, rollback, inverted) ->
+    lim_liminator:commit(OperationIDForGroup, Changes, LimitContext).
+
+-spec resolve_group_finalization_behaviour(changes_group(), lim_context()) -> normal | inverted.
+resolve_group_finalization_behaviour({_, normal}, _) ->
+    normal;
+resolve_group_finalization_behaviour({ContextType, {invertable, session_presence}}, LimitContext) ->
+    case lim_context:get_value(ContextType, session, LimitContext) of
+        {ok, undefined} ->
+            normal;
+        {ok, _Some} ->
+            inverted;
+        {error, {unsupported, _}} ->
+            %% If context doesn't support session value then we treat it as
+            %% normal finalization.
+            normal;
+        {error, notfound} ->
+            normal
+    end.
+
+operation_id_for_group(OperationID, {_, normal}) ->
+    OperationID;
+operation_id_for_group(OperationID, {_, {invertable, session_presence}}) ->
+    <<OperationID/binary, "/inverted/session-presence">>.
+
+-spec collect_grouped_changes(hold | commit, lim_changes(), lim_context()) ->
+    {ok, #{changes_group() => lim_changes()}} | {error, config_error() | {processor(), get_limit_error()}}.
+collect_grouped_changes(Stage, LimitChanges, LimitContext) ->
+    collect_grouped_changes(Stage, LimitChanges, LimitContext, #{}).
+
+collect_grouped_changes(_, [], _, Acc) ->
+    {ok, Acc};
+collect_grouped_changes(Stage, [LimitChange | Other], LimitContext, Acc0) ->
+    do(fun() ->
+        #limiter_LimitChange{id = ID, version = Version} = LimitChange,
+        %% NOTE We use only "woody_context" of given limit context to get
+        %% handler.
+        {Handler, Config} = unwrap(get_handler(ID, Version, LimitContext)),
+        Change = unwrap(Handler, Handler:make_change(Stage, LimitChange, Config, LimitContext)),
+        %% NOTE Because rules to resolve change's finalization behaviour can be
+        %% dependent on operation's context ("context" key of given limit
+        %% context map), each group is discriminated by bothe context type and
+        %% finalization behaviour values.
+        %% Thus, we must ensure changes are in appropriate group before we
+        %% evalute groups behaviour against operation's limit context.
+        ContextType = context_type(Config),
+        FinalizationBehaviour = finalization_behaviour(Config),
+        Group = {ContextType, FinalizationBehaviour},
+        Acc1 = maps:update_with(Group, fun(Changes) -> [Change | Changes] end, [Change], Acc0),
+        unwrap(collect_grouped_changes(Stage, Other, LimitContext, Acc1))
+    end).
+
+%% NOTE Used only for `get_values/2' function that doesn't care about groups of
+%% changes, nor resulting list's order.
 collect_changes(_Stage, [], _LimitContext) ->
     {ok, []};
 collect_changes(Stage, [LimitChange = #limiter_LimitChange{id = ID, version = Version} | Other], LimitContext) ->
@@ -503,7 +586,8 @@ unmarshal_limit_config(#domain_LimitConfigObject{
         scopes = Scopes,
         description = Description,
         op_behaviour = OpBehaviour,
-        currency_conversion = CurrencyConversion
+        currency_conversion = CurrencyConversion,
+        finalization_behaviour = FinalizationBehaviour
     }
 }) ->
     genlib_map:compact(#{
@@ -517,9 +601,17 @@ unmarshal_limit_config(#domain_LimitConfigObject{
         scope => maybe_apply(Scopes, fun unmarshal_scope/1),
         description => Description,
         op_behaviour => maybe_apply(OpBehaviour, fun unmarshal_op_behaviour/1),
-        currency_conversion => CurrencyConversion =/= undefined
+        currency_conversion => CurrencyConversion =/= undefined,
+        finalization_behaviour => unmarshal_finalization_behaviour(FinalizationBehaviour)
     }).
 
+unmarshal_finalization_behaviour(undefined) ->
+    normal;
+unmarshal_finalization_behaviour({normal, #limiter_config_Normal{}}) ->
+    normal;
+unmarshal_finalization_behaviour({invertable, {session_presence, #limiter_config_Inversed{}}}) ->
+    {invertable, session_presence}.
+
 unmarshal_time_range_type({calendar, CalendarType}) ->
     {calendar, unmarshal_calendar_time_range_type(CalendarType)};
 unmarshal_time_range_type({interval, #limiter_config_TimeRangeTypeInterval{amount = Amount}}) ->
@@ -619,7 +711,8 @@ unmarshal_config_object_test() ->
         type => {turnover, number},
         scope => ordsets:from_list([party, shop, {destination_field, [<<"path">>, <<"to">>, <<"field">>]}]),
         description => <<"description">>,
-        currency_conversion => true
+        currency_conversion => true,
+        finalization_behaviour => {invertable, session_presence}
     },
     Object = #domain_LimitConfigObject{
         ref = #domain_LimitConfigRef{id = <<"id">>},
@@ -639,7 +732,8 @@ unmarshal_config_object_test() ->
                 }}
             ]),
             description = <<"description">>,
-            currency_conversion = #limiter_config_CurrencyConversion{}
+            currency_conversion = #limiter_config_CurrencyConversion{},
+            finalization_behaviour = {invertable, {session_presence, #limiter_config_Inversed{}}}
         }
     },
     ?assertEqual(Config, unmarshal_limit_config(Object)).