influx

defmodule Influx do
  @moduledoc Influx.Docs.influx()
  require Influx.Util
  import Influx.Util, only: [bangify!: 1, safe_in: 2]
  import Influx.Config, only: [write_options: 1]

  ### Types ###

  @typedoc ~S"Influx host url."
  @type host :: URI.t() | String.t()

  @typedoc ~S"Influx write option."
  @type write_option ::
          :database | :username | :password | :precision | :consistency | :retention_policy

  @typedoc ~S"Influx write options."
  @type write_options :: [{write_option, term}]

  @typedoc ~S"Influx request error."
  @type error :: {:error, atom | Influx.Error.t()}

  @typedoc ~S"Influx write and query response."
  @type result :: {:ok, Influx.Response.t()} | error

  @typedoc ~S"Influx profiling.."
  @type profile :: :all | :block | :goroutine | :heap | :mutex | :threadcreate

  ### ###

  @doc @moduledoc
  defmacro __using__(opts \\ []) do
    otp = opts[:otp_app] || raise "Need to set `otp_app:`."
    adapter = opts[:adapter]

    quote location: :keep do
      unless @moduledoc do
        @moduledoc Influx.Docs.influx()
      end

      @config Application.get_env(unquote(otp), __MODULE__)
      @adapter @config[:adapter] || unquote(adapter) || Influx.Adapters.HTTPX
      @url URI.parse(@config[:url] || "")
      @host to_string(%{
              @url
              | authority: nil,
                fragment: nil,
                query: nil,
                userinfo: nil
            })

      default_opts =
        case String.split(@url.userinfo || "", ":", trim: true) do
          [] -> []
          [user] -> [username: user]
          [user, password] -> [username: user, password: password]
        end

      @default_write_opts if @config[:database],
                            do: Keyword.put(default_opts, :database, @config[:database]),
                            else: default_opts

      @doc Influx.Docs.ping()
      @spec ping :: :ok | Influx.error()
      def ping, do: Influx.ping(@adapter, @host)

      @doc ~S"""
      See `ping/0`.
      """
      @spec ping! :: :ok | no_return
      def ping!, do: Influx.ping(@adapter, @host)

      @doc Influx.Docs.info()
      @spec info :: {:ok, Influx.Info.t()} | Influx.error()
      def info, do: Influx.info(@adapter, @host)

      @doc ~S"""
      See `info/0`.
      """
      @spec info! :: Influx.Info.t() | no_return
      def info!, do: Influx.info!(@adapter, @host)

      @doc Influx.Docs.debug()
      @spec debug(atom | [atom]) :: {:ok, map} | Influx.error()
      def debug(getter \\ []), do: Influx.debug(@adapter, @host, getter)

      @doc ~S"""
      See `debug/0`.
      """
      @spec debug!(atom | [atom]) :: map | no_return
      def debug!(getter \\ []), do: Influx.debug!(@adapter, @host, getter)

      @doc Influx.Docs.requests()
      @spec requests(pos_integer) :: {:ok, map} | Influx.error()
      def requests(duration \\ 10), do: Influx.requests(@adapter, @host, duration)

      @doc ~S"""
      See `requests/0`.
      """
      @spec requests!(pos_integer) :: map | no_return
      def requests!(duration \\ 10), do: Influx.requests!(@adapter, @host, duration)

      @doc Influx.Docs.profile()
      @spec profile(Influx.profile()) :: {:ok, map} | Influx.error()
      def profile(profile \\ :all), do: Influx.profile(@adapter, @host, profile)

      @doc ~S"""
      See `requests/0`.
      """
      @spec profile!(Influx.profile()) :: map | no_return
      def profile!(profile \\ :all), do: Influx.profile!(@adapter, @host, profile)

      @doc Influx.Docs.write()
      @spec write(String.t(), Influx.Config.write_options()) :: :ok | Influx.error()
      def write(write, opts \\ []),
        do: Influx.write(@adapter, @host, write, Keyword.merge(@default_write_opts, opts))

      @doc ~S"""
      See `requests/0`.
      """
      @spec write!(Influx.profile(), Influx.Config.write_options()) :: map | no_return
      def write!(write, opts \\ []),
        do: Influx.write!(@adapter, @host, write, Keyword.merge(@default_write_opts, opts))

      @doc Influx.Docs.write()
      @spec query(String.t(), Influx.Config.write_options()) :: :ok | Influx.error()
      def query(query, opts \\ []),
        do: Influx.query(@adapter, @host, query, Keyword.merge(@default_write_opts, opts))

      @doc ~S"""
      See `requests/0`.
      """
      @spec query!(Influx.profile(), Influx.Config.write_options()) :: map | no_return
      def query!(query, opts \\ []),
        do: Influx.query!(@adapter, @host, query, Keyword.merge(@default_write_opts, opts))

      ### Higher Level ###

      @doc ~S"""
      Select from InfluxDB.

      Still WIP.
      """
      @spec select(String.t() | [String.t()], Keyword.t()) :: {:ok, [map]} | Influx.error()
      def select(fields, opts),
        do: Influx.select(@adapter, @host, fields, Keyword.merge(@default_write_opts, opts))
    end
  end

  @doc Influx.Docs.ping()
  @spec ping(module, host) :: :ok | Influx.error()
  def ping(adapter, host), do: with({:ok, _} <- adapter.ping(host), do: :ok)

  @doc ~S"""
  See `ping/2`.
  """
  @spec ping!(module, host) :: :ok | no_return
  def ping!(adapter, host), do: bangify!(ping(adapter, host))

  @doc Influx.Docs.info()
  @spec info(module, host) :: {:ok, Influx.Info.t()} | Influx.error()
  def info(adapter, host) do
    with {:ok, %{build: build, version: version}} <- adapter.ping(host),
         {:ok, version} <- Version.parse(version || "") do
      {:ok,
       %Influx.Info{
         version: version,
         build: build
       }}
    end
  end

  @doc ~S"""
  See `info/2`.
  """
  @spec info!(module, host) :: Influx.Info.t() | no_return
  def info!(adapter, host), do: bangify!(info(adapter, host))

  @doc Influx.Docs.debug()
  @spec debug(module, host, atom | [atom]) :: {:ok, map} | Influx.error()
  def debug(adapter, host, getter \\ []) do
    with {:ok, %{data: data}} <- adapter.debug(host) do
      {:ok, safe_in(data, getter)}
    end
  end

  @doc ~S"""
  See `debug/3`.
  """
  @spec debug!(module, host, atom | [atom]) :: map | no_return
  def debug!(adapter, host, getter \\ []), do: bangify!(debug(adapter, host, getter))

  @doc Influx.Docs.requests()
  @spec requests(module, host, pos_integer) :: {:ok, map} | Influx.error()
  def requests(adapter, host, duration \\ 10) do
    with {:ok, %{data: data}} <- adapter.requests(host, duration) do
      {:ok, data}
    end
  end

  @doc ~S"""
  See `requests/3`.
  """
  @spec requests!(module, host, pos_integer) :: map | no_return
  def requests!(adapter, host, duration \\ 10), do: bangify!(requests(adapter, host, duration))

  @doc Influx.Docs.profile()
  @spec profile(module, host, Influx.profile()) :: {:ok, map} | Influx.error()
  def profile(adapter, host, profile \\ :all) do
    with {:ok, %{data: data}} <- adapter.profile(host, profile) do
      {:ok, data}
    end
  end

  @doc ~S"""
  See `profile/3`.
  """
  @spec profile!(module, host, Influx.profile()) :: map | no_return
  def profile!(adapter, host, profile \\ :all), do: bangify!(profile(adapter, host, profile))

  @spec query(module, URI.t() | String.t(), String.t(), Influx.Config.write_options()) ::
          {:ok, [map]} | Influx.error()
  def query(adapter, host, query, opts) do
    with {:ok, options} <- write_options(opts),
         {:ok, %{data: data}} <- adapter.query(host, query, options) do
      {:ok, data}
    end
  end

  @doc ~S"""
  See `query/4`.
  """
  @spec query!(module, URI.t() | String.t(), String.t(), Influx.Config.write_options()) ::
          :ok | no_return
  def query!(adapter, host, query, opts \\ []), do: bangify!(query(adapter, host, query, opts))

  @doc Influx.Docs.write()
  @spec write(module, URI.t() | String.t(), String.t(), Influx.Config.write_options()) ::
          :ok | Influx.error()
  def write(adapter, host, write, opts \\ []) do
    with {:ok, options} <- write_options(opts),
         {:ok, _response} <- adapter.write(host, write, options) do
      :ok
    end
  end

  @doc ~S"""
  See `write/4`.
  """
  @spec write!(module, URI.t() | String.t(), String.t(), Influx.Config.write_options()) ::
          :ok | no_return
  def write!(adapter, host, write, opts \\ []), do: bangify!(write(adapter, host, write, opts))

  def select(adapter, host, fields, opts) do
    query = "SELECT " <> prepare_fields(fields) <> " FROM #{opts[:from]}"
    opts = Keyword.delete(opts, :from)

    with {:ok, [%{result: result}]} <- query(adapter, host, query, opts),
         {:ok, %{columns: columns, values: values}} <- result do
      {:ok,
       Enum.map(
         values,
         fn row ->
           columns
           |> Enum.zip(row)
           |> Enum.into(%{})
           |> Map.update!("time", &NaiveDateTime.from_iso8601!/1)
         end
       )}
    end
  end

  defp prepare_fields(fields) when is_list(fields), do: Enum.join(fields, ",")
  defp prepare_fields(field), do: field
end