s2_client

defmodule S2.S2S.CheckTail do
  @moduledoc """
  Check tail position for a stream.

  Uses Mint HTTP/2 directly for consistency with the rest of the data plane.
  No `Content-Type` header is sent because the server always returns JSON for
  this endpoint regardless of the requested content type — sending `s2s/proto`
  has no effect on the response format.
  """

  require Logger

  alias S2.S2S.Shared

  @doc """
  Check the tail position of a stream.

  Returns the current sequence number and timestamp of the stream's tail —
  the position where the next appended record would be written.

  ## Options

    * `:token` — Bearer token for authentication.
    * `:recv_timeout` — Timeout in milliseconds for receiving the response (default: 5000).

  Returns `{:ok, %S2.V1.StreamPosition{}, conn}` or `{:error, reason, conn}`.
  """
  @spec call(Mint.HTTP2.t(), String.t(), String.t(), keyword()) ::
          {:ok, S2.V1.StreamPosition.t(), Mint.HTTP2.t()}
          | {:error, term(), Mint.HTTP2.t()}
  def call(conn, basin, stream, opts \\ []) do
    Logger.debug("S2S.CheckTail basin=#{basin} stream=#{stream}")
    path = Shared.records_path(stream) <> "/tail"
    token = Keyword.get(opts, :token)
    recv_timeout = Keyword.get(opts, :recv_timeout, Shared.default_timeout())
    headers = Shared.build_headers(basin, token, content_type: false)

    case Mint.HTTP2.request(conn, "GET", path, headers, nil) do
      {:ok, conn, request_ref} ->
        case Shared.receive_complete(conn, request_ref, timeout: recv_timeout) do
          {:ok, %{status: 200, data: data}, conn} ->
            parse_tail_response(data, conn)

          {:ok, %{status: status, data: data}, conn} ->
            {:error, Shared.parse_http_error(status, data), conn}

          {:error, reason, conn} ->
            {:error, reason, conn}
        end

      {:error, conn, reason} ->
        {:error, reason, conn}
    end
  end

  @doc false
  def parse_tail_response(data, conn) do
    case Jason.decode(data) do
      {:ok, %{"tail" => %{"seq_num" => seq_num, "timestamp" => timestamp}}}
      when is_integer(seq_num) and seq_num >= 0 and is_integer(timestamp) and timestamp >= 0 ->
        position = %S2.V1.StreamPosition{
          seq_num: seq_num,
          timestamp: timestamp
        }

        {:ok, position, conn}

      # Server may omit timestamp for empty streams. Default to 0 (not provided).
      {:ok, %{"tail" => %{"seq_num" => seq_num}}} when is_integer(seq_num) and seq_num >= 0 ->
        position = %S2.V1.StreamPosition{
          seq_num: seq_num,
          timestamp: 0
        }

        {:ok, position, conn}

      {:ok, %{"tail" => tail}} when is_map(tail) ->
        {:error, %S2.Error{message: "invalid tail fields: #{inspect(tail)}"}, conn}

      {:ok, body} ->
        {:error, %S2.Error{message: "missing tail key in response: #{inspect(body)}"}, conn}

      {:error, reason} ->
        {:error, %S2.Error{message: "failed to decode tail response: #{inspect(reason)}"}, conn}
    end
  end
end