spqueue

defmodule SPQueueWorker do
  use GenServer

  @moduledoc """
  A worker to handle/process messages on a persistent queue.

  You specify the queue_name of the SPQueue to work with.
  This is an atom name of a registered process that is looked up each time.

  The handler_function takes a message from the queue as argument
  and must return :ack or :nack after which the message is dequeued.

  Additionally, each periodic_interval the queue will be processed.

  The worker should be set as delegate in the SPQueue,
  so that it receives :enqueued notification messages
  whenever items have been added to the queue.

  Each time handle_queue_messages is called, be it from the :enqueued
  notification of the queue or from the :periodic_check, all messages
  will be handled until the queue is empty.

  Since the worker is in its own process, it can take its time as needed.
  """

  defstruct queue_name: "pq",
            handler_function: nil,
            periodic_interval: :timer.seconds(5)

  @doc """
  The following options are provided:
  - name: the worker process' name
  - queue_name: the name of the persistent PQ to work with
  - handler_function: function that takes a message to process and returns :ack or :nack
  - periodic_interval: trigger period processing each interval
  """
  @impl true
  def init(opts) do
    name = Keyword.get(opts, :queue_name)
    handler_function = Keyword.get(opts, :handler_function)
    periodic_interval = Keyword.get(opts, :periodic_interval)

    initial_state =
      %__MODULE__{}
      |> Map.update!(:queue_name, fn default -> name || default end)
      |> Map.update!(:handler_function, fn default -> handler_function || default end)
      |> Map.update!(:periodic_interval, fn default -> periodic_interval || default end)

    Process.send_after(self(), :periodic_check, initial_state.periodic_interval)

    {:ok, initial_state}
  end

  def start_link(opts) do
    GenServer.start_link(__MODULE__, opts, opts)
  end

  def start(opts) do
    GenServer.start(__MODULE__, opts, opts)
  end

  def stop(pq) do
    GenServer.stop(pq)
  end

  @impl true
  def handle_info(:enqueued, state) do
    new_state = handle_queue_messages(state)
    {:noreply, new_state}
  end

  @impl true
  def handle_info(:periodic_check, state) do
    new_state = handle_queue_messages(state)
    Process.send_after(self(), :periodic_check, state.periodic_interval)
    {:noreply, new_state}
  end

  def handle_queue_messages(state) do
    pq = Process.whereis(state.queue_name)

    if pq do
      drain_queue(pq, state.handler_function)
    end

    state
  end

  def drain_queue(pq, handler_function) do
    if !SPQueue.empty?(pq) do
      message = SPQueue.head(pq)

      case apply(handler_function, [message]) do
        :ack -> SPQueue.dequeue(pq, ack: true)
        :nack -> SPQueue.dequeue(pq, ack: false)
      end

      drain_queue(pq, handler_function)
    end
  end
end