eventodb_kit

defmodule EventodbKit.Consumer.Position do
  @moduledoc """
  Position tracking for consumers.
  """

  import Ecto.Query
  alias EventodbKit.Schema.ConsumerPosition

  @doc """
  Loads the current position for a consumer.
  Returns 0 if no position is found.
  """
  def load(repo, namespace, category, consumer_id) do
    query =
      from(p in ConsumerPosition,
        where:
          p.namespace == ^namespace and
            p.category == ^category and
            p.consumer_id == ^consumer_id
      )

    case repo.one(query) do
      nil -> 0
      position -> position.position
    end
  end

  @doc """
  Saves the current position for a consumer.
  """
  def save(repo, namespace, category, consumer_id, position) do
    attrs = %{
      namespace: namespace,
      category: category,
      consumer_id: consumer_id,
      position: position,
      updated_at: DateTime.utc_now()
    }

    %ConsumerPosition{}
    |> ConsumerPosition.changeset(attrs)
    |> repo.insert(
      on_conflict: {:replace, [:position, :updated_at]},
      conflict_target: [:namespace, :category, :consumer_id]
    )
  end
end