canal

-module(canal).
-include("canal_internal.hrl").

-export([
    auth/1,
    read/1,
    reauth/0,
    write/2
]).

-export([
    start_link/0
]).

-behaviour(gen_server).
-export([
    code_change/3,
    handle_call/3,
    handle_cast/2,
    handle_info/2,
    init/1,
    terminate/2
]).

-record(auth, {
    payload   = undefined          :: {string(), binary()} | undefined,
    timestamp = erlang:timestamp() :: erlang:timestamp(),
    token     = undefined          :: binary() | undefined,
    ttl       = undefined          :: non_neg_integer() | undefined
}).

-type auth() :: #auth{}.

-record(state, {
    auth            = undefined                   :: auth() | undefined,
    cache           = #{}                         :: #{binary() => binary()},
    request_timeout = undefined                   :: non_neg_integer(),
    requests        = #{}                         :: #{req_id() => req()},
    url             = undefined                   :: binary()
}).

-type state() :: #state{}.

-define(JSON, "application/json").


%% API

-spec auth(auth_method()) -> ok | {error, term()}.

auth(Creds = {approle, _Id, _SecretId}) ->
    application:set_env(?APP, credentials, Creds, [{persistent, true}]),
    gen_server:call(?MODULE, {auth, Creds});

auth(Creds = {ldap, _Username, _Password}) ->
    application:set_env(?APP, credentials, Creds, [{persistent, true}]),
    gen_server:call(?MODULE, {auth, Creds}).


-spec read(binary()) -> {ok, term()} | {error, term()}.

read(Key) ->
    gen_server:call(?MODULE, {read, Key}).


-spec reauth() -> ok.

reauth() ->
    gen_server:cast(?MODULE, reauth).


-spec start_link() -> {ok, pid()}.

start_link() ->
    gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).


-spec write(iodata(), term()) -> ok | {error, term()}.

write(Key, Val) ->
    Body = ?ENCODE(#{<<"value">> => Val}),
    gen_server:call(?MODULE, {write, Key, Body}).


%% gen_server callbacks

-spec code_change(term() | {down, term()}, state(), term()) -> {ok, state()}.

code_change(_Old, State, _Extra) ->
    {ok, State}.


-spec handle_call(_, _, state()) ->
    {reply, term(), state()} | {noreply, state()} | {stop, term(), state()}.

handle_call({auth, Creds}, _From, State) ->
    Payload = make_auth_request(Creds, State),
    case do_auth(Payload, State) of
        {ok, Data} ->
            {reply, ok, update_auth(State, Data, Payload)};
        Err = {error, _} ->
            canal_utils:error_msg("canal: auth failed with ~p", [Err]),
            {reply, Err, State}
    end;


handle_call({read, Key}, From, State) ->
    Headers = headers(State),
    Url = url(State, ["/v1/secret/", Key]),
    {Opts, HttpOpts} = opts(async, req_timeout(State)),

    Request = {Url, Headers},
    {ok, ReqId} = httpc:request(get, Request, HttpOpts, Opts),

    {noreply, add_request(ReqId, {From, read, Key}, State)};

handle_call({write, Key, Body}, From, State) ->
    Headers = headers(State),
    Url = url(State, ["/v1/secret/", Key]),
    {Opts, HttpOpts} = opts(async, req_timeout(State)),

    Request = {Url, Headers, ?JSON, Body},
    {ok, ReqId} = httpc:request(post, Request, HttpOpts, Opts),

    {noreply, add_request(ReqId, {From, write, Key}, State)}.


-spec handle_cast(_, state()) -> {noreply, state()}.

% payload = undefined disambiguates a #auth{} from a passed-in token from canal
% doing its own auth
handle_cast(reauth, State = #state{auth = #auth{payload = undefined}}) ->
    {noreply, State};

handle_cast(reauth, State = #state{auth = #auth{payload = Payload}}) ->
    State2 = case do_auth(Payload, State) of
        {ok, Data} ->
            update_auth(State, Data, Payload);
        Err = {error, _} ->
            canal_utils:error_msg("canal: auth failed with ~p", [Err]),
            State
    end,
    {noreply, State2};

handle_cast(reauth, State = #state{auth = undefined}) ->
    case ?GET_OPT(credentials) of
        undefined ->
            Msg = "canal: not authenticated, and no credentials stored",
            canal_utils:error_msg(Msg),
            {stop, {error, no_credentials}, State};
        Creds ->
            canal_utils:info_msg("canal: auth using credentials in env"),
            Payload = make_auth_request(Creds, State),
            case do_auth(Payload, State) of
                {ok, Data} ->
                    {noreply, update_auth(State, Data, Payload)};
                Err = {error, _} ->
                    canal_utils:error_msg("canal: auth failed with ~p", [Err]),
                    {stop, Err, State}
            end
    end;

handle_cast(_Req, State) ->
    {noreply, State}.


-spec handle_info(_, state()) -> {noreply, state()}.

handle_info({http, Response = {RequestId, _}}, State) ->
    case req_type(RequestId, State) of
        read ->
            handle_read_response(Response, State);
        write ->
            handle_write_response(Response, State);
        _ ->
            {noreply, del_request(RequestId, State)}
    end;


handle_info({token, Token}, State) ->
    case do_lookup(Token, State) of
        {ok, State2} ->
            {noreply, State2};
        Err = {error, _} ->
            Fmt = "canal: token lookup failed with ~p",
            Msg = io_lib:format(Fmt, [Err]),
            canal_utils:error_msg(Msg),
            {stop, {error, token_auth_failed}, State}
    end;

handle_info(_Req, State) ->
    {noreply, State}.


-spec init(_) -> {ok, state()}.

init(_) ->
    Url = ?GET_OPT(url),
    Timeout = ?GET_OPT(timeout),

    case ?GET_OPT(token) of
        undefined ->
            ok;
        Token ->
            self() ! {token, iolist_to_binary(Token)}
    end,

    case ?GET_OPT(credentials) of
        undefined ->
            ok;
        _Creds ->
            gen_server:cast(self(), reauth())
    end,

    State = #state{
        url = Url,
        request_timeout = Timeout
    },

    {ok, State}.


-spec terminate(term(), state()) -> ok.
terminate(_Reason, _State) ->
    ok.


%% private

-spec add_request(req_id(), req(), state()) -> state().

add_request(ReqId, Req, State = #state{requests = Requests}) ->
    State#state{requests = Requests#{ReqId => Req}}.


-spec del_request(req_id(), state()) -> state().

del_request(ReqId, State = #state{requests = Requests}) ->
    Requests2 = maps:remove(ReqId, Requests),
    State#state{requests = Requests2}.


do_auth({Url, Body}, State) ->
    do_auth2(Url, Body, req_timeout(State)).


do_auth2(Url, Body, Timeout) ->
    {Opts, HttpOpts} = opts(sync, Timeout),

    Request = {Url, [], ?JSON, Body},
    {ok, {{_NewVersion, StatusCode, _Status}, _RespHeaders, RespBody}} =
        httpc:request(post, Request, HttpOpts, Opts),

    case do_auth3(RespBody) of
        {ok, Auth} ->
            {ok, Auth};
        {error, Err} ->
            {error, {StatusCode, Err}}
    end.


do_auth3(Body) ->
    case ?DECODE(Body) of
        #{<<"auth">> := Auth} ->
            {ok, Auth};
        #{<<"errors">> := Err} ->
            {error, Err}
    end.


do_lookup(Token, State) ->
    Url = url(State, ["/v1/auth/token/lookup-self"]),
    Headers = [{"X-Vault-Token", binary_to_list(Token)}],
    {Opts, HttpOpts} = opts(sync, req_timeout(State)),

    Request = {Url, Headers},
    {ok, {{_NewVersion, StatusCode, _Status}, _RespHeaders, RespBody}} =
        httpc:request(get, Request, HttpOpts, Opts),

    case do_lookup2(Token, RespBody, State) of
        {ok, State2} ->
            {ok, State2};
        {error, Errors} ->
            {error, {StatusCode, Errors}}
    end.


do_lookup2(Token, Body, State) ->
    case ?DECODE(Body) of
        #{<<"data">> := Data} ->
            #{
                <<"ttl">> := Ttl
            } = Data,
            Auth2 = #auth{
                token = Token,
                ttl = Ttl
            },
            Msg = "canal: token found. Disabling reauthentication.",
            canal_utils:info_msg(Msg),
            {ok, State#state{auth = Auth2}};
        #{<<"errors">> := Errors} ->
            {error, Errors}
    end.


handle_read_response(
        {RequestId, {{_, StatusCode, _}, _, Reply}},
        State = #state{cache = Cache, requests = Requests}
    ) ->

    #{RequestId := {From, read, Key}} = Requests,
    Reply2 = ?DECODE(Reply),
    {Ret, State2} = case Reply2 of
        #{<<"data">> := Data} ->
            {{ok, Data}, State#state{cache = Cache#{Key => Data}}};
        #{<<"errors">> := Err} ->
            {{error, {StatusCode, Err}}, State}
    end,
    gen_server:reply(From, Ret),
    {noreply, del_request(RequestId, State2)};

handle_read_response(
        {RequestId, Err},
        State = #state{cache = Cache, requests = Requests}
    ) ->
    #{RequestId := {From, read, Key}} = Requests,
    Ret = case Cache of
        #{Key := Data} ->
            canal_utils:info_msg(
                "no response; sending key ~p from cache", [Key]),
            {ok, Data};
        _ -> Err
    end,
    gen_server:reply(From, Ret),
    {noreply, del_request(RequestId, State)}.


handle_write_response({RequestId, Response}, State) ->
    From = req_origin(RequestId, State),
    Reply = case Response of
        {{_, 204, _}, _, _} ->
            ok;
        {{_, StatusCode, _}, _, Body} ->
            case ?DECODE(Body) of
                #{<<"errors">> := Errors} ->
                    {error, {StatusCode, Errors}};
                _ ->
                    ok
            end
    end,

    gen_server:reply(From, Reply),
    {noreply, del_request(RequestId, State)}.


headers(State) ->
    case token(State) of
        {ok, Token} ->
            [{"X-Vault-Token", binary_to_list(Token)}];
        _ ->
            []
    end.


-spec make_auth_request(auth_method(), state()) -> {string(), binary()}.

make_auth_request({approle, RoleId, SecretId}, State) ->
    Map = #{<<"role_id">> => RoleId, <<"secret_id">> => SecretId},
    make_auth_request2("approle", Map, State);

make_auth_request({ldap, Username, Password}, State) ->
    Map = #{<<"password">> => Password},
    Url = url(State, ["/v1/auth/ldap/login/", Username]),
    Body = ?ENCODE(Map),
    {Url, Body}.


-spec make_auth_request2(iolist(), #{binary() => binary()}, state()) ->
    {string(), binary()}.

make_auth_request2(AuthMethod, Map, State) ->
    Url = url(State, ["/v1/auth/", AuthMethod, "/login"]),
    Body = ?ENCODE(Map),
    {Url, Body}.


-spec opts(sync | async, integer() | atom()) -> {list(), list()}.

opts(Sync, Timeout) ->
    Sync2 = case Sync of
        sync -> true;
        async -> false
    end,
    HttpOpts = [{timeout, Timeout}],
    Opts = [{body_format, binary}, {sync, Sync2}],
    {Opts, HttpOpts}.


-spec req_origin(req_id(), state()) -> {pid(), reference()} | undefined.

req_origin(RequestId, #state{requests = Requests}) ->
    case Requests of
        #{RequestId := {From, _Type, _Key}} ->
            From;
        _ ->
            undefined
    end.


-spec req_timeout(state()) -> timeout().

req_timeout(#state{request_timeout = Timeout}) ->
    Timeout.


-spec req_type(req_id(), state()) -> read | write | undefined.

req_type(RequestId, #state{requests = Requests}) ->
    case Requests of
        #{RequestId := {_, Type, _}} ->
            Type;
        #{RequestId := _} ->
            undefined;
        _ ->
            badkey
    end.


-spec token(state()) -> {ok, binary()} | undefined.

token(#state{auth = #auth{token = Token}}) when is_binary(Token) ->
    {ok, Token};

token(_) -> undefined.

-spec update_auth(state(), map(), {string(), binary()} | undefined) ->
    state().

update_auth(State, Data, Payload) ->
    #{
        <<"client_token">> := Token,
        <<"lease_duration">> := Ttl
    } = Data,
    NewAuth = #auth{
        payload = Payload,
        timestamp = erlang:timestamp(),
        token = Token,
        ttl = Ttl
    },
    ReauthTime = case floor(Ttl * 0.9) of
        N when N >= 0 -> N
    end,
    {Megas, Seconds, _Micros} = erlang:timestamp(),
    TimeAtReauth = {Megas, Seconds + ReauthTime, 0},
    DatetimeAtReauth = calendar:now_to_local_time(TimeAtReauth),
    canal_utils:info_msg("canal: will reauth at ~p", [DatetimeAtReauth]),
    {ok, _} = timer:apply_after(timer:seconds(ReauthTime), ?MODULE, reauth, []),
    State#state{auth = NewAuth}.


-spec url(state(), iolist()) -> string().

url(#state{url = BaseUrl}, IOList) ->
    binary_to_list(iolist_to_binary([BaseUrl | IOList])).