influxdb

-module(influxdb).
-export([
    query/2,
    query/3,
    query/4,
    write/2,
    write/3
]).
-export_type([
    config/0,
    time_unit/0,
    query/0,
    query_parameters/0,
    point/0
]).


-type config() :: influxdb_config:config().
-type time_unit() :: hour | minute | second | millisecond | microsecond | nanosecond.


-spec query(config(), query()) ->
      ok
    | {ok, [result()]}
    | {error, {not_found, string()}}
    | {error, {server_error, string()}}.
query(Config, Query) ->
    query(Config, Query, #{}, #{}).


-spec query(config(), query(), query_parameters()) ->
      ok
    | {ok, [result()]}
    | {error, {not_found, string()}}
    | {error, {server_error, string()}}.
query(Config, Query, Parameters) ->
    query(Config, Query, Parameters, #{}).


-spec query(config(), query(), query_parameters(), query_options()) ->
      ok
    | {ok, [result()]}
    | {error, {not_found, string()}}
    | {error, {server_error, string()}}.
-type query() :: iodata().
-type query_parameters() :: #{atom() => atom() | binary() | number()}.
-type query_options() :: #{
    timeout => timeout(),
    precision => time_unit(),
    retention_policy => iodata()
}.
-type result() :: influxdb_http:result().
query(#{host := Host, port := Port, username := Username, password := Password} = Config, Query, Parameters, Options) when is_map(Parameters), is_map(Options) ->
    Timeout = maps:get(timeout, Options, infinity),
    Url = influxdb_uri:encode(#{
        scheme => "http",
        host => Host,
        port => Port,
        path => "/query",
        query => url_query(Config, Options)
    }),
    Body = influxdb_uri:encode_query(#{
        q => Query,
        params => jsone:encode(Parameters)
    }),
    influxdb_http:post(Url, Username, Password, "application/x-www-form-urlencoded", Body, Timeout).


url_query(Config, Options) ->
    maps:fold(fun
        (precision, Value, Acc) -> maps:put("epoch", precision(Value), Acc);
        (retention_policy, Value, Acc) -> maps:put("rp", Value, Acc);
        (_Key, _Value, Acc) -> Acc
    end, default_url_query(Config), Options).


default_url_query(#{database := Database}) ->
    #{"db" => Database, "epoch" => precision(nanosecond)};
default_url_query(#{}) ->
    #{"epoch" => precision(nanosecond)}.


precision(hour) -> "h";
precision(minute) -> "m";
precision(second) -> "s";
precision(millisecond) -> "ms";
precision(microsecond) -> "u";
precision(nanosecond) -> "ns".


-spec write(config(), [point()]) ->
      ok
    | {error, {not_found, string()}}
    | {error, {server_error, string()}}.
write(Config, Measurements) ->
    write(Config, Measurements, #{}).


-spec write(config(), [point()], write_options()) ->
      ok
    | {error, {not_found, string()}}
    | {error, {server_error, string()}}.
-type point() :: influxdb_line_encoding:point().
-type write_options() :: #{
    timeout => timeout(),
    precision => time_unit(),
    retention_policy => string()
}.
write(#{host := Host, port := Port, username := Username, password := Password, database := Database}, Measurements, Options) ->
    Timeout = maps:get(timeout, Options, infinity),
    Url = influxdb_uri:encode(#{
        scheme => "http",
        host => Host,
        port => Port,
        path => "/write",
        query => maps:fold(fun
            (precision, Value, Acc) -> maps:put("precision", precision(Value), Acc);
            (retention_policy, Value, Acc) -> maps:put("rp", Value, Acc);
            (_Key, _Value, Acc) -> Acc
        end, #{"db" => Database}, Options)
    }),
    Body = influxdb_line_encoding:encode(Measurements),
    influxdb_http:post(Url, Username, Password, "application/octet-stream", Body, Timeout).