MG-178: Add version to schema (#19)

* Fix upgrade mg proto

* Introduce machinery_backend_marshaller

* Revert "Introduce machinery_backend_marshaller"

This reverts commit cbc1c677c079ca0f48e1b053f618e5e9d6aec214.

* Rework schema versioning

rebar.config

@@ -35,7 +35,7 @@
         {git, "https://github.com/rbkmoney/woody_erlang.git", {branch, "master"}}
     },
     {mg_proto,
-        {git, "https://github.com:/bkmoney/machinegun_proto.git", {branch, "master"}}
+        {git, "https://github.com/rbkmoney/machinegun_proto.git", {branch, "master"}}
     }
 ]}.
 

rebar.lock

@@ -26,7 +26,7 @@
  {<<"metrics">>,{pkg,<<"metrics">>,<<"1.0.1">>},2},
  {<<"mg_proto">>,
   {git,"https://github.com/rbkmoney/machinegun_proto.git",
-       {ref,"ebae56fe2b3e79e4eb34afc8cb55c9012ae989f8"}},
+       {ref,"eac772bb8446fcd2f439232bf10fa086c336aca6"}},
   0},
  {<<"mimerl">>,{pkg,<<"mimerl">>,<<"1.2.0">>},2},
  {<<"parse_trans">>,{pkg,<<"parse_trans">>,<<"3.3.0">>},3},

src/machinery_mg_backend.erl

@@ -281,10 +281,11 @@ marshal(range, {Cursor, Limit, Direction}) ->
 marshal({history, Schema}, V) ->
     marshal({list, {event, Schema}}, V);
 marshal({event, Schema}, {EventID, CreatedAt, Body}) ->
+    Version = machinery_mg_schema:get_version(Schema, event),
     #mg_stateproc_Event{
         'id'         = marshal(event_id, EventID),
         'created_at' = marshal(timestamp, CreatedAt),
-        'data'       = marshal({schema, Schema, event}, Body)
+        'data'       = marshal({schema, Schema, {event, Version}}, Body)
     };
 
 marshal({signal, Schema}, {init, Args}) ->
@@ -310,6 +311,7 @@ marshal({call_result, Schema}, {Response, #{} = V}) ->
     };
 
 marshal({state_change, Schema}, #{} = V) ->
+    Version = machinery_mg_schema:get_version(Schema, aux_state),
     #mg_stateproc_MachineStateChange{
         events = [
             #mg_stateproc_Content{data = Event}
@@ -318,7 +320,7 @@ marshal({state_change, Schema}, #{} = V) ->
         % TODO
         % Provide this to logic handlers as well
         aux_state = #mg_stateproc_Content{
-            data = marshal({schema, Schema, aux_state}, maps:get(aux_state, V, undefined))
+            data = marshal({schema, Schema, {aux_state, Version}}, maps:get(aux_state, V, undefined))
         }
     };
 
@@ -467,7 +469,7 @@ unmarshal(
         'id'            = ID,
         'history'       = History,
         'history_range' = Range,
-        'aux_state'     = #mg_stateproc_Content{format_version = _Version, data = AuxState}
+        'aux_state'     = #mg_stateproc_Content{format_version = Version, data = AuxState}
     }
 ) ->
     #{
@@ -475,7 +477,7 @@ unmarshal(
         id              => unmarshal(id, ID),
         history         => unmarshal({history, Schema}, History),
         range           => unmarshal(range, Range),
-        aux_state       => unmarshal({maybe, {schema, Schema, aux_state}}, AuxState)
+        aux_state       => unmarshal({maybe, {schema, Schema, {aux_state, Version}}}, AuxState)
     };
 
 unmarshal({history, Schema}, V) ->
@@ -484,12 +486,17 @@ unmarshal({history, Schema}, V) ->
 unmarshal(
     {event, Schema},
     #mg_stateproc_Event{
-        'id'         = EventID,
-        'created_at' = CreatedAt,
-        'data'       = Payload
+        'id'             = EventID,
+        'created_at'     = CreatedAt,
+        'format_version' = Version,
+        'data'           = Payload
     }
 ) ->
-    {unmarshal(event_id, EventID), unmarshal(timestamp, CreatedAt), unmarshal({schema, Schema, event}, Payload)};
+    {
+        unmarshal(event_id, EventID),
+        unmarshal(timestamp, CreatedAt),
+        unmarshal({schema, Schema, {event, Version}}, Payload)
+    };
 
 unmarshal({signal, Schema}, {init, #mg_stateproc_InitSignal{arg = Args}}) ->
     {init, unmarshal({schema, Schema, {args, init}}, Args)};

src/machinery_mg_schema.erl

@@ -15,25 +15,34 @@
         call
     } |
     response |
-    event |
-    aux_state.
+    {event, Version} |
+    {aux_state, Version}.
 -type v(T) ::
     T.
 
+-type vt() :: aux_state | event.
+-type version() :: undefined | integer().
+
 -callback marshal(t(), v(_)) ->
     machinery_msgpack:t().
 
 -callback unmarshal(t(), machinery_msgpack:t()) ->
     v(_).
 
+-callback get_version(vt()) ->
+    version().
+
 -export_type([schema/0]).
 -export_type([t/0]).
 -export_type([v/1]).
+-export_type([vt/0]).
+-export_type([version/0]).
 
 %% API
 
 -export([marshal/3]).
 -export([unmarshal/3]).
+-export([get_version/2]).
 
 -spec marshal(schema(), t(), v(_)) ->
     machinery_msgpack:t().
@@ -44,3 +53,8 @@ marshal(Schema, T, V) ->
     v(_).
 unmarshal(Schema, T, V) ->
     Schema:unmarshal(T, V).
+
+-spec get_version(schema(), vt()) ->
+    version().
+get_version(Schema, T) ->
+    Schema:get_version(T).

src/machinery_mg_schema_generic.erl

@@ -14,6 +14,7 @@
 
 -export([marshal/2]).
 -export([unmarshal/2]).
+-export([get_version/1]).
 
 -import(machinery_msgpack, [
     nil/0,
@@ -36,6 +37,12 @@ marshal(_T, V) ->
 unmarshal(_T, V) ->
     unmarshal(V).
 
+-spec get_version(machinery_mg_schema:vt()) ->
+    machinery_mg_schema:version().
+
+get_version(_) ->
+    undefined.
+
 %% API
 
 -type eterm() ::