time_queue

# Current implementation is siply based on a sorted list. It would
# benefit from a proper implementation of abstract ptiority queue
defmodule TimeQueue do
  require Record

  Record.defrecord(:trec, tref: nil, val: nil)

  @ttl_units [
    # :millisecond, # no single millisecond
    :ms,
    :second,
    :seconds,
    :minute,
    :minutes,
    :hour,
    :hours,
    :day,
    :days,
    :week,
    :weeks
  ]

  @type time_unit ::
          :ms
          | :second
          | :seconds
          | :minute
          | :minutes
          | :hour
          | :hours
          | :day
          | :days
          | :week
          | :weeks

  @type ttl :: {pos_integer, time_unit}

  @opaque entry :: record(:trec, tref: {pos_integer, integer}, val: any)

  @opaque t :: [entry]
  @opaque tref :: {integer, integer}

  defguardp is_ttl(ttl) when is_integer(elem(ttl, 0)) and elem(ttl, 1) in @ttl_units

  @spec new :: t
  def new(),
    do: []

  @spec peek(t) :: :empty | {:delay, non_neg_integer} | {:ok, entry}
  @spec peek(t, now_ms :: integer) ::
          :empty | {:delay, non_neg_integer} | {:ok, entry}
  def peek(tq, now \\ system_time())

  def peek([], _),
    do: :empty

  def peek([trec(tref: {ts, _}) = first | _tq], now) when ts <= now,
    do: {:ok, first}

  def peek([trec(tref: {ts, _}) | _tq], now),
    do: {:delay, ts - now}

  @spec pop(t) :: :empty | {:delay, non_neg_integer} | {:ok, TimeQueue.entry()}

  def pop(tq, now \\ system_time()) do
    with {:ok, entry} <- peek(tq, now) do
      tq = delete(tq, entry)
      {:ok, entry, tq}
    end
  end

  @spec delete(t, entry) :: t
  def delete(tq, entry) do
    tq -- [entry]
  end

  @spec enqueue(t, ttl, any, now :: integer) :: {:ok, tref, t}
  def enqueue(tq, ttl, val, now \\ system_time())

  def enqueue(tq, ttl, val, now) when is_ttl(ttl), do: enqueue_abs(tq, ttl_add(ttl, now), val)

  def enqueue(tq, ttl, val, now) when is_integer(ttl),
    do: enqueue_abs(tq, now + ttl, val)

  @spec enqueue_abs(t, end_time :: integer, value :: any) :: {:ok, tref, t}
  def enqueue_abs(tq, ts, val) do
    tref = {ts, :erlang.unique_integer()}
    entry = trec(tref: tref, val: val)
    tq = insert(tq, entry)
    {:ok, tref, tq}
  end

  defp insert([], entry), do: [entry]

  defp insert([trec(tref: {next, _}) = candidate | tq], trec(tref: {ts, _}) = entry)
       when next <= ts,
       do: [candidate | insert(tq, entry)]

  defp insert([trec(tref: {next, _}) = candidate | tq], trec(tref: {ts, _}) = entry)
       when next > ts,
       do: [entry, candidate | tq]

  @spec value(entry) :: any
  def value(trec(val: val)), do: val

  defp system_time(),
    do: :erlang.system_time(:millisecond)

  @spec ttl_to_milliseconds(ttl) :: number

  defp ttl_to_milliseconds({n, :ms}) when is_integer(n) and n > 0,
    do: n

  defp ttl_to_milliseconds({_, _} = ttl) when is_ttl(ttl),
    do: ttl_to_seconds(ttl) * 1000

  defp ttl_to_seconds({seconds, unit}) when unit in [:second, :seconds],
    do: seconds

  defp ttl_to_seconds({minutes, unit}) when unit in [:minute, :minutes],
    do: minutes * 60

  defp ttl_to_seconds({hours, unit}) when unit in [:hour, :hours],
    do: hours * 60 * 60

  defp ttl_to_seconds({days, unit}) when unit in [:day, :days],
    do: days * 24 * 60 * 60

  defp ttl_to_seconds({weeks, unit}) when unit in [:week, :weeks],
    do: weeks * 7 * 24 * 60 * 60

  defp ttl_to_seconds({_, unit}),
    do: raise("Unknown TTL unit: #{unit}")

  defp ttl_add(ttl, int),
    do: ttl_to_milliseconds(ttl) + int
end