klife

defmodule Klife do
  @moduledoc false

  alias Klife.Record
  alias Klife.Producer
  alias Klife.Consumer.Fetcher
  alias Klife.TxnProducerPool
  alias Klife.Connection.Controller, as: ConnController

  alias Klife.MetadataCache

  @produce_opts [
    producer: [
      type: :atom,
      required: false,
      doc:
        "Producer's name that will override the `default_producer` configuration. Ignored inside transactions."
    ],
    partitioner: [
      type: :atom,
      required: false,
      doc: "Module that will override `default_partitioner` configuration."
    ]
  ]

  @async_opts [
    callback: [
      type: :any,
      required: false,
      doc:
        "MFA or function/1 that will be called with the produce result. The result is injected as the first argument on MFA and is the only argument for anonymous functions"
    ]
  ]

  @txn_opts [
    pool_name: [
      type: :atom,
      required: false,
      doc: "Txn pool's name that will override the `default_txn_pool` configuration."
    ]
  ]

  @fetch_opts [
    fetcher: [
      type: :atom,
      required: false,
      doc: "Fetcher's name that will override the `default_fetcher` configuration."
    ],
    isolation_level: [
      type: {:in, [:read_committed, :read_uncommitted]},
      required: false,
      doc:
        "Controls whether the fetch response includes uncommitted transactional records. Defaults to `:read_committed`."
    ],
    max_bytes: [
      type: :non_neg_integer,
      required: false,
      doc:
        "Maximum number of bytes to return per fetch request. Defaults to `100_000` for `fetch/4` and `500_000` for `fetch_async/4`."
    ]
  ]

  @doc false
  def get_produce_opts(), do: @produce_opts
  @doc false
  def get_txn_opts(), do: @txn_opts
  @doc false
  def get_async_opts(), do: @async_opts
  @doc false
  def get_fetch_opts(), do: @fetch_opts

  @doc false
  def produce(%Record{} = record, client, opts \\ []) do
    [resp] = produce_batch([record], client, opts)
    resp
  end

  @doc false
  def produce_batch([%Record{} | _] = records, client, opts \\ []) do
    records = prepare_records(records, client, opts)

    if TxnProducerPool.in_txn?(client) do
      TxnProducerPool.produce(records, client, opts)
    else
      if ConnController.disabled_feature?(client, :producer) do
        raise "Produce API called but producer feature is disabled (client=#{inspect(client)}). Check logs for details."
      end

      Producer.produce(records, client, opts)
    end
  end

  @doc false
  def produce_async(%Record{} = record, client, opts \\ []) do
    if ConnController.disabled_feature?(client, :producer) do
      raise "Produce API called but producer feature is disabled (client=#{inspect(client)}). Check logs for details."
    end

    prepared_rec = prepare_records(record, client, opts)
    Producer.produce_async([prepared_rec], client, opts)
  end

  @doc false
  def produce_batch_async([%Record{} | _] = records, client, opts \\ []) do
    if ConnController.disabled_feature?(client, :producer) do
      raise "Produce API called but producer feature is disabled (client=#{inspect(client)}). Check logs for details."
    end

    case opts[:callback] do
      nil ->
        records = prepare_records(records, client, opts)
        Producer.produce_async(records, client, opts)

      {m, f, args} ->
        Task.start(fn -> apply(m, f, [produce_batch(records, client, opts) | args]) end)
        :ok

      fun when is_function(fun, 1) ->
        Task.start(fn -> fun.(produce_batch(records, client, opts)) end)
        :ok
    end
  end

  @doc false
  def produce_batch_txn([%Record{} | _] = records, client, opts \\ []) do
    transaction(
      fn -> records |> produce_batch(client, opts) |> Record.verify_batch() end,
      client,
      opts
    )
  end

  @doc false
  def transaction(fun, client, opts \\ []) do
    if ConnController.disabled_feature?(client, :txn_producer) do
      raise "Transaction API called but txn_producer feature is disabled (client=#{inspect(client)}). Check logs for details."
    end

    TxnProducerPool.run_txn(client, get_txn_pool(client, opts), fun)
  end

  def fetch(topic, partition, offset, client, opts \\ []) do
    tpo = {topic, partition, offset}
    Fetcher.fetch(tpo, client, opts)
  end

  def fetch(tpo_list, client, opts) when is_list(tpo_list) do
    Fetcher.fetch(tpo_list, client, opts)
  end

  @doc false
  def fetch_async(topic, partition, offset, client, opts \\ []) do
    tpo = {topic, partition, offset}
    Fetcher.fetch_async(tpo, client, opts)
  end

  defp get_txn_pool(client, opts) do
    Keyword.get(opts, :pool_name) || client.get_default_txn_pool()
  end

  def add_partition(%Record{} = record, client, opts \\ []) do
    case record do
      %Record{partition: nil, topic: topic} ->
        case MetadataCache.get_metadata(client, topic, 0) do
          {:ok,
           %{
             default_partitioner: default_partitioner_mod,
             max_partition: max_partition
           }} ->
            partitioner_mod = Keyword.get(opts, :partitioner, default_partitioner_mod)

            %{record | partition: partitioner_mod.get_partition(record, max_partition)}

          {:error, :not_found} ->
            {:error, :unkown_metadata_for_topic}
        end

      record ->
        record
    end
  end

  def add_partition!(%Record{} = record, client, opts \\ []) do
    case add_partition(record, client, opts) do
      %Record{} = rec -> rec
      err -> raise "Error on add partition: #{inspect(err)}"
    end
  end

  def log_metadata do
    [
      :client,
      :broker_id,
      :host,
      :topic,
      :partition,
      :offset,
      :group,
      :producer,
      :pool,
      :consumer_group_mod,
      :coordinator_id,
      :error_code,
      :error_message,
      :error,
      :reason,
      :message,
      :feature,
      :api_key,
      :requested_offset,
      :reset_offset,
      :latest_committed_offset,
      :latest_processed_offset,
      :stacktrace
    ]
  end

  defp prepare_records(%Record{} = rec, client, opts) do
    [new_rec] = prepare_records([rec], client, opts)
    new_rec
  end

  defp prepare_records(recs, client, opts) when is_list(recs) do
    recs
    |> Enum.with_index(1)
    |> Enum.map(fn {rec, idx} ->
      rec
      |> Map.replace!(:__estimated_size, Record.estimate_size(rec))
      |> Map.replace!(:__batch_index, idx)
      |> add_partition!(client, opts)
    end)
  end
end