pulsar_elixir

defmodule Pulsar.Message do
  @moduledoc """
  Represents a message received from a Pulsar topic.

  This struct encapsulates all information about a message delivered to a consumer callback.

  ## Fields

  - `command` - For non-chunked messages: single command struct. For chunked messages: list of
    commands from all chunks.
    Type: `struct() | [struct()]`

  - `metadata` - For non-chunked messages: single metadata struct. For chunked messages: list of
    metadata from all chunks.
    Type: `struct() | [struct()]`

  - `payload` - The actual message payload as a binary. For chunked messages, this is the
    assembled complete payload.

  - `single_metadata` - For non-batch messages: nil. For batched messages: single message metadata.
    For chunked messages: list of metadata from all chunks.
    Type: `nil | struct() | [struct()]`

  - `broker_metadata` - For non-chunked messages: single broker metadata. For chunked messages:
    list of broker metadata from all chunks.
    Type: `term() | [term()]`

  - `message_id_to_ack` - For non-chunked messages: single message ID. For batch messages: message
    ID with batch_index. For chunked messages: list of all chunk message IDs.
    Type: `term() | [term()]`

  - `chunk_metadata` - Metadata about chunked messages (nil for non-chunked messages).
    For complete chunked messages: `%{chunked: true, complete: true, uuid: "...", num_chunks: N}`
    For incomplete chunked messages: `%{chunked: true, complete: false, error: :reason, uuid: "..."}`

  ## Usage

  Messages are received in the `handle_message/2` callback:

      def handle_message(%Pulsar.Message{} = message, state) do
        # Access fields directly
        payload = message.payload

        {:ok, state}
      end

  ## Pattern Matching Examples

      # Match only the payload
      def handle_message(%Pulsar.Message{payload: payload}, state) do
        process(payload)
        {:ok, state}
      end

      # Access all fields via the struct (non-chunked)
      def handle_message(%Pulsar.Message{} = msg, state) do
        redelivery_count = Pulsar.Message.redelivery_count(msg)
        producer = msg.metadata.producer_name
        {:ok, state}
      end

      # Manual acknowledgment using message_id_to_ack
      def handle_message(%Pulsar.Message{message_id_to_ack: ack_id} = msg, state) do
        spawn(fn ->
          case process_async(msg) do
            :ok -> Pulsar.Consumer.ack(self(), ack_id)
            {:error, _} -> Pulsar.Consumer.nack(self(), ack_id)
          end
        end)
        {:noreply, state}
      end
  """

  @type t :: %__MODULE__{
          command: struct() | [struct()],
          metadata: struct() | [struct()],
          payload: binary(),
          single_metadata: struct() | nil | [struct()],
          broker_metadata: term() | [term()],
          message_id_to_ack: term() | [term()],
          chunk_metadata: map() | nil
        }

  defstruct [
    :command,
    :metadata,
    :payload,
    :single_metadata,
    :broker_metadata,
    :message_id_to_ack,
    :chunk_metadata
  ]

  @doc """
  Returns the maximum redelivery count across all commands.

  For chunked messages, returns the maximum redelivery count from all chunks.
  For non-chunked messages, returns the redelivery count from the single command.

  ## Examples

      iex> Pulsar.Message.redelivery_count(message)
      3
  """
  @spec redelivery_count(t()) :: non_neg_integer()
  def redelivery_count(%__MODULE__{command: command}) when is_list(command) do
    command
    |> Enum.map(& &1.redelivery_count)
    |> Enum.max(fn -> 0 end)
  end

  def redelivery_count(%__MODULE__{command: command}) do
    command.redelivery_count
  end

  @doc """
  Returns the number of broker messages (permits) consumed.

  For non-chunked messages, this is always 1.
  For chunked messages, this is the number of chunks actually received.

  This is used for flow control permit accounting.

  ## Examples

      iex> Pulsar.Message.num_broker_messages(non_chunked_message)
      1

      iex> Pulsar.Message.num_broker_messages(complete_chunked_message)
      3  # if message had 3 chunks

      iex> Pulsar.Message.num_broker_messages(incomplete_chunked_message)
      2  # if only 2 out of 3 chunks were received before timeout
  """
  @spec num_broker_messages(t()) :: pos_integer()
  def num_broker_messages(%__MODULE__{chunk_metadata: %{message_ids: ids}}) when is_list(ids) do
    length(ids)
  end

  def num_broker_messages(%__MODULE__{}), do: 1

  @doc """
  Returns `true` if the message is a chunked message, `false` otherwise.

  This checks for the presence of chunk metadata.

  ## Examples

      iex> Pulsar.Message.chunked?(message)
      true

      iex> Pulsar.Message.chunked?(non_chunked_message)
      false
  """
  @spec chunked?(t()) :: boolean()
  def chunked?(%__MODULE__{chunk_metadata: %{chunked: true}}), do: true
  def chunked?(%__MODULE__{}), do: false

  @doc """
  Returns `true` if the chunked message is complete, `false` otherwise.

  For non-chunked messages, always returns `true` since they are inherently complete.
  For chunked messages, returns `true` only if all chunks were successfully received.

  ## Examples

      iex> Pulsar.Message.complete?(complete_chunked_message)
      true

      iex> Pulsar.Message.complete?(incomplete_chunked_message)
      false

      iex> Pulsar.Message.complete?(non_chunked_message)
      true
  """
  @spec complete?(t()) :: boolean()
  def complete?(%__MODULE__{chunk_metadata: %{complete: complete}}), do: complete
  def complete?(%__MODULE__{}), do: true
end