ex_flux

defmodule ExFlux.Conn.HTTP do
  @moduledoc """
  DSL around HTTPoison for interacting with the endpoints exposed by InfluxDB
  """

  @epoch_map %{
    hour: "h",
    minute: "m",
    second: "s",
    millisecond: "ms",
    microsecond: "us",
    nanosecond: "ns",
    hours: "h",
    minutes: "m",
    seconds: "s",
    milliseconds: "ms",
    microseconds: "us",
    nanoseconds: "ns"
  }

  @doc """
  check the status of the influxdb instance
  """
  def ping(opts) do
    headers = [] |> with_auth(opts)

    opts
    |> base_uri()
    |> endpoint("ping")
    |> HTTPoison.get!(headers, opts[:http_opts] || [])
    |> ping_response()
  rescue
    e in [HTTPoison.Error] -> {:error, e.reason}
  end

  @doc """
  queries that do not mutate the data (i.e. select and show). "select ... into"
  queries do mutations, so `post_query/2` should be used in this case
  """
  def query(query_string, opts) do
    headers = [] |> with_auth(opts)

    uri =
      opts
      |> base_uri()
      |> endpoint("query")

    params =
      query_string
      |> with_q()
      |> with_database(opts)
      |> with_epoch(opts)

    request_opts = Keyword.put(opts[:http_opts] || [], :params, params)

    uri
    |> HTTPoison.get!(headers, request_opts)
    |> query_response(opts)
  rescue
    e in [HTTPoison.Error] -> {:error, e.reason || e.message}
  end

  @doc """
  queries that mutate the database or insert data (i.e. select...into, alter,
  create, etc)
  """
  def post_query(query_string, opts) do
    headers =
      [{"content-type", "application/x-www-form-urlencoded"}] |> with_auth(opts)

    uri =
      opts
      |> base_uri()
      |> endpoint("query")

    params =
      query_string
      |> with_q()
      |> with_database(opts)
      |> with_epoch(opts)
      |> URI.encode_query()

    http_opts = opts[:http_opts] || []

    uri
    |> HTTPoison.post!(params, headers, http_opts)
    |> query_response(opts)
  rescue
    e in [HTTPoison.Error] -> {:error, e.reason || e.message}
  end

  @doc """
  ship datapoints to a pre-existing database when UDP is unreliable or unacceptable
  """
  def write(encoded_points, opts) do
    headers =
      [{"content-type", "application/x-www-form-urlencoded"}] |> with_auth(opts)

    uri =
      opts
      |> base_uri()
      |> endpoint("write")

    params =
      []
      |> with_database(opts)
      |> with_precision(opts)
      |> with_retention_policy(opts)

    request_opts = Keyword.put(opts[:http_opts] || [], :params, params)

    uri
    |> HTTPoison.post!(encoded_points, headers, request_opts)
    |> write_response(opts)
  rescue
    e in [HTTPoison.Error] -> {:error, e.reason || e.message}
  end

  # request builder functions
  defp with_auth(headers, opts) do
    # add the authorization header if needed
    case opts[:auth] do
      {username, password} ->
        [
          {"Authorization",
           "Basic " <> Base.encode64("#{username}:#{password}")}
          | headers
        ]

      _ ->
        headers
    end
  end

  defp with_q(query_string) do
    [{"q", query_string}]
  end

  defp with_database(params, %{database: db}) when is_binary(db),
    do: [{"db", db} | params]

  defp with_database(params, _opts), do: params

  defp with_precision(params, opts) do
    case Map.get(@epoch_map, opts[:epoch]) do
      p when not is_nil(p) -> [{"precision", p} | params]
      _ -> params
    end
  end

  defp with_epoch(params, opts) do
    case Map.get(@epoch_map, opts[:epoch]) do
      e when not is_nil(e) -> [{"epoch", e} | params]
      _ -> params
    end
  end

  defp with_retention_policy(params, opts) do
    case opts[:retention_policy] do
      rp when not is_nil(rp) -> [{"rp", rp} | params]
      _ -> params
    end
  end

  # uri builder fns
  defp base_uri(opts), do: "http#{secure?(opts)}://#{opts[:host]}#{port?(opts)}"

  defp secure?(%{secure?: true}), do: "s"
  defp secure?(_opts), do: ""

  defp port?(%{http_port: port}) when is_integer(port), do: ":#{port}"
  defp port?(_opts), do: ""

  defp endpoint(base, which), do: base <> "/" <> which

  # response handlers
  defp ping_response(%{status_code: 204}), do: :ok
  defp ping_response(resp), do: {:error, resp}

  defp query_response(%{status_code: 200, body: body}, opts),
    do: opts.json_encoder.decode!(body)

  defp query_response(resp, opts), do: process_error_response(resp, opts)

  defp write_response(%{status_code: 204}, _opts), do: :ok
  defp write_response(resp, opts), do: process_error_response(resp, opts)

  defp process_error_response(%{headers: head, body: body} = resp, opts) do
    if json?(head) do
      {:error, body |> opts.json_encoder.decode!() |> Map.get("error")}
    else
      {:error, resp}
    end
  end

  defp json?(headers), do: headers |> Enum.any?(&content_type_json/1)

  defp content_type_json({"Content-Type", "application/json"}), do: true
  defp content_type_json(_), do: false
end