ex_data_dog

defmodule ExDataDog do
  use GenServer

  @default_port      8125
  @default_host      "127.0.0.1"
  @default_namespace nil
  @default_sink      nil
  @timing_stub       1.234

  @type statsd_port :: number
  @type host        :: String.t
  @type sink        :: String.t
  @type name        :: String.t
  @type namespace   :: String.t
  @type options     :: [
    port:      statsd_port,
    host:      host,
    namespace: namespace,
    sink:      sink,
    name:      name
  ]

  def start_link(options \\ []) do
    state = %{
      port:      Keyword.get(options, :port,      Application.get_env(:ex_data_dog, :port,      @default_port)),
      host:      Keyword.get(options, :host,      Application.get_env(:ex_data_dog, :host,      @default_host)) |> parse_host(),
      namespace: Keyword.get(options, :namespace, Application.get_env(:ex_data_dog, :namespace, @default_namespace)),
      sink:      Keyword.get(options, :sink,      Application.get_env(:ex_data_dog, :sink,      @default_sink)),
      socket:    nil
    }

    GenServer.start_link(__MODULE__, state, Keyword.merge([name: __MODULE__], options))
  end

  def stop(name \\__MODULE__),
    do: GenServer.stop(name)

  def flush(name \\__MODULE__),
    do: GenServer.call(name, :flush)

  defp parse_host(host) when is_binary(host) do
    case host |> to_char_list() |> :inet.parse_address() do
      {:error, _}    -> host |> String.to_atom()
      {:ok, address} -> address
    end
  end

  def counter(amount, metric, options \\ default_options()) do
    sampling options, fn(decision) ->
      case decision do
        {:sample, rate} ->
          transmit({metric, amount, :c}, options, rate)
          amount
        _ ->
          amount
      end
    end
  end

  def count(collection, metric, options \\ default_options()) do
    sampling options, fn(decision) ->
      case decision do
        {:sample, rate} ->
          transmit({metric, Enum.count(collection), :c}, options, rate)
          collection
        _ ->
          collection
      end
    end
  end

  def increment(metric, options \\ default_options()),
    do: counter(1, metric, options)

  def decrement(metric, options \\ default_options()),
    do: counter(-1, metric, options)

  def gauge(amount, metric, options \\ [tags: []]) do
    transmit({metric, amount, :g}, options)
    amount
  end

  def set(member, metric, options \\ [tags: []]) do
    transmit({metric, member, :s}, options)
    member
  end

  def timer(amount, metric, options \\ default_options()) do
    sampling options, fn(decision) ->
      case decision do
        {:sample, rate} ->
          transmit({metric, amount, :ms}, options, rate)
          amount
        _ ->
          amount
      end
    end
  end

  def timing(metric, fun, options \\ default_options()) do
    sampling options, fn(decision) ->
      case decision do
        {:sample, rate} ->
          {time, value} = :timer.tc(fun)
          amount        = if Application.get_env(:ex_data_dog, :test_mode, false), do: @timing_stub, else: time / 1000.0

          transmit({metric, amount, :ms}, options, rate)
          value
        _ ->
          fun.()
      end
    end
  end

  @doc """
  Record a histogram value (DogStatsD-only).

  * `sample_rate`: Limit how often the metric is collected
  * `tags`: Add tags to entry (DogStatsD-only)

  It returns the value given as the first argument, making it suitable for
  pipelining.
  """
  def histogram(amount, metric, options \\ default_options()) do
    sampling options, fn(decision) ->
      case decision do
        {:sample, rate} ->
          transmit({metric, amount, :h}, options, rate)
          amount
        _ ->
          amount
      end
    end
  end

  @doc """
  Time a function using a histogram metric (DogStatsD-only).

  * `sample_rate`: Limit how often the metric is collected
  * `tags`: Add tags to entry (DogStatsD-only)

  It returns the result of the function call, making it suitable
  for pipelining.
  """
  def histogram_timing(metric, fun, options \\ default_options()) do
    sampling options, fn(decision) ->
      case decision do
        {:sample, rate} ->
          {time, value} = :timer.tc(fun)
          amount        = if Application.get_env(:ex_data_dog, :test_mode, false), do: @timing_stub, else: time / 1000.0

          transmit({metric, amount, :h}, options, rate)
          value
        _ ->
          fun.()
        end
      end
    end

  defp default_options do
    [
      sample_rate: 1,
      tags:        [],
      name:        __MODULE__,
    ]
  end

  @doc """
  Emit event.

  `text` supports line breaks, only first 4KB will be transmitted.

  Available options:
  * `tags`: Add tags to entry (DogStatsD-only)
  * `priority`: Can be *normal* or *low*, default *normal*
  * `alert_type`: Can be *error*, *warning*, *info* or *success*, default *info*
  * `aggregation_key`: Assign an aggregation key to the event, to group it with some others
  * `hostname`: Assign a hostname to the event
  * `source_type_name`: Assign a source type to the event
  * `date_happened`: Assign a timestamp to the event, default current time

  It returns the title of the event, making it suitable for pipelining.
  """
  def event(title, text \\ "", options \\ [tags: []]) do
    transmit({:event, title, text, options}, options)
    title
  end


  defp sampling(options, fun) when is_list(options) do
    case Keyword.get(options, :sample_rate, 1) do
      1 ->
        fun.({:sample, 1})
      sample_rate ->
        sample(sample_rate, fun)
    end
  end

  defp sample(sample_rate, fun) do
    case :rand.uniform <= sample_rate do
      true ->
        fun.({:sample, sample_rate})
      _ ->
        fun.(:no_sample)
    end
  end

  defp transmit(message, options),
    do: transmit(message, options, 1)
  defp transmit(message, options, sample_rate) do
    name = Keyword.get(options, :name, __MODULE__)

    GenServer.cast(name, {:transmit, message, options, sample_rate})
  end

  defp packet({:event, title, text, opts}, namespace, tags, sample_rate) do
    text = text |> String.replace("\n","\\n") |> String.slice(0, 4096)

    [
      stat_name(:event, namespace),
      ":#{title}|#{text}",
      opts[:priority]         && "|p:#{opts[:priority]}"         || "",
      opts[:alert_type]       && "|t:#{opts[:alert_type]}"       || "",
      opts[:source_type_name] && "|s:#{opts[:source_type_name]}" || "",
      opts[:aggregation_key]  && "|k:#{opts[:aggregation_key]}"  || "",
      opts[:hostname]         && "|h:#{opts[:hostname]}"         || "",
      opts[:date_happened]    && "|d:#{opts[:date_happened]}"    || "",
      sample_rate_suffix(sample_rate),
      tags_suffix(tags)
    ]
  end

  defp packet({key, value, type}, namespace, tags, sample_rate) do
    [
      stat_name(key, namespace),
      ":#{value}|#{type}",
      sample_rate_suffix(sample_rate),
      tags_suffix(tags)
    ]
  end

  defp sample_rate_suffix(1), do: ""
  defp sample_rate_suffix(sample_rate) do
    [
      "|@",
      :io_lib.format('~.2f', [sample_rate]),
    ]
  end

  defp tags_suffix([]), do: ""
  defp tags_suffix(tags) do
    [
      "|#",
      Enum.join(tags, ",")
    ]
  end

  defp stat_name(key, nil),
    do: key
  defp stat_name(key, namespace),
    do: "#{namespace}.#{key}"

  def handle_cast({:transmit, message, options, sample_rate}, %{sink: sink} = state) when is_list(sink) do
    tags = Keyword.get(options, :tags, [])
    pkt  = message |> packet(state.namespace, tags, sample_rate) |> IO.iodata_to_binary()

    {:noreply, %{state | sink: [pkt | sink]}}
  end

  def handle_cast({:transmit, message, options, sample_rate}, state) do
    tags          = Keyword.get(options, :tags, [])
    pkt           = message |> packet(state.namespace, tags, sample_rate) |> IO.iodata_to_binary()
    {:ok, socket} = :gen_udp.open(0, [:binary])

    :gen_udp.send(socket, state.host, state.port, pkt)
    :gen_udp.close(socket)

    {:noreply, state}
  end

  def handle_call(:flush, _from, state),
    do: {:reply, :ok, state}
end