mississippi

# Copyright 2024 SECO Mind Srl
# SPDX-License-Identifier: Apache-2.0

defmodule Mississippi.Consumer do
  @moduledoc """
  This module defines the supervision tree of Mississippi.Consumer.
  """

  use Supervisor

  alias Mississippi.Consumer.ConsumersSupervisor
  alias Mississippi.Consumer.Options

  require Logger

  @type init_options() :: [unquote(NimbleOptions.option_typespec(Options.definition()))]

  def start_link(init_opts) do
    Supervisor.start_link(__MODULE__, init_opts, name: __MODULE__)
  end

  @impl true
  def init(init_opts) do
    opts = NimbleOptions.validate!(init_opts, Options.definition())

    amqp_consumer_options = opts[:amqp_consumer_options]

    queue_config = opts[:mississippi_config][:queues]

    message_handler = opts[:mississippi_config][:message_handler]

    channels_per_connection = amqp_consumer_options[:channels]

    queue_count = queue_config[:range_end] - queue_config[:range_start] + 1

    # Invariant: we use one channel for one queue.
    connection_number = Kernel.ceil(queue_count / channels_per_connection)

    _ =
      Logger.debug("Have #{queue_count} queues and #{channels_per_connection} channels per connection")

    _ =
      Logger.debug(
        "Have #{connection_number} connections and a total of #{connection_number * channels_per_connection} channels"
      )

    children = [
      {ExRabbitPool.PoolSupervisor,
       rabbitmq_config: amqp_consumer_options, connection_pools: [events_consumer_pool_config(connection_number)]},
      {ConsumersSupervisor, queues: queue_config, message_handler: message_handler}
    ]

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

  defp events_consumer_pool_config(connection_number) do
    [
      name: {:local, :events_consumer_pool},
      worker_module: ExRabbitPool.Worker.RabbitConnection,
      size: connection_number,
      max_overflow: 0
    ]
  end
end