nldoc_stream

defmodule NLdoc.Stream.ConsumerServer do
  @moduledoc """
  A GenServer responsible for managing message consumption and forwarding events
  to the calling Plug process.
  """
  use GenServer
  require Logger

  alias NLdoc.Stream.{Message, Message.Header}

  @type state() :: %{
          caller: pid(),
          id: String.t(),
          channel: AMQP.Channel.t(),
          queue: String.t(),
          consumer_tag: String.t()
        }

  @spec start_link(keyword()) :: GenServer.on_start()
  def start_link(opts) do
    GenServer.start_link(__MODULE__, opts, name: opts[:name])
  end

  @impl true
  @spec init(%{
          caller: pid(),
          id: String.t(),
          offset: :first | number(),
          queue: String.t(),
          channel: AMQP.Channel.t()
        }) :: {:ok, state()} | {:stop, term()}
  def init(state = %{}) do
    case state.channel
         |> consume(state.queue, self(), offset: state.offset, filter_value: state.id) do
      {:ok, consumer_tag} ->
        {:ok, %{state | consumer_tag: consumer_tag}}

      {:error, reason} ->
        {:stop, reason}
    end
  end

  @impl true
  def handle_info({:basic_consume_ok, _info}, state) do
    {:noreply, state}
  end

  def handle_info({:basic_cancel, %{consumer_tag: tag}}, state = %{caller: caller}) do
    caller |> send({:unsubscribe, tag})
    {:stop, :normal, state}
  end

  def handle_info(
        {:basic_deliver, payload, msg},
        state = %{caller: caller, id: id, channel: channel}
      ) do
    if msg |> Message.matches_stream_filter?(id) do
      caller |> send({:message, payload})
    end

    channel |> AMQP.Basic.ack(msg.delivery_tag)
    {:noreply, state}
  end

  def handle_info(:unsubscribe, state = %{channel: channel, consumer_tag: tag}) do
    {:ok, _} = channel |> AMQP.Basic.cancel(tag)
    {:noreply, state}
  end

  @impl true
  def terminate(_reason, %{caller: caller, channel: channel, consumer_tag: tag}) do
    caller |> send({:unsubscribe, tag})
    channel |> AMQP.Queue.unsubscribe(tag)
    :ok
  end

  @type offset_opt() :: {:offset, integer() | :first | nil}
  @type filter_value_opt() :: {:filter_value, String.t() | nil}

  @type opt :: offset_opt() | filter_value_opt()

  @spec consume(
          channel :: AMQP.Channel.t(),
          queue :: String.t(),
          consumer_pid :: pid(),
          opts :: [opt()]
        ) :: {:ok, String.t()} | AMQP.Basic.error()
  defp consume(channel, queue, consumer_pid, opts),
    do: channel |> AMQP.Basic.consume(queue, consumer_pid, arguments: opts |> amqp_arguments())

  @spec amqp_arguments(arguments :: [opt()]) :: [Header.t()]

  defp amqp_arguments([{:filter_value, value} | xs]) when is_binary(value),
    do: [{"x-stream-filter-value", :longstr, value} | xs |> arguments()]

  defp arguments([{:offset, offset} | xs]) when is_integer(offset),
    do: [{"x-stream-offset", :long, offset} | xs |> arguments()]

  defp arguments([{:offset, :first} | xs]),
    do: [{"x-stream-offset", :longstr, "first"} | xs |> arguments()]

  defp arguments([_ | xs]),
    do: xs |> arguments()

  defp arguments([]),
    do: []
end