Initial

2024-03-27 14:45:50 +09:00
commit 3f80b11d63
15 changed files with 2891 additions and 0 deletions
@@ -0,0 +1,50 @@
+%%% @doc
+%%% The Hakuzaru Application Interface
+%%% 
+%%% This module provides the Erlang system application behavior only.
+%%% Please refer to the hz.erl module for the API.
+%%% @end
+
+-module(hakuzaru).
+-behavior(application).
+
+
+% OTP Application Interface
+-export([start/0, stop/0]).
+-export([start/2, stop/1]).
+
+
+
+-spec start() -> ok | {error, Reason :: term()}.
+%% @doc
+%% Public function for manually starting the Hakuzaru application.
+%%
+%% NOTE:
+%% To start it as a subordinate service within your own supervision tree rather than
+%% as a peer Erlang application within your node, add the hz_sup to your own
+%% supervision tree.
+
+start() ->
+    application:start(hakuzaru).
+
+
+-spec stop() -> ok | {error, Reason :: term()}.
+%% @doc
+%% Public function for manually stopping the Hakuzaru application.
+
+stop() ->
+    application:stop(hakuzaru).
+
+
+-spec start(normal, term()) -> {ok, pid()}.
+%% @private
+
+start(normal, _Args) ->
+    hz_sup:start_link().
+
+
+-spec stop(term()) -> ok.
+%% @private
+
+stop(_State) ->
+    ok.
@@ -0,0 +1,240 @@
+%%% @private
+
+-module(hz_fetcher).
+-vsn("0.4.1").
+-author("Craig Everett <ceverett@tsuriai.jp>").
+-copyright("Craig Everett <ceverett@tsuriai.jp>").
+-license("MIT").
+
+-export([connect/4, slowly_connect/4]).
+
+-include("$zx_include/zx_logger.hrl").
+
+
+connect(Node = {Host, Port}, Request, From, Timeout) ->
+    Timer = erlang:send_after(Timeout, self(), timeout),
+    Options = [{mode, binary}, {nodelay, true}, {active, once}],
+    case gen_tcp:connect(Host, Port, Options, 3000) of
+        {ok, Sock} -> do(Request, Sock, Node, From, Timer);
+        Error      -> gen_server:reply(From, Error)
+    end.
+
+do(Request, Sock, Node, From, Timer) ->
+    Formed = unicode:characters_to_list(form(Request, Node)),
+    case gen_tcp:send(Sock, Formed) of
+        ok    -> await(Sock, From, Timer);
+        Error -> gen_server:reply(From, Error)
+    end.
+
+await(Sock, From, Timer) ->
+    receive
+        {tcp, Sock, Bin} ->
+            parse(Bin, Sock, From, Timer);
+        {tcp_closed, Sock} ->
+            ok = erlang:cancel_timer(Timer, [{async, true}]),
+            gen_server:reply(From, {error, enotconn});
+        timeout ->
+            gen_server:reply(From, {error, timeout})
+        after 120000 ->
+            gen_server:reply(From, {error, timeout})
+    end.
+
+
+form({get, Path}, Node) ->
+    ["GET ", Path, " HTTP/1.1\r\n",
+     "Host: ", host_string(Node), "\r\n",
+     "User-Agent: Kanou/0.1.0\r\n",
+     "Accept: */*\r\n\r\n"];
+form({post, Path, Payload}, Node) ->
+    ByteSize = integer_to_list(byte_size(Payload)),
+    ["POST ", Path, " HTTP/1.1\r\n",
+     "Host: ", host_string(Node), "\r\n",
+     "Content-Type: application/json\r\n", 
+     "Content-Length: ", ByteSize, "\r\n",
+     "User-Agent: Kanou/0.1.0\r\n",
+     "Accept: */*\r\n\r\n",
+     Payload].
+
+
+host_string({Address, Port}) when is_list(Address) ->
+    PortS = integer_to_list(Port),
+    [Address, ":", PortS];
+host_string({Address, Port}) when is_atom(Address) ->
+    AddressS = atom_to_list(Address),
+    PortS = integer_to_list(Port),
+    [AddressS, ":", PortS];
+host_string({Address, Port}) ->
+    AddressS = inet:ntoa(Address),
+    PortS = integer_to_list(Port),
+    [AddressS, ":", PortS].
+
+
+parse(Received, Sock, From, Timer) ->
+    case Received of
+        <<"HTTP/1.1 200 OK\r\n", Tail/binary>> ->
+            parse2(200, Tail, Sock, From, Timer);
+        <<"HTTP/1.1 400 Bad Request\r\n", Tail/binary>> ->
+            parse2(400, Tail, Sock, From, Timer);
+        <<"HTTP/1.1 404 Not Found\r\n", Tail/binary>> ->
+            parse2(404, Tail, Sock, From, Timer);
+        <<"HTTP/1.1 500 Internal Server Error\r\n", Tail/binary>> ->
+            parse2(500, Tail, Sock, From, Timer);
+        _ ->
+            ok = zx_net:disconnect(Sock),
+            ok = erlang:cancel_timer(Timer, [{async, true}]),
+            gen_server:reply(From, {error, {received, Received}})
+    end.
+
+parse2(Code, Received, Sock, From, Timer) ->
+    case read_headers(Sock, Received) of
+        {ok, Headers, Rest} -> consume(Code, Rest, Headers, Sock, From, Timer);
+        Error               -> gen_server:reply(From, Error)
+    end.
+
+
+consume(Code, Rest, Headers, Sock, From, Timer) ->
+    case maps:find(<<"content-length">>, Headers) of
+        error ->
+            ok = erlang:cancel_timer(Timer, [{async, true}]),
+            gen_server:reply(From, {error, {headers, Headers}});
+        {ok, <<"0">>} ->
+            ok = erlang:cancel_timer(Timer, [{async, true}]),
+            Result = case Code =:= 200 of true -> ok; false -> {error, Code} end,
+            gen_server:reply(From, Result);
+        {ok, Size} ->
+            try
+                Length = binary_to_integer(Size),
+                consume2(Length, Rest, Sock, From, Timer)
+            catch
+                error:badarg ->
+                    ok = erlang:cancel_timer(Timer, [{async, true}]),
+                    gen_server:reply(From, {error, {headers, Headers}})
+            end
+    end.
+
+consume2(Length, Received, Sock, From, Timer) ->
+    Size = byte_size(Received),
+    if
+        Size == Length ->
+            ok = erlang:cancel_timer(Timer, [{async, true}]),
+            ok = zx_net:disconnect(Sock),
+            Result = zj:decode(Received),
+            gen_server:reply(From, Result);
+        Size  < Length ->
+            consume3(Length, Received, Sock, From, Timer);
+        Size >  Length ->
+            ok = erlang:cancel_timer(Timer, [{async, true}]),
+            gen_server:reply(From, {error, bad_length})
+    end.
+
+consume3(Length, Received, Sock, From, Timer) ->
+    ok = inet:setopts(Sock, [{active, once}]),
+    receive
+        {tcp, Sock, Bin} ->
+            consume2(Length, <<Received/binary, Bin/binary>>, Sock, From, Timer);
+        timeout ->
+            gen_server:reply(From, {error, {timeout, Received}})
+    end.
+
+
+read_headers(Socket, <<"\r">>) ->
+    ok = inet:setopts(Socket, [{active, once}]),
+    receive
+        {tcp, Socket, Bin} -> read_headers(Socket, <<"\r", Bin/binary>>);
+        timeout            -> {error, timeout}
+        after 120000       -> {error, timeout}
+    end;
+read_headers(_, <<"\r\n", Received/binary>>) ->
+    log(info, "~p Headers died at: ~p", [?LINE, Received]),
+    {error, headers};
+read_headers(Socket, Received) ->
+    read_hkey(Socket, Received, <<>>, #{}).
+
+read_hkey(Socket, <<Char, Rest/binary>>, Acc, Headers)
+        when $A =< Char, Char =< $Z ->
+    read_hkey(Socket, Rest, <<Acc/binary, (Char + 32)>>, Headers);
+read_hkey(Socket, <<Char, Rest/binary>>, Acc, Headers)
+        when 32 =< Char, Char =< 57;
+             59 =< Char, Char =< 126 ->
+    read_hkey(Socket, Rest, <<Acc/binary, Char>>, Headers);
+read_hkey(Socket, <<":", Rest/binary>>, Key, Headers) ->
+    skip_hblanks(Socket, Rest, Key, Headers);
+read_hkey(_, <<"\r\n", Rest/binary>>, <<>>, Headers) ->
+    {ok, Headers, Rest};
+read_hkey(Socket, <<>>, Acc, Headers) ->
+    ok = inet:setopts(Socket, [{active, once}]),
+    receive
+        {tcp, Socket, Bin} -> read_hkey(Socket, Bin, Acc, Headers);
+        timeout            -> {error, timeout}
+        after 120000       -> {error, timeout}
+    end;
+read_hkey(_, Received, _, _) ->
+    log(info, "~p Headers died at: ~p", [?LINE, Received]),
+    {error, headers}.
+
+skip_hblanks(Socket, <<" ", Rest/binary>>, Key, Headers) ->
+    skip_hblanks(Socket, Rest, Key, Headers);
+skip_hblanks(Socket, <<>>, Key, Headers) ->
+    ok = inet:setopts(Socket, [{active, once}]),
+    receive
+        {tcp, Socket, Bin} -> skip_hblanks(Socket, Bin, Key, Headers);
+        timeout            -> {error, timeout}
+        after 120000       -> {error, timeout}
+    end;
+skip_hblanks(_, Received = <<"\r", _/binary>>, _, _) ->
+    log(info, "~p Headers died at: ~p", [?LINE, Received]),
+    {error, headers};
+skip_hblanks(_, Received = <<"\n", _/binary>>, _, _) ->
+    log(info, "~p Headers died at: ~p", [?LINE, Received]),
+    {error, headers};
+skip_hblanks(Socket, Rest, Key, Headers) ->
+    read_hval(Socket, Rest, <<>>, Key, Headers).
+
+read_hval(_, Received = <<"\r\n", _/binary>>, <<>>, _, _) ->
+    log(info, "~p Headers died at: ~p", [?LINE, Received]),
+    {error, headers};
+read_hval(Socket, <<"\r\n", Rest/binary>>, Val, Key, Headers) ->
+    read_hkey(Socket, Rest, <<>>, maps:put(Key, Val, Headers));
+read_hval(Socket, <<Char, Rest/binary>>, Acc, Key, Headers)
+        when 32 =< Char, Char =< 126 ->
+    read_hval(Socket, Rest, <<Acc/binary, Char>>, Key, Headers);
+read_hval(Socket, <<>>, Val, Key, Headers) ->
+    ok = inet:setopts(Socket, [{active, once}]),
+    receive
+        {tcp, Socket, Bin} -> read_hval(Socket, Bin, Val, Key, Headers);
+        timeout            -> {error, timeout}
+        after 120000       -> {error, timeout}
+    end;
+read_hval(_, Received, _, _, _) ->
+    log(info, "~p Headers died at: ~p", [?LINE, Received]),
+    {error, headers}.
+
+
+slowly_connect(Node, {get, Path}, From, Timeout) ->
+    HttpOptions = [{connect_timeout, 3000}, {timeout, Timeout}],
+    URL = lists:flatten(url(Node, Path)),
+    Request = {URL, []},
+    Result =
+        case httpc:request(get, Request, HttpOptions, []) of
+            {ok, {{_, 200, _}, _, JSON}} -> zj:decode(JSON);
+            {ok, {{_, BAD, _}, _, _}}    -> {error, BAD};
+            BAD                          -> {error, BAD}
+        end,
+    gen_server:reply(From, Result);
+slowly_connect(Node, {post, Path, Payload}, From, Timeout) ->
+    HttpOptions = [{connect_timeout, 3000}, {timeout, Timeout}],
+    URL = lists:flatten(url(Node, Path)),
+    Request = {URL, [], "application/json", Payload},
+    Result =
+        case httpc:request(post, Request, HttpOptions, []) of
+            {ok, {{_, 200, _}, _, JSON}} -> zj:decode(JSON);
+            {ok, {{_, BAD, _}, _, _}}    -> {error, BAD};
+            BAD                          -> {error, BAD}
+        end,
+    gen_server:reply(From, Result).
+
+
+url({Node, Port}, Path) when is_list(Node) ->
+    ["https://", Node, ":", integer_to_list(Port), Path];
+url({Node, Port}, Path) when is_tuple(Node) ->
+    ["https://", inet:ntoa(Node), ":", integer_to_list(Port), Path].
@@ -0,0 +1,289 @@
+%%% @private
+%%% Hakuzaru Request Manager for Erlang
+%%%
+%%% This process is responsible for remembering the configured nodes and dispatching
+%%% requests to them. Request dispatch is made in a round-robin fashion with forwarded
+%%% gen_server return `From' values passed to the request worker instead of being
+%%% responded to directly by the manager itself (despite requests being generated as
+%%% gen_server:call/3s.
+%%% @end
+
+-module(hz_man).
+-vsn("0.4.1").
+-behavior(gen_server).
+-author("Craig Everett <ceverett@tsuriai.jp>").
+-copyright("Craig Everett <ceverett@tsuriai.jp>").
+-license("MIT").
+
+%% Admin functions
+-export([network_id/0,  network_id/1,
+         tls/0,         tls/1,
+         chain_nodes/0, chain_nodes/1,
+         timeout/0,     timeout/1]).
+
+%% The whole point of this module:
+-export([request/1, request/2]).
+
+%% gen_server goo
+-export([start_link/0]).
+-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
+         code_change/3, terminate/2]).
+
+%% TODO: Make logging more flexible
+-include("$zx_include/zx_logger.hrl").
+
+
+%%% Type and Record Definitions
+
+-record(fetcher,
+        {pid  = none :: none | pid(),
+         mon  = none :: none | reference(),
+         time = none :: none | integer(), % nanosecond timestamp
+         node = none :: none | hz:chain_node(),
+         from = none :: none | gen_server:from(),
+         req  = none :: none | binary()}).
+
+-record(s,
+        {network_id  = "gm_mainnet" :: string(),
+         tls         = false        :: boolean(),
+         chain_nodes = {[], []}     :: {[hz:chain_node()], [hz:chain_node()]},
+         sticky      = none         :: none | hz:chain_node(),
+         fetchers    = []           :: [#fetcher{}],
+         timeout     = 5000         :: pos_integer()}).
+
+
+-type state()   :: #s{}.
+
+
+
+%%% Service Interface
+
+-spec network_id() -> Name
+    when Name :: hz:network_id().
+
+network_id() ->
+    gen_server:call(?MODULE, network_id).
+
+
+-spec network_id(Name) -> ok
+    when Name :: hz:network_id().
+
+network_id(Name) ->
+    gen_server:cast(?MODULE, {network_id, Name}).
+
+
+-spec tls() -> boolean().
+
+tls() ->
+    gen_server:call(?MODULE, tls).
+
+
+-spec tls(boolean()) -> ok.
+
+tls(Boolean) ->
+    gen_server:cast(?MODULE, {tls, Boolean}).
+
+
+-spec chain_nodes() -> Used
+    when Used :: [hz:chain_node()].
+
+chain_nodes() ->
+    gen_server:call(?MODULE, chain_nodes).
+
+
+-spec chain_nodes(ToUse) -> ok
+    when ToUse :: [hz:chain_nodes()].
+
+chain_nodes(ToUse) ->
+    gen_server:cast(?MODULE, {chain_nodes, ToUse}).
+
+
+-spec timeout() -> Value
+    when Value :: pos_integer().
+
+timeout() ->
+    gen_server:call(?MODULE, timeout).
+
+
+-spec timeout(Value) -> ok
+    when Value :: pos_integer().
+
+timeout(Value) when 0 < Value, Value =< 120000 ->
+    gen_server:cast(?MODULE, {timeout, Value}).
+
+
+-spec request(Path) -> {ok, Value} | {error, Reason}
+    when Path   :: unicode:charlist(),
+         Value  :: map(),
+         Reason :: hz:chain_error().
+
+request(Path) ->
+    gen_server:call(?MODULE, {request, {get, Path}}, infinity).
+
+
+-spec request(Path, Data) -> {ok, Value} | {error, Reason}
+    when Path   :: unicode:charlist(),
+         Data   :: unicode:charlist(),
+         Value  :: map(),
+         Reason :: hz:chain_error().
+
+request(Path, Data) ->
+    gen_server:call(?MODULE, {request, {post, Path, Data}}, infinity).
+
+
+
+%%% Startup Functions
+
+
+-spec start_link() -> Result
+    when Result :: {ok, pid()}
+                 | {error, Reason :: term()}.
+%% @private
+%% This should only ever be called by v_clients (the service-level supervisor).
+
+start_link() ->
+    gen_server:start_link({local, ?MODULE}, ?MODULE, none, []).
+
+
+-spec init(none) -> {ok, state()}.
+%% @private
+%% Called by the supervisor process to give the process a chance to perform any
+%% preparatory work necessary for proper function.
+
+init(none) ->
+    ok = io:format("hz_man starting.~n"),
+    State = #s{},
+    {ok, State}.
+
+
+
+%%% gen_server Message Handling Callbacks
+
+
+handle_call({request, Request}, From, State) ->
+    NewState = do_request(Request, From, State),
+    {noreply, NewState};
+handle_call(network_id, _, State = #s{network_id = Name}) ->
+    {reply, Name, State};
+handle_call(tls, _, State = #s{tls = TLS}) ->
+    {reply, TLS, State};
+handle_call(chain_nodes, _, State = #s{chain_nodes = {Wait, Used}}) ->
+    Nodes = lists:append(Wait, Used),
+    {reply, Nodes, State};
+handle_call(timeout, _, State = #s{timeout = Value}) ->
+    {reply, Value, State};
+handle_call(Unexpected, From, State) ->
+    ok = log(warning, "Unexpected call from ~tp: ~tp~n", [From, Unexpected]),
+    {noreply, State}.
+
+
+handle_cast({network_id, Name}, State) ->
+    {noreply, State#s{network_id = Name}};
+handle_cast({tls, Boolean}, State) ->
+    NewState = do_tls(Boolean, State),
+    {noreply, NewState};
+handle_cast({chain_nodes, []}, State) ->
+    {noreply, State#s{chain_nodes = none}};
+handle_cast({chain_nodes, ToUse}, State) ->
+    {noreply, State#s{chain_nodes = {ToUse, []}}};
+handle_cast({timeout, Value}, State) ->
+    {noreply, State#s{timeout = Value}};
+handle_cast(Unexpected, State) ->
+    ok = log(warning, "Unexpected cast: ~tp~n", [Unexpected]),
+    {noreply, State}.
+
+
+handle_info({'DOWN', Mon, process, PID, Info}, State) ->
+    NewState = handle_down(PID, Mon, Info, State),
+    {noreply, NewState};
+handle_info(Unexpected, State) ->
+    ok = log("Unexpected info: ~tp~n", [Unexpected]),
+    {noreply, State}.
+
+
+handle_down(_, Mon, normal, State = #s{fetchers = Fetchers}) ->
+    NewFetchers = lists:keydelete(Mon, #fetcher.mon, Fetchers),
+    State#s{fetchers = NewFetchers};
+handle_down(PID, Mon, Info, State = #s{fetchers = Fetchers}) ->
+    case lists:keytake(Mon, #fetcher.mon, Fetchers) of
+        {value, #fetcher{time = Time, node = Node, from = From, req = R}, Remaining} ->
+            TS = calendar:system_time_to_rfc3339(Time, [{unit, nanosecond}]),
+            Format =
+                "ERROR ~ts: Fetcher process ~130tp exited while making request to ~130tp~n"
+                "Exit reason:~n"
+                "~tp~n"
+                "Request contents:~n"
+                "~tp~n~n",
+            Formatted = io_lib:format(Format, [TS, PID, Node, Info, R]),
+            Message = unicode:characters_to_list(Formatted),
+            ok = gen_server:reply(From, {error, Message}),
+            State#s{fetchers = Remaining};
+        false ->
+            Unexpected = {'DOWN', Mon, process, PID, Info},
+            ok = log(warning, "Unexpected info: ~w", [Unexpected]),
+            State
+    end.
+
+
+
+
+%%% OTP Service Functions
+
+code_change(_, State, _) ->
+    {ok, State}.
+
+
+terminate(_, _) ->
+    ok.
+
+
+
+%%% Doer Functions
+
+do_tls(true, State) ->
+    ok = ssl:start(),
+    State#s{tls = true};
+do_tls(false, State) ->
+    State#s{tls = false};
+do_tls(_, State) ->
+    State.
+
+
+do_request(_, From, State = #s{chain_nodes = {[], []}}) ->
+    ok = gen_server:reply(From, {error, no_nodes}),
+    State;
+do_request(Request,
+           From,
+           State = #s{tls         = false,
+                      fetchers    = Fetchers,
+                      chain_nodes = {[Node | Rest], Used},
+                      timeout     = Timeout}) ->
+    Now = erlang:system_time(nanosecond),
+    Fetcher = fun() -> hz_fetcher:connect(Node, Request, From, Timeout) end,
+    {PID, Mon} = spawn_monitor(Fetcher),
+    New = #fetcher{pid  = PID,
+                   mon  = Mon,
+                   time = Now,
+                   node = Node,
+                   from = From,
+                   req  = Request},
+    State#s{fetchers = [New | Fetchers], chain_nodes = {Rest, [Node | Used]}};
+do_request(Request,
+           From,
+           State = #s{tls         = true,
+                      fetchers    = Fetchers,
+                      chain_nodes = {[Node | Rest], Used},
+                      timeout     = Timeout}) ->
+    Now = erlang:system_time(nanosecond),
+    Fetcher = fun() -> hz_fetcher:slowly_connect(Node, Request, From, Timeout) end,
+    {PID, Mon} = spawn_monitor(Fetcher),
+    New = #fetcher{pid  = PID,
+                   mon  = Mon,
+                   time = Now,
+                   node = Node,
+                   from = From,
+                   req  = Request},
+    State#s{fetchers = [New | Fetchers], chain_nodes = {Rest, [Node | Used]}};
+do_request(Request, From, State = #s{chain_nodes = {[], Used}}) ->
+    Fresh = lists:reverse(Used),
+    do_request(Request, From, State#s{chain_nodes = {Fresh, []}}).
@@ -0,0 +1,43 @@
+%%% @private
+%%% Hakuzaru Erlang Gajumaru application supervisor
+%%%
+%%% The very top level supervisor in the system. It only has one service branch: the
+%%% "hz_man" (Hakuzaru Manager).
+%%%
+%%% See: http://erlang.org/doc/design_principles/applications.html
+%%% See: http://zxq9.com/archives/1311
+%%% @end
+
+-module(hz_sup).
+-vsn("0.4.1").
+-behaviour(supervisor).
+-author("Craig Everett <zxq9@zxq9.com>").
+-copyright("Craig Everett <zxq9@zxq9.com>").
+-license("GPL-3.0-or-later").
+
+-export([start_link/0]).
+-export([init/1]).
+
+
+-spec start_link() -> {ok, pid()}.
+%% @private
+%% This supervisor's own start function.
+
+start_link() ->
+  supervisor:start_link({local, ?MODULE}, ?MODULE, []).
+
+
+-spec init([]) -> {ok, {supervisor:sup_flags(), [supervisor:child_spec()]}}.
+%% @private
+%% The OTP init/1 function.
+
+init([]) ->
+    RestartStrategy = {one_for_one, 0, 60},
+    Manager   = {hz_man,
+                 {hz_man, start_link, []},
+                 permanent,
+                 5000,
+                 worker,
+                 [hz_man]},
+    Children  = [Manager],
+    {ok, {RestartStrategy, Children}}.