%%%-------------------------------------------------------------------
%%% @doc
%%% Short-lived Docker API client process.
%%%
%%% Each request owns one process and one gun Unix socket connection.
%%% The process exits after the Docker API response finishes.
%%% @end
%%%-------------------------------------------------------------------
-module(docker_client).

%% API
-export([request/4, start_stream/5]).

-define(DOCKER_SOCKET, "/var/run/docker.sock").
-define(RESPONSE_TIMEOUT, 5000).
-define(BODY_TIMEOUT, 10000).
-define(STREAM_BODY_TIMEOUT, 30000).
-define(CLIENT_TIMEOUT, 60000).

%%%===================================================================
%%% API
%%%===================================================================

-spec request(Method :: string(), Path :: string(), Body :: binary(), Headers :: list()) ->
    {ok, StatusCode :: integer(), RespHeaders :: proplists:proplist(), RespBody :: binary()} | {error, any()}.
request(Method, Path, Body, Headers) when is_list(Method), is_list(Path), is_binary(Body), is_list(Headers) ->
    Owner = self(),
    Ref = make_ref(),
    {Pid, MRef} = spawn_monitor(fun() -> request_worker(Owner, Ref, Method, Path, Body, Headers) end),
    receive
        {docker_client, Ref, {result, Result}} ->
            erlang:demonitor(MRef, [flush]),
            Result;
        {'DOWN', MRef, process, Pid, Reason} ->
            {error, {client_down, Reason}}
    after ?CLIENT_TIMEOUT ->
        exit(Pid, shutdown),
        erlang:demonitor(MRef, [flush]),
        {error, timeout}
    end.

-spec start_stream(Owner :: pid(), Method :: string(), Path :: string(), Body :: binary(), Headers :: list()) ->
    {ok, Ref :: reference(), Pid :: pid(), MRef :: reference()}.
start_stream(Owner, Method, Path, Body, Headers)
  when is_pid(Owner), is_list(Method), is_list(Path), is_binary(Body), is_list(Headers) ->
    Ref = make_ref(),
    {Pid, MRef} = spawn_monitor(fun() -> stream_worker(Owner, Ref, Method, Path, Body, Headers) end),
    {ok, Ref, Pid, MRef}.

%%%===================================================================
%%% Worker functions
%%%===================================================================

-spec request_worker(pid(), reference(), string(), string(), binary(), list()) -> ok.
request_worker(Owner, Ref, Method, Path, Body, Headers) ->
    Result = do_request(Method, Path, Body, Headers),
    send_owner(Owner, Ref, {result, Result}).

-spec stream_worker(pid(), reference(), string(), string(), binary(), list()) -> ok.
stream_worker(Owner, Ref, Method, Path, Body, Headers) ->
    OwnerRef = erlang:monitor(process, Owner),
    _Result = do_stream(Owner, OwnerRef, Ref, Method, Path, Body, Headers),
    erlang:demonitor(OwnerRef, [flush]),
    ok.

%%%===================================================================
%%% Internal functions
%%%===================================================================

-spec do_request(string(), string(), binary(), list()) ->
    {ok, integer(), proplists:proplist(), binary()} | {error, any()}.
do_request(Method, Path, Body, Headers) ->
    case open_connection() of
        {ok, ConnPid} ->
            try
                StreamRef = gun:request(ConnPid, Method, Path, Headers, Body),
                receive_response(ConnPid, StreamRef)
            after
                close_connection(ConnPid)
            end;
        {error, Reason} ->
            {error, Reason}
    end.

-spec do_stream(pid(), reference(), reference(), string(), string(), binary(), list()) ->
    ok | {error, binary()}.
do_stream(Owner, OwnerRef, Ref, Method, Path, Body, Headers) ->
    case open_connection() of
        {ok, ConnPid} ->
            try
                StreamRef = gun:request(ConnPid, Method, Path, Headers, Body),
                receive_stream_response(Owner, OwnerRef, Ref, ConnPid, StreamRef)
            after
                close_connection(ConnPid)
            end;
        {error, Reason0} ->
            Reason = format_error(Reason0),
            send_owner(Owner, Ref, {error, Reason}),
            {error, Reason}
    end.

-spec receive_response(pid(), reference()) ->
    {ok, integer(), proplists:proplist(), binary()} | {error, any()}.
receive_response(ConnPid, StreamRef) ->
    receive
        {gun_response, ConnPid, StreamRef, nofin, Status, Headers} ->
            receive_body(ConnPid, StreamRef, Status, Headers, <<>>);
        {gun_response, ConnPid, StreamRef, fin, Status, Headers} ->
            {ok, Status, Headers, <<>>};
        {gun_error, ConnPid, StreamRef, Reason} ->
            {error, {http_error, Reason}};
        {gun_error, ConnPid, Reason} ->
            {error, {http_error, Reason}};
        {gun_down, ConnPid, _, Reason, _} ->
            {error, {http_closed, Reason}}
    after ?RESPONSE_TIMEOUT ->
        {error, timeout}
    end.

-spec receive_body(pid(), reference(), integer(), proplists:proplist(), iodata()) ->
    {ok, integer(), proplists:proplist(), binary()} | {error, any()}.
receive_body(ConnPid, StreamRef, Status, Headers, Acc) ->
    receive
        {gun_data, ConnPid, StreamRef, fin, Data} ->
            Body = iolist_to_binary([Acc, Data]),
            {ok, Status, Headers, Body};
        {gun_data, ConnPid, StreamRef, nofin, Data} ->
            receive_body(ConnPid, StreamRef, Status, Headers, [Acc, Data]);
        {gun_error, ConnPid, StreamRef, Reason} ->
            {error, {http_error, Reason}};
        {gun_error, ConnPid, Reason} ->
            {error, {http_error, Reason}};
        {gun_down, ConnPid, _, Reason, _} ->
            {error, {http_closed, Reason}}
    after ?BODY_TIMEOUT ->
        {error, timeout}
    end.

-spec receive_stream_response(pid(), reference(), reference(), pid(), reference()) -> ok | {error, binary()}.
receive_stream_response(Owner, OwnerRef, Ref, ConnPid, StreamRef) ->
    receive
        {gun_response, ConnPid, StreamRef, nofin, Status, Headers} when Status >= 200, Status < 300 ->
            send_owner(Owner, Ref, {response, Status, Headers}),
            receive_stream_body(Owner, OwnerRef, Ref, ConnPid, StreamRef);
        {gun_response, ConnPid, StreamRef, fin, Status, Headers} when Status >= 200, Status < 300 ->
            send_owner(Owner, Ref, {response, Status, Headers}),
            send_owner(Owner, Ref, done),
            ok;
        {gun_response, ConnPid, StreamRef, nofin, Status, _Headers} ->
            receive_stream_error_body(Owner, OwnerRef, Ref, ConnPid, StreamRef, Status, <<>>);
        {gun_response, ConnPid, StreamRef, fin, Status, _Headers} ->
            Reason = http_status_error(Status, <<>>),
            send_owner(Owner, Ref, {error, Reason}),
            {error, Reason};
        {gun_error, ConnPid, StreamRef, Reason0} ->
            Reason = format_error(Reason0),
            send_owner(Owner, Ref, {error, Reason}),
            {error, Reason};
        {gun_error, ConnPid, Reason0} ->
            Reason = format_error(Reason0),
            send_owner(Owner, Ref, {error, Reason}),
            {error, Reason};
        {gun_down, ConnPid, _, Reason0, _} ->
            Reason = format_error(Reason0),
            send_owner(Owner, Ref, {error, Reason}),
            {error, Reason};
        {'DOWN', OwnerRef, process, _Pid, Reason0} ->
            {error, format_error({owner_down, Reason0})}
    after ?RESPONSE_TIMEOUT ->
        send_owner(Owner, Ref, {error, <<"处理超时"/utf8>>}),
        {error, <<"timeout">>}
    end.

-spec receive_stream_body(pid(), reference(), reference(), pid(), reference()) -> ok | {error, binary()}.
receive_stream_body(Owner, OwnerRef, Ref, ConnPid, StreamRef) ->
    receive
        {gun_data, ConnPid, StreamRef, fin, Data} ->
            maybe_send_stream_data(Owner, Ref, Data),
            send_owner(Owner, Ref, done),
            ok;
        {gun_data, ConnPid, StreamRef, nofin, Data} ->
            maybe_send_stream_data(Owner, Ref, Data),
            receive_stream_body(Owner, OwnerRef, Ref, ConnPid, StreamRef);
        {gun_error, ConnPid, StreamRef, Reason0} ->
            Reason = format_error(Reason0),
            send_owner(Owner, Ref, {error, Reason}),
            {error, Reason};
        {gun_error, ConnPid, Reason0} ->
            Reason = format_error(Reason0),
            send_owner(Owner, Ref, {error, Reason}),
            {error, Reason};
        {gun_down, ConnPid, _, Reason0, _} ->
            Reason = format_error(Reason0),
            send_owner(Owner, Ref, {error, Reason}),
            {error, Reason};
        {'DOWN', OwnerRef, process, _Pid, Reason0} ->
            {error, format_error({owner_down, Reason0})}
    after ?STREAM_BODY_TIMEOUT ->
        send_owner(Owner, Ref, {error, <<"timeout">>}),
        {error, <<"timeout">>}
    end.

-spec receive_stream_error_body(pid(), reference(), reference(), pid(), reference(), integer(), iodata()) ->
    {error, binary()}.
receive_stream_error_body(Owner, OwnerRef, Ref, ConnPid, StreamRef, Status, Acc) ->
    receive
        {gun_data, ConnPid, StreamRef, fin, Data} ->
            Reason = http_status_error(Status, iolist_to_binary([Acc, Data])),
            send_owner(Owner, Ref, {error, Reason}),
            {error, Reason};
        {gun_data, ConnPid, StreamRef, nofin, Data} ->
            receive_stream_error_body(Owner, OwnerRef, Ref, ConnPid, StreamRef, Status, [Acc, Data]);
        {gun_error, ConnPid, StreamRef, Reason0} ->
            Reason = format_error(Reason0),
            send_owner(Owner, Ref, {error, Reason}),
            {error, Reason};
        {gun_error, ConnPid, Reason0} ->
            Reason = format_error(Reason0),
            send_owner(Owner, Ref, {error, Reason}),
            {error, Reason};
        {gun_down, ConnPid, _, Reason0, _} ->
            Reason = format_error(Reason0),
            send_owner(Owner, Ref, {error, Reason}),
            {error, Reason};
        {'DOWN', OwnerRef, process, _Pid, Reason0} ->
            {error, format_error({owner_down, Reason0})}
    after ?BODY_TIMEOUT ->
        Reason = http_status_error(Status, iolist_to_binary(Acc)),
        send_owner(Owner, Ref, {error, Reason}),
        {error, Reason}
    end.

-spec open_connection() -> {ok, pid()} | {error, any()}.
open_connection() ->
    case gun:open_unix(?DOCKER_SOCKET, #{}) of
        {ok, ConnPid} ->
            case gun:await_up(ConnPid) of
                {ok, _} ->
                    {ok, ConnPid};
                {error, Reason} ->
                    close_connection(ConnPid),
                    {error, Reason}
            end;
        {error, Reason} ->
            {error, Reason}
    end.

-spec close_connection(pid()) -> ok.
close_connection(ConnPid) ->
    catch gun:close(ConnPid),
    ok.

-spec send_owner(pid(), reference(), term()) -> ok.
send_owner(Owner, Ref, Event) ->
    Owner ! {docker_client, Ref, Event},
    ok.

-spec maybe_send_stream_data(pid(), reference(), iodata()) -> ok.
maybe_send_stream_data(Owner, Ref, Data) ->
    case iolist_to_binary(Data) of
        <<>> ->
            ok;
        DataBin ->
            send_owner(Owner, Ref, {data, DataBin})
    end.

-spec http_status_error(integer(), binary()) -> binary().
http_status_error(Status, <<>>) ->
    iolist_to_binary(io_lib:format("docker http status ~B", [Status]));
http_status_error(Status, Body) when is_binary(Body) ->
    StatusBin = integer_to_binary(Status),
    <<"docker http status ", StatusBin/binary, ": ", Body/binary>>.

-spec format_error(term()) -> binary().
format_error(Reason) when is_binary(Reason) ->
    Reason;
format_error(Reason) ->
    iolist_to_binary(io_lib:format("~p", [Reason])).