fixes for discovery and tpicapi, test are passed

cleverfox · cleverfox · commit 555216f2a348 · 2023-04-08T16:29:27.000+03:00
diff --git a/apps/tpic2/src/tpic2_client.erl b/apps/tpic2/src/tpic2_client.erl
@@ -74,10 +74,25 @@ connection_process(Parent, Host, Port, Opts) ->
         {ok, Socket} = ssl:connect(TCPSocket, SSLOpts),
         ssl:setopts(Socket, [{active, once}]),
         {ok,PeerInfo}=ssl:connection_information(Socket),
+        PeerPK=case ssl:peercert(Socket) of
+                 {ok, PC} ->
+                   DCert=tpic2:extract_cert_info(public_key:pkix_decode_cert(PC,otp)),
+                   case DCert of
+                     #{pubkey:=Der} ->
+                       Der;
+                     _ ->
+                       ?LOG_NOTICE("Unknown cert ~p",[DCert]),
+                       undefined
+                   end;
+                 {error, no_peercert} ->
+                   undefined
+               end,
+
         State=#{
                 ref=>maps:get(ref, Opts, undefined),
                 socket=>Socket,
                 peerinfo=>PeerInfo,
+                pubkey=>PeerPK,
                 timer=>undefined,
                 transport=>ranch_ssl,
                 parent=>Parent,
diff --git a/apps/tpic2/src/tpic2_tls.erl b/apps/tpic2/src/tpic2_tls.erl
@@ -81,7 +81,7 @@ conn_proto(Parent, Ref, Socket, Transport, Opts, <<"tpic2">>, PeerPK) ->
           timer=>undefined,
           transport=>Transport,
           protocol => tpic2,
-          peerpk => PeerPK,
+          pubkey => PeerPK,
           nodeid=> try
                      nodekey:get_pub()
                    catch _:_ -> atom_to_binary(node(),utf8)
@@ -96,7 +96,7 @@ conn_proto(_Parent, _Ref, Socket, Transport, _Opts, _, _) ->
   Transport:close(Socket).
 
 loop1(State=#{socket:=Socket,role:=Role,opts:=Opts,
-              transport:=Transport,peerpk:=Pubkey}) ->
+              transport:=Transport,pubkey:=Pubkey}) ->
   %{ok,PC}=ssl:peercert(Socket),
   %DCert=tpic2:extract_cert_info(public_key:pkix_decode_cert(PC,otp)),
   %Pubkey=case DCert of
@@ -166,7 +166,7 @@ loop1(State=#{socket:=Socket,role:=Role,opts:=Opts,
       if WhatToDo==shutdown ->
            done;
          true ->
-           ?MODULE:loop(State#{pubkey=>Pubkey})
+           ?MODULE:loop(State)
       end
   end.
 
@@ -318,7 +318,9 @@ handle_msg(#{null:=<<"hello">>,
                   OldSID
                  }
       end,
-  {ok, PPID}=gen_server:call(tpic2_cmgr, {peer,PK, Reg}),
+  {_OkOrExists,PPID}=gen_server:call(tpic2_cmgr, {peer,PK, Reg}),
+  %logger:notice("RegRes ~p ~p: ~p",[PK, Reg, RegRes]),
+  %{ok, PPID}=RegRes,
   lists:foreach(fun(Addr) ->
                     gen_server:call(PPID, {add, binary_to_list(Addr), Port})
                 end,
diff --git a/apps/tpnode/src/discovery.erl b/apps/tpnode/src/discovery.erl
@@ -165,16 +165,16 @@ handle_call({get_pid, Name}, _From, #{local_services:=Dict} = State) when is_bin
             end,
     {reply, Reply, State};
 
-handle_call({lookup, Pred}, _From, State) when is_function(Pred) ->
-    {reply, query(Pred, State), State};
-
 %% get list of ip and port for service with name Name (local and remote)
 handle_call({lookup, Name}, _From, State) ->
     {reply, query(Name, State), State};
 
 handle_call({lookup, Name, Chain}, _From, State) ->
     {reply, query(Name, Chain, State), State};
 
+handle_call({lookup_raw, Name, Chain}, _From, State) ->
+    {reply, query_raw(Name, Chain, State), State};
+
 handle_call({lookup_remote, Name}, _From, #{remote_services := RemoteDict} = State) ->
   {reply, query_remote(Name, RemoteDict, blockchain:chain()), State};
 
@@ -331,7 +331,7 @@ get_local_addresses(State) ->
 
 
 % --------------------------------------------------------
-announce_one_service(Name, TranslatedAddress, Ttl, Scopes) ->
+prepare_announce_one_service(Name, TranslatedAddress, Ttl, Scopes) ->
     try
 %%        TranslatedAddress = add_hostname(translate_address(Address), Hostname),
 
@@ -350,16 +350,26 @@ announce_one_service(Name, TranslatedAddress, Ttl, Scopes) ->
             scopes => Scopes,
             chain => blockchain:chain()
         },
-        AnnounceBin = pack(Announce),
-        send_service_announce(AnnounceBin)
+        {ok, pack(Announce)}
     catch
         Err:Reason ->
             ?LOG_ERROR(
                 "Announce with name ~p and address ~p and scopes ~p hasn't made because ~p ~p",
                 [Name, TranslatedAddress, Scopes, Err, Reason]
-            )
+            ),
+            error
     end.
 
+%announce_one_service(Name, TranslatedAddress, Ttl, Scopes) ->
+%  case prepare_announce_one_service(Name, TranslatedAddress, Ttl, Scopes) of
+%    {ok, AnnounceBin} ->
+%        send_service_announce(AnnounceBin),
+%        ok;
+%    error ->
+%      error
+%  end.
+
+
 % ------------------------------------------------------------
 -spec is_local_service(Announce :: #{ 'nodeid' := _, _ := _ }) -> boolean().
 
@@ -418,49 +428,60 @@ get_local_names(Names) ->
 % --------------------------------------------------------
 
 % make announce of our local services with tpic scope
-make_announce(#{names:=Names} = _Dict, State) ->
-  ?LOG_DEBUG("Announcing our local services"),
+prepare_announce(#{names:=Names} = _Dict, State) ->
   Ttl = max(get_config(intrachain_ttl, 300, State), 30),
   Hostname = application:get_env(tpnode, hostname, unknown),
-%%    ValidUntil = os:system_time(seconds) + get_config(intrachain_ttl, 120, State),
   Addresses = get_config(addresses, get_default_addresses(), State),
   AllScopesCfg = get_config(scope, ?DEFAULT_SCOPE_CONFIG, State),
   MacroDict = get_config(macro_dict, #{}, State),
   LocalNames = get_local_names(Names),
-  Announcer = fun(Name, Counter) ->
-    Counter + lists:foldl(
-      % #{address => local4, port => 53221, proto => tpic}
-      fun(#{proto := Proto} = Address, AddrCounter) ->
-        Scopes = get_scopes(Proto, AllScopesCfg),
-        IsAdvertisable = in_scope(Proto, tpic, AllScopesCfg),
-        IsRightProto = is_right_proto(Name, Proto),
-%%        ?LOG_DEBUG("ann dbg ~p ~p ~p ~p", [Name, IsAdvertisable, IsRightProto, Address]),
-
-        if
-          IsRightProto == true andalso IsAdvertisable == true ->
-            try
-              TranslatedAddress =
-                add_hostname(substitute_macro(Address, MacroDict), Hostname),
-              announce_one_service(Name, TranslatedAddress, Ttl, Scopes),
-              AddrCounter + 1
-            catch
-              pass ->
-                ?LOG_DEBUG("skip address (can't substitute macro?): ~p", [Address]),
-                AddrCounter
-            end;
+  Announcer = fun(Name, ListAcc) ->
+                  lists:foldl(
+                    % #{address => local4, port => 53221, proto => tpic}
+                    fun(#{proto := Proto} = Address, Acc) ->
+                        Scopes = get_scopes(Proto, AllScopesCfg),
+                        IsAdvertisable = in_scope(Proto, tpic, AllScopesCfg),
+                        IsRightProto = is_right_proto(Name, Proto),
+                        if IsRightProto == true andalso IsAdvertisable == true ->
+                            try
+                              TranslatedAddress =
+                              add_hostname(substitute_macro(Address, MacroDict), Hostname),
+                              case prepare_announce_one_service(
+                                     Name,
+                                     TranslatedAddress,
+                                     Ttl,
+                                     Scopes) of
+                                {ok, Bin} ->
+                                  [Bin|Acc];
+                                error ->
+                                  Acc
+                              end
+                            catch
+                              pass ->
+                                ?LOG_DEBUG("skip address (can't substitute macro?): ~p", [Address]),
+                                Acc
+                            end;
+                          true ->
+                            Acc
+                        end;
+                       (Address, Acc) ->
+                        ?LOG_DEBUG("skip announce for invalid address ~p ~p", [Name, Address]),
+                        Acc
+                    end,
+                    ListAcc,
+                    Addresses)
+              end,
+  lists:foldl(Announcer, [], LocalNames).
 
-          true ->
-%%            ?LOG_DEBUG("skip announce for address ~p ~p", [Name, Address]),
-            AddrCounter
-        end;
-        (Address, AddrCounter) ->
-          ?LOG_DEBUG("skip announce for invalid address ~p ~p", [Name, Address]),
-          AddrCounter
-      end,
-      0,
-      Addresses)
+
+make_announce(Dict, State) ->
+  ?LOG_DEBUG("Announcing our local services"),
+  List2Announce = prepare_announce(Dict, State),
+  Announcer = fun(Bin, Cnt) ->
+                  send_service_announce(Bin),
+                  Cnt+1
               end,
-  ServicesCount = lists:foldl(Announcer, 0, LocalNames),
+  ServicesCount = lists:foldl(Announcer, 0, List2Announce),
   ?LOG_DEBUG("Announced ~p of our services", [ServicesCount]),
   ok.
 
@@ -756,11 +777,42 @@ query_remote(Name, _Dict, Chain) ->
 
 % --------------------------------------------------------
 
+query_raw(Name0, Chain0, State) ->
+    Name = convert_to_binary(Name0),
+    LC=blockchain:chain(),
+    #{local_services := #{names:=LocalDict}, remote_services := RemoteDict} = State,
+    {Chain,Local} = if is_integer(Chain0) andalso Chain0>0 andalso Chain0=/=LC ->
+                         {Chain0,[]};
+                       true ->
+                         {LC, 
+                          prepare_announce(
+                            #{names=>
+                              maps:with([Name0],LocalDict)
+                             }, State)
+                         }
+                    end,
+    RQR=fun(RName0, Dict) ->
+            Name1 = add_chain_to_name(RName0, Chain),
+            Nodes = maps:get(Name1, Dict, #{}),
+            Announces = maps:values(Nodes),
+            lists:map(
+              fun(#{bin:=B}) ->
+                  %maps:merge(Address, maps:with([nodeid,node_name,pubkey],A))
+                  B
+              end, Announces
+             )
+        end,
+
+    lists:merge(Local, RQR(Name,RemoteDict)).
+
+
 query(Name0, Chain, State) ->
     Name = convert_to_binary(Name0),
     LocalChain = blockchain:chain(),
     #{local_services := LocalDict, remote_services := RemoteDict} = State,
     Local = case Chain of
+        0 ->
+            query_local(Name, LocalDict, State);
         LocalChain ->
             query_local(Name, LocalDict, State);
         _ ->
@@ -774,11 +826,6 @@ query(Name0, Chain, State) ->
 
 % --------------------------------------------------------
 
-query(Pred, _State) when is_function(Pred) ->
-    ?LOG_ERROR("Not inmplemented"),
-    not_implemented;
-
-% find service by name
 query(Name, State) ->
     query(Name, blockchain:chain(), State).
 
@@ -870,7 +917,7 @@ process_announce(
     try
         Key = address2key(Address),
         Name = add_chain_to_name(Name0, Chain),
-        Announce = add_valid_until(Announce0, MaxTtl),
+        Announce = add_valid_until(Announce0#{bin=>AnnounceBin}, MaxTtl),
         Nodes = maps:get(Name, Dict, #{}),
         PrevAnnounce = maps:get(Key, Nodes, #{created => 0, ttl=> 0, sent_xchain => 0}),
         SentXchain = relay_announce(PrevAnnounce, Announce, AnnounceBin, XChainThrottle),
diff --git a/apps/tpnode/src/tpnode_tpicapi.erl b/apps/tpnode/src/tpnode_tpicapi.erl
@@ -30,7 +30,9 @@ init(Req0, _) ->
         throw:{return,RCode,RBody} ->
           {ok, cowboy_req:reply(RCode, #{}, RBody, Req0), #{} }
   end.
-  
+
+%% -- [ pick_block ] --
+
 handle(<<"GET">>, [<<"pick_block">>,Hash], _Req) ->
   handle(<<"GET">>, [<<"pick_block">>,Hash,<<"self">>], _Req);
 
@@ -56,13 +58,36 @@ handle(<<"GET">>, [<<"pick_block">>,THash,TRel], _Req) ->
       {404, <<"not found">>}
   end;
 
+%% -- [ sync ] --
+
 handle(<<"GET">>,[<<"sync">>,<<"request">>], _Req) ->
   {200,
-   #{<<"content-type">> =><<"binary/msgpack">>},
+   #{<<"content-type">> =><<"application/msgpack">>},
    msgpack:pack(gen_server:call(blockchain_reader,sync_req))
   };
 
-handle(<<"GET">>,[<<"txpool">>,TxID], _Req) ->
+%% -- [ discovery ] --
+
+handle(<<"GET">>,[<<"discovery">>,<<"tpicpeer">>,_BChain]=Path,Req) ->
+  case Req of
+    #{cert:=Cert} when is_binary(Cert) ->
+      handle(get, Path ,Req);
+    _ ->
+      {403, <<"unauth">>}
+  end;
+
+handle(<<"GET">>,[<<"discovery">>,Service,BChain], Req) ->
+  handle(get,[<<"discovery">>,Service,BChain], Req);
+
+handle(get,[<<"discovery">>,Service,BChain], _Req) ->
+  Chain=binary_to_integer(BChain),
+  List=gen_server:call(discovery,{lookup_raw,Service,Chain}),
+  Body=msgpack:pack(List),
+  {200, #{ <<"content-type">> => <<"application/msgpack">> }, Body};
+
+%% -- [ txstorage ] --
+
+handle(<<"GET">>,[<<"txstorage">>,TxID], _Req) ->
   case tpnode_txstorage:get_txm(TxID) of
     {ok, #{body:=Body,
            origin:=Origin,
@@ -79,7 +104,7 @@ handle(<<"GET">>,[<<"txpool">>,TxID], _Req) ->
       {404, <<"not found">>}
   end;
 
-handle(<<"PATCH">>,[<<"txpool">>,TxID], #{cert:=Cert}=_Req) when is_binary(Cert) ->
+handle(<<"PATCH">>,[<<"txstorage">>,TxID], #{cert:=Cert}=_Req) when is_binary(Cert) ->
   #{pubkey:=PubKey}=tpic2:extract_cert_info(public_key:pkix_decode_cert(Cert,otp)),
   case tpnode_txstorage:get_txm(TxID) of
     {ok, #{origin:=Origin, valid:=_Valid}} when Origin==PubKey ->
@@ -100,8 +125,7 @@ handle(<<"PATCH">>,[<<"txpool">>,TxID], #{cert:=Cert}=_Req) when is_binary(Cert)
       {404, <<"not found">>}
   end;
 
-
-handle(<<"PUT">>,[<<"txpool">>, TxID], #{cert:=Cert,has_body:=true}=Req) when is_binary(Cert) ->
+handle(<<"PUT">>,[<<"txstorage">>, TxID], #{cert:=Cert,has_body:=true}=Req) when is_binary(Cert) ->
   {ok, Body, Req1} = cowboy_req:read_body(Req),
   #{pubkey:=PubKey}=tpic2:extract_cert_info(public_key:pkix_decode_cert(Cert,otp)),
   case gen_server:call(txstorage,
@@ -117,6 +141,46 @@ handle(<<"PUT">>,[<<"txpool">>, TxID], #{cert:=Cert,has_body:=true}=Req) when is
       {{400, <<"body_mismatch">>},Req1}
   end;
 
+% -- [ tx ] --
+
+handle(<<"PUT">>, [<<"tx">>,<<"multi">>], #{has_body := true} = Req) ->
+  {ok, Body, Req1} = cowboy_req:read_body(Req),
+  { case maps:get(<<"content-type">>,Req1,undefined) of
+      <<"application/msgpack">> ->
+        {ok,Lst} = msgpack:unpack(Body),
+        if(is_list(Lst)) -> ok;
+          true -> throw({return,400,<<"nolist">>})
+        end,
+        Res=lists:map(
+              fun(Bin) when is_binary(Bin) ->
+                  case txpool:new_tx(Bin) of
+                    {ok, TxID} ->
+                      TxID;
+                    {error, Err} ->
+                      [<<"error">>,
+                       iolist_to_binary(io_lib:format("~p", [Err]))
+                      ]
+                  end
+              end, Lst),
+        {200, #{<<"content-type">> =><<"application/msgpack">>}, msgpack:pack(Res) };
+      _ ->
+        {400, <<"unexpected content-type">>}
+    end, Req};
+
+%
+handle(<<"PUT">>, [<<"tx">>], #{has_body := true} = Req) ->
+  {ok, Body, Req1} = cowboy_req:read_body(Req),
+  { case txpool:new_tx(Body) of
+      {ok, TxID} ->
+        {200,TxID};
+      {error, Err} ->
+        {500,
+         iolist_to_binary(io_lib:format("~p", [Err]))
+        }
+    end, Req1};
+
+% -- [ status ] --
+
 handle(<<"GET">>,[<<"status">>], #{cert:=Cert}) when is_binary(Cert) ->
   #{pubkey:=PubKey}=tpic2:extract_cert_info(public_key:pkix_decode_cert(Cert,otp)),
   {200,#{
diff --git a/apps/tpnode/src/tx.erl b/apps/tpnode/src/tx.erl
diff --git a/apps/tpnode/src/txpool.erl b/apps/tpnode/src/txpool.erl
