tinkex

defmodule Tinkex.TrainingClient.DataProcessor do
  @moduledoc """
  Data chunking, numbering, and tensor operations for TrainingClient.

  This module handles:
  - Chunking training data based on size limits
  - Estimating chunk sizes using byte heuristics
  - Building placeholder gradients for custom loss
  - Extracting target tokens from loss function inputs
  """

  alias Tinkex.ByteEstimator
  alias Tinkex.Error
  alias Tinkex.Types.{Datum, TensorData}

  @max_chunk_len 1024
  @max_chunk_bytes_count 5_000_000

  @doc """
  Chunk data into manageable pieces based on size and byte limits.

  Ensures no chunk exceeds:
  - #{@max_chunk_len} items
  - #{@max_chunk_bytes_count} total estimated bytes
  """
  @spec chunk_data(list()) :: [list()]
  def chunk_data(data) do
    data
    |> Enum.chunk_while(
      {[], 0},
      fn datum, {chunk, byte_count} ->
        estimated = ByteEstimator.estimate_datum_bytes(datum)

        cond do
          length(chunk) >= @max_chunk_len ->
            {:cont, chunk, {[datum], estimated}}

          byte_count + estimated > @max_chunk_bytes_count ->
            {:cont, chunk, {[datum], estimated}}

          true ->
            {:cont, {chunk ++ [datum], byte_count + estimated}}
        end
      end,
      fn
        {[], 0} -> {:cont, []}
        {chunk, _count} -> {:cont, chunk, {[], 0}}
      end
    )
  end

  @doc """
  Allocate sequential request IDs for a batch of requests.

  Returns `{[id1, id2, ...], new_counter}` where the IDs are consecutive
  starting from the current counter.
  """
  @spec allocate_request_ids(non_neg_integer(), pos_integer()) ::
          {[pos_integer()], pos_integer()}
  def allocate_request_ids(count, counter) when count <= 0, do: {[], counter}

  def allocate_request_ids(count, counter) do
    ids = Enum.to_list(counter..(counter + count - 1))
    {ids, counter + count}
  end

  @doc """
  Build placeholder gradients (zeros) for custom loss computation.

  Creates zero-filled tensors matching the shape of target_tokens for each datum.
  These are used as placeholder gradients before the actual loss computation.
  """
  @spec build_placeholder_gradients(list(Datum.t())) ::
          {:ok, [Nx.Tensor.t()]} | {:error, Error.t()}
  def build_placeholder_gradients(data) do
    data
    |> Enum.reduce_while({:ok, []}, fn datum, {:ok, acc} ->
      case fetch_target_tokens_tensor(datum) do
        {:ok, target_tensor} ->
          zero =
            Nx.broadcast(
              Nx.tensor(0.0, type: {:f, 32}),
              Nx.shape(target_tensor)
            )

          {:cont, {:ok, [zero | acc]}}

        {:error, reason} ->
          {:halt, {:error, reason}}
      end
    end)
    |> case do
      {:ok, grads_rev} -> {:ok, Enum.reverse(grads_rev)}
      {:error, _} = error -> error
    end
  end

  @doc """
  Extract target_tokens tensor from a datum's loss_fn_inputs.

  Supports both TensorData and Nx.Tensor formats.
  """
  @spec fetch_target_tokens_tensor(Datum.t()) ::
          {:ok, Nx.Tensor.t()} | {:error, Error.t()}
  def fetch_target_tokens_tensor(%Datum{loss_fn_inputs: inputs}) do
    case inputs["target_tokens"] || inputs[:target_tokens] do
      %TensorData{} = td ->
        {:ok, TensorData.to_nx(td)}

      %Nx.Tensor{} = tensor ->
        {:ok, tensor}

      nil ->
        {:error, Error.new(:validation, "target_tokens missing from loss_fn_inputs")}

      other ->
        {:error,
         Error.new(
           :validation,
           "Invalid target_tokens in loss_fn_inputs: #{inspect(other)}"
         )}
    end
  end
end