kaffe

defmodule Kaffe.GroupManager do
  @moduledoc """
  This is the main process for bootstrapping the full supervision tree to
  consume a Kafka topic via a subscriber/worker combo per topic per partition
  as part of a consumer group.

  See `Kaffe.GroupMemberSupervisor` for distinct components.

  The process begins by starting the client connection to Kafka. Then group
  members are created for each of the configured topics.

  Note that this module is is not involved in any message processing, rather
  it's role is to ensure that all of the relevant services are running.
  """

  use GenServer
  use Retry
  require Logger

  defmodule State do
    @moduledoc false

    defstruct supervisor_pid: nil,
              subscriber_name: nil,
              config: nil,
              consumer_group: nil,
              topics: nil,
              offset: nil,
              worker_manager_pid: nil
  end

  defmodule ClientDownException do
    defexception [:message]

    def exception(_term) do
      %ClientDownException{message: "Kafka client is down"}
    end
  end

  ## ==========================================================================
  ## Public API
  ## ==========================================================================

  @doc false
  def start_link(config) do
    GenServer.start_link(__MODULE__, [self(), config], name: name(config))
  end

  @doc false
  def child_spec(opts), do: super(opts)

  @doc """
  Dynamically subscribe to topics in addition to the configured topics.
  Returns the newly subscribed topics. This may not include all values if any are already subscribed to.
  """
  def subscribe_to_topics({config, topics}) do
    GenServer.call(name(config), {:subscribe_to_topics, topics})
  end

  @doc """
  List of currently subscribed topics.
  """
  def list_subscribed_topics(config) do
    GenServer.call(name(config), {:list_subscribed_topics})
  end

  ## ==========================================================================
  ## Callbacks
  ## ==========================================================================

  @impl GenServer
  def init([supervisor_pid, config]) do
    Logger.info("event#startup=#{__MODULE__} name=#{name(config)}")

    case kafka().start_client(config.endpoints, config.subscriber_name, config.consumer_config) do
      :ok ->
        :ok

      {_, :already_present} ->
        Logger.info("The brod client is already present, continuing.")
    end

    GenServer.cast(self(), {:start_group_members})

    {:ok,
     %State{
       supervisor_pid: supervisor_pid,
       subscriber_name: config.subscriber_name,
       consumer_group: config.consumer_group,
       topics: config.topics,
       config: config
     }}
  end

  # Start the subscribers and workers to process message sets
  #
  # Worker are booted before the subscribers so when the subscribers receive the
  # first messages, we know there will be a worker to do the actual processing work
  @impl GenServer
  def handle_cast({:start_group_members}, state) do
    Logger.debug("Starting worker supervisors for group manager: #{inspect(self())}")

    {:ok, worker_supervisor_pid} =
      group_member_supervisor().start_worker_supervisor(state.supervisor_pid, state.subscriber_name)

    {:ok, worker_manager_pid} =
      worker_supervisor().start_worker_manager(worker_supervisor_pid, state.subscriber_name, state.config)

    state = %State{state | worker_manager_pid: worker_manager_pid}

    :ok = subscribe_to_topics(state, state.topics)

    {:noreply, state}
  end

  # Subscribe to a new set of topics. The new list of subscribed topics will only include
  # the requested topics and none of the currently configured topics.
  @impl GenServer
  def handle_call({:subscribe_to_topics, requested_topics}, _from, %State{topics: topics} = state) do
    new_topics = requested_topics -- topics
    :ok = subscribe_to_topics(state, new_topics)

    {:reply, {:ok, new_topics}, %State{state | topics: state.topics ++ new_topics}}
  end

  # List the currently subscribed topics
  def handle_call({:list_subscribed_topics}, _from, %State{topics: topics} = state) do
    {:reply, topics, state}
  end

  ## ==========================================================================
  ## Helpers
  ## ==========================================================================

  defp subscribe_to_topics(state, topics) do
    Logger.debug("Starting group members for the following topics: #{inspect(topics)}")

    retry with: exponential_backoff() |> expiry(state.config.client_down_retry_expire),
          rescue_only: [Kaffe.GroupManager.ClientDownException] do
      Enum.each(topics, fn topic ->
        case subscribe_to_topic(state, topic) do
          {:ok, _pid} ->
            Logger.debug("Started group member for topic: #{topic}")
            :ok

          error ->
            Logger.debug("Starting group member for #{topic} failed, attempting retry with exponential backoff")

            is_client_down_error?(error)
            |> do_a_retry?(error)
        end
      end)
    after
      :ok ->
        Logger.debug("Group members succesfully started")
    else
      {:error, reason} = error ->
        Logger.error("Starting group members failed: #{inspect(reason)}")
        error

      _ = error ->
        Logger.error("Starting group members failed: #{inspect(error)}")
        {:error, error}
    end
  end

  defp do_a_retry?(true, _error), do: raise(Kaffe.GroupManager.ClientDownException)
  defp do_a_retry?(false, error), do: raise(error)

  defp subscribe_to_topic(state, topic) do
    group_member_supervisor().start_group_member(
      state.supervisor_pid,
      state.subscriber_name,
      state.consumer_group,
      state.worker_manager_pid,
      topic,
      state.config
    )
  end

  defp kafka do
    Application.get_env(:kaffe, :kafka_mod, :brod)
  end

  defp name(config) do
    :"#{__MODULE__}.#{config.subscriber_name}"
  end

  defp group_member_supervisor do
    Application.get_env(:kaffe, :group_member_supervisor_mod, Kaffe.GroupMemberSupervisor)
  end

  defp worker_supervisor do
    Application.get_env(:kaffe, :worker_supervisor_mod, Kaffe.WorkerSupervisor)
  end

  # Brod client errors are erlang exceptions and are hard to pattern match correctly.
  # This function casts the error to a string, and returns if the error is a client down error
  defp is_client_down_error?({:error, error}) do
    error_string = "#{inspect(error)}"

    cond do
      String.match?(error_string, ~r(:econnrefused)) ->
        true

      String.match?(error_string, ~r({:error, :client_down})) ->
        true

      true ->
        false
    end
  end
end