defmodule Elsa do
  @moduledoc """
  Provides public api to Elsa. Top-level short-cuts to sub-module functions
  for performing basic interactions with Kafka including listing, creating,
  deleting, and validating topics. Also provides a function for one-off
  produce_sync of message(s) to a topic.
  """

  @typedoc "named connection, must be an atom"
  @type connection :: atom
  @type hostname :: atom | String.t()
  @type portnum :: pos_integer
  @typedoc "endpoints to connect to kafka brokers"
  @type endpoints :: [{hostname, portnum}]
  @type topic :: String.t()
  @type partition :: non_neg_integer

  defdelegate list_topics(endpoints), to: Elsa.Topic, as: :list

  defdelegate topic?(endpoints, topic), to: Elsa.Topic, as: :exists?

  defdelegate create_topic(endpoints, topic, opts \\ []), to: Elsa.Topic, as: :create

  defdelegate delete_topic(endpoints, topic), to: Elsa.Topic, as: :delete

  defdelegate produce(endpoints_or_connection, topic, messages, opts \\ []), to: Elsa.Producer

  defdelegate fetch(endpoints, topic, opts \\ []), to: Elsa.Fetch

  @doc """
  Define a default client name for establishing persistent connections to
  the Kafka cluster by producers and consumers. Useful for optimizing
  interactions by passing the identifier of a standing connection instead
  of instantiating a new one at each interaction, but when only a single connection
  is required, aleviating the need for the caller to differentiate and pass
  around a name.
  """
  @spec default_client() :: atom()
  def default_client, do: :elsa_default_client

  defmodule Message do
    @moduledoc """
    Defines the structure of a Kafka message provided by the Elsa library and
    the function to construct the message struct.
    """
    import Record, only: [defrecord: 2, extract: 2]

    defrecord :kafka_message, extract(:kafka_message, from_lib: "kafka_protocol/include/kpro_public.hrl")

    @type kafka_message :: record(:kafka_message, key: term(), value: term(), offset: integer(), ts: integer())
    @type elsa_message :: %Elsa.Message{
            topic: Elsa.topic(),
            partition: Elsa.partition(),
            offset: integer,
            key: term,
            value: term,
            generation_id: integer | nil,
            headers: list
          }

    defstruct [
      :topic,
      :partition,
      :offset,
      :key,
      :value,
      :timestamp,
      :generation_id,
      :headers
    ]

    @doc """
    Constructs a message struct from the imported definition of a kafka_message as
    defined by the brod library with the addition of the topic and partition the message
    was read from as well as the optional generation id as defined by the message's relationship
    to a consumer group. Generation id defaults to `nil` in the event the message retrieved
    outside of the context of a consumer group.
    """

    @spec new(kafka_message(), keyword()) :: elsa_message()
    def new(kafka_message(offset: offset, key: key, value: value, ts: timestamp, headers: headers), attributes) do
      %Message{
        topic: Keyword.fetch!(attributes, :topic),
        partition: Keyword.fetch!(attributes, :partition),
        offset: offset,
        key: key,
        value: value,
        timestamp: timestamp,
        generation_id: Keyword.get(attributes, :generation_id),
        headers: headers
      }
    end
  end

  defmodule ConnectError do
    defexception [:message]
  end
end