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
  @moduledoc """
  This is the single module of the TimeQueue library.

  Optimization may lead to a different data structure, but at the
  momement the time queue is implemented as a simple list were entries
  are sorted with timestamps.
  """
  require Record

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

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

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

  @type timespec :: {pos_integer, timespec_unit}
  @type ttl :: timespec | integer

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

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

  defguardp is_timespec(timespec)
            when is_integer(elem(timespec, 0)) and elem(timespec, 1) in @timespec_units

  @doc """
  Creates an empty time queue.

      iex> tq = TimeQueue.new()
      []
      iex> TimeQueue.peek(tq)
      :empty
  """
  @spec new :: t
  def new,
    do: []

  @doc """
  Returns the next event of the queue with the current system time as `now`.

  See `peek/2`.
  """
  @spec peek(t) :: :empty | {:delay, non_neg_integer} | {:ok, entry}
  def peek(tq),
    do: peek(tq, system_time())

  @doc """
  Returns the next event of the queue according to the given current time in
  milliseconds.

  Possible return values are:

  - `:empty`
  - `{:ok, entry}` if the timestamp of the first entry is `<=` to the given
    current time.
  - `{:delay, ms}` if the timestamp of the first entry is `>` to the given
    current time. The remaining amount of milliseconds is returned.

  ### Example

      iex> {:ok, _tref, tq} = TimeQueue.new() |> TimeQueue.enqueue(100, :hello, _now = 0)
      iex> TimeQueue.peek(tq, _now = 20)
      {:delay, 80}
      iex> {:ok, _} = TimeQueue.peek(tq, _now = 100)
  """
  @spec peek(t, now_ms :: pos_integer) ::
          :empty | {:delay, non_neg_integer} | {:ok, entry}

  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}

  @doc """
  Extracts the next event of the queue with the current system time as `now`.

  See `pop/2`.
  """
  @spec pop(t) ::
          :empty | {:delay, non_neg_integer} | {:ok, TimeQueue.entry()}
  def pop(tq),
    do: pop(tq, system_time())

  @doc """
  Extracts the next event of the queue according to the given current time in
  milliseconds.

  Possible return values are:

  - `:empty`
  - `{:ok, entry, new_queue}` if the timestamp of the first entry is `<=` to the
    given current time. The entry is deleted from `new_queue`.
  - `{:delay, ms}` if the timestamp of the first entry is `>` to the given
    current time. The remaining amount of milliseconds is returned.

  ### Example

      iex> {:ok, _tref, tq} = TimeQueue.new() |> TimeQueue.enqueue(100, :hello, _now = 0)
      iex> TimeQueue.pop(tq, _now = 20)
      {:delay, 80}
      iex> {:ok, _, _} = TimeQueue.pop(tq, _now = 100)
  """
  @spec pop(t, now_ms :: pos_integer) ::
          :empty | {:delay, non_neg_integer} | {:ok, entry()}
  def pop(tq, now) do
    with {:ok, entry} <- peek(tq, now) do
      tq = delete(tq, entry)
      {:ok, entry, tq}
    end
  end

  @doc """
  Deletes an entry from the queue and returns the new queue.

  The function does not fail if the entry was not found and simply returns the
  queue as-is.
  """
  @spec delete(t, entry) :: t
  def delete(tq, trec() = entry),
    do: tq -- [entry]

  @doc """
  Adds a new entry to the queue with a TTL and the current system time as `now`.

  See `enqueue/4`.
  """
  @spec enqueue(t, ttl, any) :: {:ok, tref, t}
  def enqueue(tq, ttl, val),
    do: enqueue(tq, ttl, val, system_time())

  @doc """
  Adds a new entry to the queue with a TTL relative to the given timestamp in
  milliseconds.

  Returns `{:ok, tref, new_queue}` where `tref` is a timer reference (not
  used yed).
  """
  @spec enqueue(t, ttl, any, now :: integer) :: {:ok, tref, t}
  def enqueue(tq, ttl, val, now_ms)

  def enqueue(tq, ttl, val, now) when is_timespec(ttl),
    do: enqueue_abs(tq, timespec_add(ttl, now), val)

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

  @doc """
  Adds a new entry to the queue with an absolute timestamp.

  Returns `{:ok, tref, new_queue}` where `tref` is a timer reference (not
  used yed).
  """
  @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

  @doc """
  Returns the value of an queue entry.
      iex> tq = TimeQueue.new()
      iex> {:ok, _, tq} = TimeQueue.enqueue(tq, 10, :my_value)
      iex> Process.sleep(10)
      iex> {:ok, entry} = TimeQueue.peek(tq)
      iex> TimeQueue.value(entry)
      :my_value
  """
  @spec value(entry) :: any
  def value(trec(val: val)), do: val

  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]

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

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

  defp ttl_to_milliseconds({_, _} = ttl) when is_timespec(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 timespec_add(ttl, int),
    do: ttl_to_milliseconds(ttl) + int
end