aerospike_driver

defmodule Aerospike.Protocol.CDT do
  @moduledoc false

  # Shared MessagePack payload builders for list/map/bit/HLL CDT operations.

  alias Aerospike.Protocol.AsmMsg.Operation
  alias Aerospike.Protocol.MessagePack

  # CDT operation values are MessagePack in a bin; wire particle type is BLOB (4), matching
  # Go/Java clients (`ParticleType.BLOB` for packed CDT payloads).
  @particle_blob 4

  @context_eval 0xFF

  @doc """
  List or generic CDT read: `op_type` wire byte 3, payload MessagePack blob.
  """
  @spec list_read_op(String.t(), integer(), [term()], [Aerospike.Ctx.step()] | nil) ::
          Operation.t()
  def list_read_op(bin_name, op_code, args, ctx \\ nil)
      when is_binary(bin_name) and is_integer(op_code) and is_list(args) do
    %Operation{
      op_type: Operation.op_cdt_read(),
      particle_type: @particle_blob,
      bin_name: bin_name,
      data: encode_payload(op_code, args, ctx)
    }
  end

  @doc """
  List or generic CDT modify: `op_type` wire byte 4.
  """
  @spec list_modify_op(String.t(), integer(), [term()], [Aerospike.Ctx.step()] | nil) ::
          Operation.t()
  def list_modify_op(bin_name, op_code, args, ctx \\ nil)
      when is_binary(bin_name) and is_integer(op_code) and is_list(args) do
    %Operation{
      op_type: Operation.op_cdt_modify(),
      particle_type: @particle_blob,
      bin_name: bin_name,
      data: encode_payload(op_code, args, ctx)
    }
  end

  @doc """
  Map CDT read — same wire op bytes as list CDT (3), different opcode namespace (64+).
  """
  @spec map_read_op(String.t(), integer(), [term()], [Aerospike.Ctx.step()] | nil) ::
          Operation.t()
  def map_read_op(bin_name, op_code, args, ctx \\ nil)
      when is_binary(bin_name) and is_integer(op_code) and is_list(args) do
    # Go uses `_MAP_READ` for operate flag routing; wire `.op` is still 3.
    %{list_read_op(bin_name, op_code, args, ctx) | map_cdt: true}
  end

  @doc """
  Map CDT modify — wire op byte 4.
  """
  @spec map_modify_op(String.t(), integer(), [term()], [Aerospike.Ctx.step()] | nil) ::
          Operation.t()
  def map_modify_op(bin_name, op_code, args, ctx \\ nil)
      when is_binary(bin_name) and is_integer(op_code) and is_list(args) do
    %{list_modify_op(bin_name, op_code, args, ctx) | map_cdt: true}
  end

  @doc "Bit read (wire op 12)."
  @spec bit_read_op(String.t(), integer(), [term()], [Aerospike.Ctx.step()] | nil) ::
          Operation.t()
  def bit_read_op(bin_name, op_code, args, ctx \\ nil)
      when is_binary(bin_name) and is_integer(op_code) and is_list(args) do
    %Operation{
      op_type: Operation.op_bit_read(),
      particle_type: @particle_blob,
      bin_name: bin_name,
      data: encode_payload(op_code, args, ctx)
    }
  end

  @doc "Bit modify (wire op 13)."
  @spec bit_modify_op(String.t(), integer(), [term()], [Aerospike.Ctx.step()] | nil) ::
          Operation.t()
  def bit_modify_op(bin_name, op_code, args, ctx \\ nil)
      when is_binary(bin_name) and is_integer(op_code) and is_list(args) do
    %Operation{
      op_type: Operation.op_bit_modify(),
      particle_type: @particle_blob,
      bin_name: bin_name,
      data: encode_payload(op_code, args, ctx)
    }
  end

  @doc "HLL read (wire op 15); server does not support nested HLL context."
  @spec hll_read_op(String.t(), integer(), [term()]) :: Operation.t()
  def hll_read_op(bin_name, op_code, args)
      when is_binary(bin_name) and is_integer(op_code) and is_list(args) do
    %Operation{
      op_type: Operation.op_hll_read(),
      particle_type: @particle_blob,
      bin_name: bin_name,
      data: encode_simple_array(op_code, args)
    }
  end

  @doc "HLL modify (wire op 16)."
  @spec hll_modify_op(String.t(), integer(), [term()]) :: Operation.t()
  def hll_modify_op(bin_name, op_code, args)
      when is_binary(bin_name) and is_integer(op_code) and is_list(args) do
    %Operation{
      op_type: Operation.op_hll_modify(),
      particle_type: @particle_blob,
      bin_name: bin_name,
      data: encode_simple_array(op_code, args)
    }
  end

  @doc false
  @spec cdt_pack_arg(term()) :: term()
  def cdt_pack_arg({:bytes, _} = b), do: b

  # UTF-8 string arguments must use Aerospike's particle-wrapped encoding inside
  # CDT blobs (not plain MessagePack strings). See `MessagePack` `:particle_string`.
  def cdt_pack_arg(s) when is_binary(s), do: {:particle_string, s}

  def cdt_pack_arg(list) when is_list(list) do
    Enum.map(list, &cdt_pack_arg/1)
  end

  def cdt_pack_arg(map) when is_map(map) do
    Map.new(map, fn {k, v} -> {cdt_pack_arg(k), cdt_pack_arg(v)} end)
  end

  def cdt_pack_arg(other), do: other

  defp encode_simple_array(op_code, args) do
    MessagePack.pack!([op_code | Enum.map(args, &cdt_pack_arg/1)])
  end

  defp encode_payload(op_code, args, ctx) do
    inner = [op_code | Enum.map(args, &cdt_pack_arg/1)]

    case ctx do
      nil ->
        MessagePack.pack!(inner)

      [] ->
        MessagePack.pack!(inner)

      steps when is_list(steps) ->
        ctx_elems =
          Enum.flat_map(steps, fn {id, val} ->
            [id, cdt_pack_arg(val)]
          end)

        MessagePack.pack!([@context_eval, ctx_elems, inner])
    end
  end
end