[WIP] Switch wallets to mg + implement as an event source

apps/fistful/src/ff_wallet.erl

@@ -7,17 +7,24 @@
 -type identity() :: ff_identity:identity().
 -type currency() :: ff_currency:id().
 -type wid()      :: ff_party:wallet().
--type id()       :: machinery:id().
+-type id(T)      :: T.
 
 -type wallet() :: #{
-    id       := id(),
+    id       := id(_),
     identity := identity(),
     name     := binary(),
     currency := currency(),
-    wid      := wid()
+    wid      => wid()
 }.
 
+-type ev() ::
+    {created, wallet()} |
+    {wid_set, wid()}.
+
+-type outcome() :: [ev()].
+
 -export_type([wallet/0]).
+-export_type([ev/0]).
 
 -export([id/1]).
 -export([identity/1]).
@@ -30,27 +37,33 @@
 -export([is_accessible/1]).
 -export([close/1]).
 
+-export([collapse_events/1]).
+-export([apply_events/2]).
+-export([apply_event/2]).
+
+-export([dehydrate/1]).
+-export([hydrate/2]).
+
 %% Pipeline
 
 -import(ff_pipeline, [do/1, unwrap/1, unwrap/2]).
 
 %% Accessors
 
--spec id(wallet())         -> id().
+-spec id(wallet())         -> id(_).
 -spec identity(wallet())   -> identity().
 -spec name(wallet())       -> binary().
 -spec currency(wallet())   -> currency().
--spec wid(wallet())        -> wid().
 
 id(#{id := V})             -> V.
 identity(#{identity := V}) -> V.
 name(#{name := V})         -> V.
 currency(#{currency := V}) -> V.
-wid(#{wid := V})           -> V.
 
--spec set_wid(wid(), wallet()) -> wallet().
+-spec wid(wallet()) -> ff_map:result(wid()).
 
-set_wid(V, W = #{}) -> W#{wid => V}.
+wid(Wallet) ->
+    ff_map:find(wid, Wallet).
 
 -spec account(wallet()) ->
     {ok, ff_transaction:account()} |
@@ -58,29 +71,27 @@ set_wid(V, W = #{}) -> W#{wid => V}.
 
 account(Wallet) ->
     do(fun () ->
-        unwrap(ff_party:get_wallet_account(
-            ff_identity:party(identity(Wallet)),
-            wid(Wallet)
-        ))
+        WID = unwrap(wid(Wallet)),
+        unwrap(ff_party:get_wallet_account(ff_identity:party(identity(Wallet)), WID))
     end).
 
 %%
 
--spec create(id(), identity(), binary(), currency()) ->
-    {ok, wallet()}.
+-spec create(id(_), identity(), binary(), currency()) ->
+    {ok, outcome()}.
 
 create(ID, Identity, Name, Currency) ->
     do(fun () ->
-        #{
+        [{created, #{
             id       => ID,
             identity => Identity,
             name     => Name,
             currency => Currency
-        }
+        }}]
     end).
 
 -spec setup_wallet(wallet()) ->
-    {ok, wallet()} |
+    {ok, outcome()} |
     {error,
         {inaccessible, blocked | suspended} |
         {contract, notfound} |
@@ -96,24 +107,31 @@ setup_wallet(Wallet) ->
             name     => name(Wallet),
             currency => currency(Wallet)
         },
+        % TODO
+        %  - There is an opportunity for a race where someone can block party
+        %    right before we create a party wallet.
         WID        = unwrap(ff_party:create_wallet(ff_identity:party(Identity), Contract, Prototype)),
-        set_wid(WID, Wallet)
+        [{wid_set, WID}]
     end).
 
 -spec is_accessible(wallet()) ->
     {ok, accessible} |
-    {error, {inaccessible, suspended | blocked}}.
+    {error,
+        {wid, notfound} |
+        {inaccessible, suspended | blocked}
+    }.
 
 is_accessible(Wallet) ->
     do(fun () ->
         Identity   = identity(Wallet),
+        WID        = unwrap(wid, wid(Wallet)),
         accessible = unwrap(ff_identity:is_accessible(Identity)),
-        accessible = unwrap(ff_party:is_wallet_accessible(ff_identity:party(Identity), wid(Wallet))),
+        accessible = unwrap(ff_party:is_wallet_accessible(ff_identity:party(Identity), WID)),
         accessible
     end).
 
 -spec close(wallet()) ->
-    ok |
+    {ok, outcome()} |
     {error,
         {inaccessible, blocked | suspended} |
         {account, pending}
@@ -123,5 +141,56 @@ close(Wallet) ->
     do(fun () ->
         accessible = unwrap(is_accessible(Wallet)),
         % TODO
-        ok
+        []
     end).
+
+%%
+
+-spec collapse_events([ev(), ...]) ->
+    wallet().
+
+collapse_events(Evs) when length(Evs) > 0 ->
+    apply_events(Evs, undefined).
+
+-spec apply_events([ev()], undefined | wallet()) ->
+    undefined | wallet().
+
+apply_events(Evs, Identity) ->
+    lists:foldl(fun apply_event/2, Identity, Evs).
+
+-spec apply_event(ev(), undefined | wallet()) ->
+    wallet().
+
+apply_event({created, Wallet}, undefined) ->
+    Wallet;
+apply_event({wid_set, WID}, Wallet) ->
+    Wallet#{wid => WID}.
+
+%%
+
+-spec dehydrate(ev()) ->
+    term().
+
+-spec hydrate(term(), undefined | wallet()) ->
+    ev().
+
+dehydrate({created, Wallet}) ->
+    {created, #{
+        id       => id(Wallet),
+        name     => name(Wallet),
+        identity => ff_identity:id(identity(Wallet)),
+        currency => currency(Wallet)
+    }};
+dehydrate({wid_set, WID}) ->
+    {wid_set, WID}.
+
+hydrate({created, V}, undefined) ->
+    {ok, IdentitySt} = ff_identity_machine:get(maps:get(identity, V)),
+    {created, #{
+        id       => maps:get(id, V),
+        name     => maps:get(name, V),
+        identity => ff_identity_machine:identity(IdentitySt),
+        currency => maps:get(currency, V)
+    }};
+hydrate({wid_set, WID}, _) ->
+    {wid_set, WID}.

apps/fistful/src/ff_wallet_machine.erl

@@ -3,7 +3,8 @@
 %%%
 %%% TODOs
 %%%
-%%%  - Pattern `NS, ID, Backend` repeats everytime.
+%%%  - ~~Pattern `NS, ID, Backend` repeats everytime.~~ Well, there's just `NS`
+%%%    instead but it looks not so bright still.
 %%%
 
 -module(ff_wallet_machine).
@@ -81,9 +82,9 @@ create(ID, #{identity := IdentityID, name := Name, currency := Currency}, Ctx) -
     do(fun () ->
         Identity = ff_identity_machine:identity(unwrap(identity, ff_identity_machine:get(IdentityID))),
         _        = unwrap(currency, ff_currency:get(Currency)),
-        Wallet0  = unwrap(ff_wallet:create(ID, Identity, Name, Currency)),
-        Wallet1  = unwrap(ff_wallet:setup_wallet(Wallet0)),
-        unwrap(machinery:start(?NS, ID, {Wallet1, Ctx}, backend()))
+        Events0  = unwrap(ff_wallet:create(ID, Identity, Name, Currency)),
+        Events1  = unwrap(ff_wallet:setup_wallet(ff_wallet:collapse_events(Events0))),
+        unwrap(machinery:start(?NS, ID, {Events0 ++ Events1, Ctx}, backend()))
     end).
 
 -spec get(id()) ->
@@ -100,22 +101,23 @@ backend() ->
 
 %% machinery
 
--type ev()           ::
-    {created, wallet()}.
+-type ev() ::
+    ff_wallet:ev().
 
 -type auxst() ::
     #{ctx => ctx()}.
 
--type machine()      :: machinery:machine(ev(), auxst()).
--type result()       :: machinery:result(ev(), auxst()).
+-type ts_ev(T)       :: {ev, timestamp(), T}.
+-type machine()      :: machinery:machine(ts_ev(ev()), auxst()).
+-type result()       :: machinery:result(ts_ev(ev()), auxst()).
 -type handler_opts() :: machinery:handler_opts(_).
 
--spec init({wallet(), ctx()}, machine(), _, handler_opts()) ->
+-spec init({[ev()], ctx()}, machine(), _, handler_opts()) ->
     result().
 
-init({Wallet, Ctx}, #{}, _, _Opts) ->
+init({Events, Ctx}, #{}, _, _Opts) ->
     #{
-        events    => emit_ts_event({created, Wallet}),
+        events    => emit_ts_events(Events),
         aux_state => #{ctx => Ctx}
     }.
 
@@ -141,23 +143,23 @@ collapse_history(History, St) ->
 
 merge_event({_ID, _Ts, TsEv}, St0) ->
     {EvBody, St1} = merge_ts_event(TsEv, St0),
-    merge_event_body(EvBody, St1).
+    merge_event_body(ff_wallet:hydrate(EvBody, maybe(wallet, St1)), St1).
 
-merge_event_body({created, Wallet}, St) ->
+merge_event_body(Ev, St) ->
     St#{
-        wallet => Wallet
+        wallet => ff_wallet:apply_event(Ev, maybe(wallet, St))
     }.
 
-%%
+maybe(wallet, St) ->
+    maps:get(wallet, St, undefined).
 
-emit_ts_event(E) ->
-    emit_ts_events([E]).
+%%
 
 emit_ts_events(Es) ->
     emit_ts_events(Es, machinery_time:now()).
 
 emit_ts_events(Es, Ts) ->
-    [{ev, Ts, Body} || Body <- Es].
+    [{ev, Ts, ff_wallet:dehydrate(Body)} || Body <- Es].
 
 merge_ts_event({ev, Ts, Body}, St = #{times := {Created, _Updated}}) ->
     {Body, St#{times => {Created, Ts}}};

apps/fistful/test/ff_wallet_SUITE.erl

@@ -39,11 +39,14 @@ all() ->
 -spec init_per_suite(config()) -> config().
 
 init_per_suite(C) ->
-    IBO = #{name => {?MODULE, identities}},
-    WBO = #{name => {?MODULE, wallets}},
+    BeConf = #{schema => machinery_mg_schema_generic},
+    Be = {machinery_mg_backend, BeConf#{
+        client => ff_woody_client:new("http://machinegun:8022/v1/automaton")
+    }},
     {StartedApps, _StartupCtx} = ct_helper:start_apps([
         lager,
         scoper,
+        woody,
         {dmt_client, [
             {max_cache_size, #{
                 elements => 1
@@ -59,8 +62,8 @@ init_per_suite(C) ->
                 'accounter' => ff_woody_client:new("http://shumway:8022/accounter")
             }},
             {backends, #{
-                'identity'  => machinery_gensrv_backend:new(IBO),
-                'wallet'    => machinery_gensrv_backend:new(WBO)
+                'ff/identity'  => Be,
+                'ff/wallet'    => Be
             }},
             {providers,
                 get_provider_config()
@@ -68,8 +71,25 @@ init_per_suite(C) ->
         ]}
     ]),
     SuiteSup = ct_sup:start(),
-    {ok, _} = supervisor:start_child(SuiteSup, machinery_gensrv_backend:child_spec(ff_identity_machine, IBO)),
-    {ok, _} = supervisor:start_child(SuiteSup, machinery_gensrv_backend:child_spec(ff_wallet_machine, WBO)),
+    BeOpts = #{event_handler => scoper_woody_event_handler},
+    Routes = machinery_mg_backend:get_routes(
+        [
+            {{fistful, ff_identity_machine},
+                #{path => <<"/v1/stateproc/identity">>, backend_config => BeConf}},
+            {{fistful, ff_wallet_machine},
+                #{path => <<"/v1/stateproc/wallet">>, backend_config => BeConf}}
+        ],
+        BeOpts
+    ),
+    {ok, _} = supervisor:start_child(SuiteSup, woody_server:child_spec(
+        ?MODULE,
+        BeOpts#{
+            ip                => {0, 0, 0, 0},
+            port              => 8022,
+            handlers          => [],
+            additional_routes => Routes
+        }
+    )),
     C1 = ct_helper:makeup_cfg(
         [ct_helper:test_case_name(init), ct_helper:woody_ctx()],
         [
@@ -81,6 +101,7 @@ init_per_suite(C) ->
         | C]
     ),
     ok = ct_domain_config:upsert(get_domain_config(C1)),
+    ok = timer:sleep(1000),
     C1.
 
 -spec end_per_suite(config()) -> _.
@@ -255,6 +276,9 @@ get_domain_config(C) ->
 
 get_default_termset() ->
     #domain_TermSet{
+        % TODO
+        %  - Strangely enough, hellgate checks wallet currency against _payments_
+        %    terms.
         payments = #domain_PaymentsServiceTerms{
             currencies = {value, ?ordset([?cur(<<"RUB">>)])},
             categories = {value, ?ordset([?cat(1)])},