rabbit_mq

defmodule MQ.Topology.Setup do
  @moduledoc """
  Use to perform a one-off setup of exchanges
  and queues for `dev` and `test` environments.
  """

  alias AMQP.{
    Channel,
    Connection,
    Exchange,
    Queue
  }

  alias MQ.AMQPConfig
  alias MQ.Topology.Config
  alias MQ.Topology.Config.{DeadLetterQueueConfig, QueueConfig, TopicExchangeConfig}

  require Logger

  defmodule Result do
    defstruct channel: nil, exchanges: [], errors: []
  end

  @spec run() :: {:ok, any()}
  def run do
    topology = Config.gen()
    amqp_url = AMQPConfig.url()

    with {:ok, connection} <- Connection.open(amqp_url),
         {:ok, channel} <- Channel.open(connection),
         results <- setup(channel, topology),
         :ok <- Channel.close(channel) do
      {:ok, results}
    end
  end

  defp setup(channel, topology) do
    topology
    |> Task.async_stream(&setup_exchange(channel, &1))
    |> Enum.to_list()
  end

  defp setup_exchange(channel, %TopicExchangeConfig{
         name: exchange,
         durable: durable,
         queues: queues
       }) do
    channel
    |> assert_exchange(:topic, exchange, durable)
    |> setup_queues(queues)
  end

  defp setup_queues(channel, queues) when is_list(queues) do
    Enum.map(queues, &setup_queues(channel, &1))
  end

  defp setup_queues(
         channel,
         {%DeadLetterQueueConfig{} = dead_letter_queue, %QueueConfig{} = queue}
       ) do
    channel
    |> assert_dead_letter_queue(dead_letter_queue)
    |> assert_queue(queue)
    |> bind_queue(queue)
  end

  defp assert_exchange(channel, :topic, name, durable) do
    :ok = Exchange.topic(channel, name, durable: durable)
    channel
  end

  defp assert_dead_letter_queue(channel, %DeadLetterQueueConfig{name: name, durable: durable}) do
    {:ok, _} = Queue.declare(channel, name, durable: durable)
    channel
  end

  defp assert_queue(channel, %QueueConfig{
         name: name,
         durable: durable,
         exclusive: exclusive,
         args: arguments
       }) do
    {:ok, %{queue: queue}} =
      Queue.declare(channel, name, durable: durable, exclusive: exclusive, arguments: arguments)

    {channel, queue}
  end

  defp bind_queue({channel, queue}, %QueueConfig{binding: {exchange, "", routing_key}} = config) do
    :ok = Queue.bind(channel, queue, exchange, routing_key: routing_key)
    result(queue, config)
  end

  defp bind_queue(
         {channel, queue},
         %QueueConfig{binding: {exchange, queue, routing_key}} = config
       ) do
    :ok = Queue.bind(channel, queue, exchange, routing_key: routing_key)
    result(queue, config)
  end

  defp result(queue, %QueueConfig{binding: {exchange, _queue, routing_key}} = config) do
    result =
      config
      |> Map.take([:durable, :exclusive, :args])
      |> Map.merge(%{
        exchange: exchange,
        queue: queue,
        routing_key: routing_key
      })

    Logger.debug("Declared #{queue} queue: #{inspect(result)}")

    result
  end
end