amqpx

defmodule Amqpx.Helper do
  @moduledoc """
  Helper functions
  """

  alias AMQP.{Exchange, Queue}

  def consumers_supervisor_configuration(handlers_conf, connection_params) do
    Enum.map(
      handlers_conf,
      &Supervisor.child_spec(
        {Amqpx.Consumer, Map.put(&1, :connection_params, connection_params)},
        id: UUID.uuid1()
      )
    )
  end

  def producer_supervisor_configuration(producer_conf, connection_params) do
    {Amqpx.Producer, Map.put(producer_conf, :connection_params, connection_params)}
  end

  def declare(
        channel,
        %{
          queue: qname,
          opts: opts
        } = queue
      ) do
    case Enum.find(opts[:arguments], &match?({"x-dead-letter-exchange", _, _}, &1)) do
      {_, _, dle} ->
        case Enum.find(opts[:arguments], &match?({"x-dead-letter-routing-key", _, _}, &1)) do
          {_, _, dlrk} ->
            setup_dead_lettering(channel, %{
              queue: "#{qname}_errored",
              exchange: dle,
              routing_key: dlrk
            })

          nil ->
            setup_dead_lettering(channel, %{queue: "#{qname}_errored", exchange: dle})
        end

      nil ->
        nil
    end

    setup_queue(channel, queue)
  end

  def declare(channel, queue) do
    setup_queue(channel, queue)
  end

  def setup_dead_lettering(channel, %{queue: dlq, exchange: ""}) do
    Queue.declare(channel, dlq, durable: true)
  end

  def setup_dead_lettering(channel, %{queue: dlq, exchange: exchange, routing_key: routing_key}) do
    Exchange.declare(channel, exchange, :topic, durable: true)
    Queue.declare(channel, dlq, durable: true)
    Queue.bind(channel, dlq, exchange, routing_key: routing_key)
  end

  def setup_dead_lettering(channel, %{queue: dlq, exchange: exchange}) do
    Exchange.declare(channel, exchange, :topic, durable: true)
    Queue.declare(channel, dlq, durable: true)
  end

  def setup_queue(channel, %{
        queue: queue,
        exchanges: exchanges,
        opts: opts
      }) do
    {:ok, _} = Queue.declare(channel, queue, opts)

    Enum.each(exchanges, &setup_exchange(channel, queue, &1))
  end

  def setup_queue(channel, %{
        queue: queue,
        exchanges: exchanges
      }) do
    {:ok, _} = Queue.declare(channel, queue)

    Enum.each(exchanges, &setup_exchange(channel, queue, &1))
  end

  def setup_exchange(channel, queue, %{
        name: name,
        type: type,
        routing_keys: routing_keys,
        opts: opts
      })
      when type in [:direct, :topic] do
    Exchange.declare(channel, name, type, opts)

    Enum.each(routing_keys, fn rk ->
      :ok = Queue.bind(channel, queue, name, routing_key: rk)
    end)
  end

  def setup_exchange(channel, queue, %{
        name: name,
        type: type,
        routing_keys: routing_keys
      })
      when type in [:direct, :topic] do
    Exchange.declare(channel, name, type)

    Enum.each(routing_keys, fn rk ->
      :ok = Queue.bind(channel, queue, name, routing_key: rk)
    end)
  end

  def setup_exchange(channel, queue, %{name: name, type: :fanout, opts: opts}) do
    Exchange.declare(channel, name, :fanout, opts)
    Queue.bind(channel, queue, name)
  end

  def setup_exchange(channel, queue, %{name: name, type: :fanout}) do
    Exchange.declare(channel, name, :fanout)
    Queue.bind(channel, queue, name)
  end

  def setup_exchange(_chan, _queue, conf) do
    raise "Unhandled exchange configuration #{inspect(conf)}"
  end

  def setup_exchange(channel, %{name: name, type: type, opts: opts}) do
    Exchange.declare(channel, name, type, opts)
  end

  def setup_exchange(channel, %{name: name, type: type}) do
    Exchange.declare(channel, name, type)
  end
end