telemetry_influxdb

defmodule TelemetryInfluxDB.HTTP.EventHandler do
  @moduledoc false
  require Logger
  alias TelemetryInfluxDB.HTTP.Pool

  alias TelemetryInfluxDB.Formatter
  import HTTPoison.Response
  alias TelemetryInfluxDB, as: InfluxDB
  require Logger

  @spec start_link(InfluxDB.config()) :: GenServer.on_start()
  def start_link(config) do
    GenServer.start_link(__MODULE__, config)
  end

  def init(config) do
    config = %{config | port: :erlang.integer_to_binary(config.port)}
    Process.flag(:trap_exit, true)
    handler_ids = attach_events(config.events, config)

    {:ok, %{handler_ids: handler_ids}}
  end

  @spec attach_events(InfluxDB.event_spec(), InfluxDB.config()) :: list(InfluxDB.handler_id())
  def attach_events(event_specs, config) do
    Enum.map(event_specs, fn e ->
      pool_name = Pool.get_name(config.reporter_name)

      telemetry_config =
        Map.delete(config, :events)
        |> Map.put(:pool_name, pool_name)
        |> Map.put(:metadata_tag_keys, e[:metadata_tag_keys] || [])

      handler_id = handler_id(e.name, config.reporter_name)
      :ok = :telemetry.attach(handler_id, e.name, &__MODULE__.handle_event/4, telemetry_config)
      handler_id
    end)
  end

  @spec handle_event(
          InfluxDB.event_name(),
          InfluxDB.event_measurements(),
          InfluxDB.event_metadata(),
          InfluxDB.config()
        ) :: :ok
  def handle_event(event, measurements, metadata, config) do
    url = build_url(config)

    event_tags = Map.get(metadata, :tags, %{})
    event_metadatas = Map.take(metadata, config.metadata_tag_keys)

    tags =
      Map.merge(config.tags, event_tags)
      |> Map.merge(event_metadatas)

    body = Formatter.format(event, measurements, tags)

    headers = Map.merge(authentication_header(config), binary_data_header())

    :wpool.cast(config.pool_name, {__MODULE__, :send_event, [url, body, headers]})
  end

  @spec send_event(binary, any, any) :: :ok
  def send_event(url, body, headers) do
    process_response(HTTPoison.post(url, body, headers))
  end

  def handle_info({:EXIT, _pid, reason}, state) do
    {:stop, reason, state}
  end

  defp process_response({:ok, %HTTPoison.Response{status_code: 204}}), do: :ok

  defp process_response({:ok, %HTTPoison.Response{status_code: 404}}) do
    Logger.error("Failed to push data to InfluxDB. Invalid credentials")
    :ok
  end

  defp process_response({:ok, %{status_code: 401}}) do
    Logger.error("Failed to push data to InfluxDB. Invalid credentials")
    :ok
  end

  defp process_response(res) do
    Logger.error("Failed to send event to InfluxDB. Response: #{inspect(res)}")
    :ok
  end

  defp build_url(%{version: :v1, host: host, port: port, db: db}) do
    query = URI.encode_query(%{db: db})
    host <> ":" <> port <> "/write?" <> query
  end

  defp build_url(%{version: :v2, host: host, port: port, org: org, bucket: bucket}) do
    query = URI.encode_query(%{bucket: bucket, org: org})
    host <> ":" <> port <> "/api/v2/write?" <> query
  end

  defp authentication_header(%{version: :v1, username: username, password: password}) do
    %{"Authorization" => "Basic #{Base.encode64(username <> ":" <> password)}"}
  end

  defp authentication_header(%{version: :v2, token: token}) do
    %{"Authorization" => "Token #{token}"}
  end

  defp binary_data_header() do
    %{"Content-Type" => "text/plain"}
  end

  def terminate(_reason, state) do
    for handler_id <- state.handler_ids do
      :telemetry.detach(handler_id)
    end

    :ok
  end

  @spec handler_id(InfluxDB.event_name(), binary()) :: InfluxDB.handler_id()
  defp handler_id(event_name, prefix) do
    {__MODULE__, event_name, prefix}
  end
end