verk

defmodule Verk.QueueStatsCounters do
  @moduledoc """
  This module is responsible for abstracting the logic of keeping counters for
  each queue.
  """

  @counters_table :queue_stats
  @ets_options [:ordered_set, :named_table, read_concurrency: true, keypos: 1]

  @doc """
  Initializes the ets tables for the queue stats.
  """
  @spec init :: :ok
  def init do
    :ets.new(@counters_table, @ets_options)
    :ok
  end

  @doc """
  It outputs the current stats about each queue and `total` and search for a `prefix` if provided
  """
  def all(prefix) do
    prefix = to_charlist(prefix)

    match =
      if prefix != '' do
        [
          {{prefix ++ :"$1", :"$2", :"$3", :"$4", :_, :_}, [],
           [{{prefix ++ :"$1", :"$2", :"$3", :"$4"}}]}
        ]
      else
        [{{:"$1", :"$2", :"$3", :"$4", :_, :_}, [], [{{:"$1", :"$2", :"$3", :"$4"}}]}]
      end

    :ets.select(@counters_table, match)
  end

  @doc """
  It Resets the started counter of a `queue`
  """
  @spec reset_started(binary) :: :ok
  def reset_started(queue) do
    queue = to_charlist(queue)

    unless :ets.update_element(@counters_table, queue, {2, 0}) do
      true = :ets.insert_new(@counters_table, new_element(queue))
    end
  end

  @doc """
  Updates the counters according to the event that happened.
  """
  @spec register(:started | :finished | :failed, binary) :: integer
  def register(:started, queue) do
    update = {2, 1}
    update_counters(to_charlist(queue), update)
    update_counters(:total, update)
  end

  def register(:finished, queue) do
    updates = [{3, 1}, {2, -1}]
    update_counters(to_charlist(queue), updates)
    update_counters(:total, updates)
  end

  def register(:failed, queue) do
    updates = [{4, 1}, {2, -1}]
    update_counters(to_charlist(queue), updates)
    update_counters(:total, updates)
  end

  @doc """
  Saves processed and failed total counts to Redis.
  """
  @spec persist :: :ok | {:error, term}
  def persist do
    cmds =
      Enum.reduce(counters(), [], fn {queue, _started, processed, failed, last_processed,
                                      last_failed},
                                     commands ->
        delta_processed = processed - last_processed
        delta_failed = failed - last_failed
        :ets.update_counter(@counters_table, queue, [{5, delta_processed}, {6, delta_failed}])

        [
          incrby(queue, :processed, delta_processed)
          | [incrby(queue, :failed, delta_failed) | commands]
        ]
      end)

    cmds |> Enum.reject(&(&1 == nil)) |> flush_to_redis!
  end

  defp flush_to_redis!([]), do: :ok

  defp flush_to_redis!(cmds) do
    case Redix.pipeline(Verk.Redis, cmds) do
      {:ok, _} -> :ok
      {:error, reason} -> {:error, reason}
    end
  end

  defp counters, do: :ets.tab2list(@counters_table)

  # started, finished, failed, last_started, last_failed
  defp update_counters(queue, operations) do
    :ets.update_counter(@counters_table, queue, operations, new_element(queue))
  end

  defp new_element(queue), do: {queue, 0, 0, 0, 0, 0}

  defp incrby(_, _, 0), do: nil

  defp incrby(:total, attribute, increment) do
    ["INCRBY", "stat:#{attribute}", increment]
  end

  defp incrby(queue, attribute, increment) do
    ["INCRBY", "stat:#{attribute}:#{queue}", increment]
  end
end