plug_amqp

defmodule Plug.Amqp do
  @moduledoc """
  TODO
  """

  use Supervisor

  require Logger

  alias AMQPHelpers.Reliability.{Consumer, Producer}
  alias Plug.AMQP.Conn

  @supervisor_options ~w(strategy max_restarts max_seconds name)a

  # TODO: Add support to url option, like "amqp://my_broker:5672".

  @typedoc "TODO"
  @type option ::
          Supervisor.option()
          | Supervisor.init_option()
          | {:consumer_options, Consumer.options()}
          | {:plug, module() | {module() | keyword()}}
          | {:producer_options, Consumer.options()}

  @typedoc "TODO"
  @type options :: [option()]

  @spec start_link(options()) :: Supervisor.on_start()
  def start_link(options \\ []) do
    {supervisor_opts, plug_amqp_opts} = Keyword.split(options, @supervisor_options)

    Supervisor.start_link(__MODULE__, plug_amqp_opts, supervisor_opts)
  end

  @impl true
  def init(opts) do
    supervisor = self()

    consumer_opts =
      opts
      |> Keyword.fetch!(:consumer_options)
      |> Keyword.put(:message_handler, &handle_message(supervisor, &1, &2, opts))
      |> Keyword.put(:requeue, get_in(opts, [:consumer_options, :requeue]))
      |> Keyword.put(:task_supervisor, {:via, __MODULE__, {supervisor, Task.Supervisor}})

    producer_opts = Keyword.fetch!(opts, :producer_options)

    children = [
      {Consumer, consumer_opts},
      {Producer, producer_opts},
      Task.Supervisor
    ]

    Supervisor.init(children, strategy: :one_for_one)
  end

  #
  # Name Registry Implementation
  #

  @doc false
  def register_name(_name, _pid), do: :yes

  @doc false
  def unregister_name(name), do: name

  @doc false
  def whereis_name({supervisor, module}), do: whereis_name(supervisor, module)

  @doc false
  def whereis_name(supervisor, module) do
    supervisor
    |> Supervisor.which_children()
    |> Enum.find(&match?({^module, _pid, _kind, _modules}, &1))
    |> elem(1)
  end

  #
  # Message Handling Implementation
  #

  # TODO: Add more options to customize the nack of the messages.

  @spec handle_message(pid(), binary(), map(), options()) :: :ok | {:error, term()}
  defp handle_message(supervisor, payload, meta, opts) do
    producer = whereis_name(supervisor, Producer)

    {plug, plug_opts} = fetch_plug!(opts)
    conn = Conn.conn(producer, payload, meta)

    try do
      conn
      |> plug.call(plug_opts)
      |> send_resp()
    catch
      _kind, reason ->
        Logger.error("Cannot handle message from Plug.AMQP",
          reason: reason,
          stacktace: __STACKTRACE__
        )

        {:error, reason}
    else
      _conn ->
        :ok
    end
  end

  #
  # Misc
  #

  @spec fetch_plug!(options()) :: {module(), keyword()} | no_return()
  defp fetch_plug!(opts) do
    case Keyword.fetch!(opts, :plug) do
      {module, opts} -> {module, opts}
      module -> {module, []}
    end
  end

  defp send_resp(conn = %Plug.Conn{state: :set}), do: Plug.Conn.send_resp(conn)
  defp send_resp(conn = %Plug.Conn{}), do: conn
end