nebula_graph_ex

defmodule NebulaGraphEx.Thrift.Client do
  @moduledoc """
  Low-level Thrift binary protocol request/response over a framed socket.

  This module encodes Thrift `CALL` messages and decodes `REPLY` messages
  using the binary protocol. It is intentionally ignorant of business logic —
  `NebulaGraphEx.Thrift.GraphService` builds on this to provide typed calls.

  ## Thrift binary message layout (inside the frame)

      << 0x80, 0x01,        :: 2 bytes  — version + message type (CALL = 1)
         0x00, 0x00,        :: 2 bytes  — unused (version high bits)
         name_len::32-big,  :: 4 bytes  — method name length
         name::binary,      :: N bytes  — method name
         seq_id::32-big,    :: 4 bytes  — sequence number
         ...struct fields   :: variable — Thrift struct encoding
         0x00 >>            :: 1 byte   — STOP field

  Replies mirror this layout but with message type `REPLY` (2) or
  `EXCEPTION` (3).
  """

  alias NebulaGraphEx.Transport

  # Thrift message types
  @msg_call 1
  @msg_reply 2
  @msg_exception 3

  # Thrift field types
  @type_stop 0
  @type_bool 2
  @type_byte 3
  @type_i16 6
  @type_i32 8
  @type_i64 10
  @type_double 4
  @type_binary 11
  @type_struct 12
  @type_map 13
  @type_set 14
  @type_list 15

  @doc """
  Encodes and sends a Thrift `CALL` message, then receives and decodes the
  `REPLY`.

  * `socket` — a socket returned by `NebulaGraphEx.Transport.connect/3`
  * `method` — binary method name, e.g. `"execute"`
  * `args_iodata` — already-encoded struct fields (from `encode_struct/1`)
  * `seq_id` — sequence number (any integer; 1 is fine for sequential use)
  * `recv_timeout` — milliseconds to wait for the reply frame

  Returns `{:ok, reply_payload}` where `reply_payload` is the raw bytes of
  the reply message body (after the message header), or `{:error, reason}`.
  """
  @spec call(Transport.socket(), binary(), iodata(), integer(), timeout()) ::
          {:ok, binary()} | {:error, term()}
  def call(socket, method, args_iodata, seq_id \\ 1, recv_timeout \\ 15_000)
      when is_binary(method) do
    payload = encode_message(method, @msg_call, seq_id, args_iodata)

    with :ok <- Transport.send_frame(socket, IO.iodata_to_binary(payload)),
         {:ok, frame} <- Transport.recv_frame(socket, recv_timeout) do
      decode_reply(frame)
    end
  end

  @doc """
  Sends a one-way Thrift `CALL` (no reply expected, e.g. `signout`).
  """
  @spec cast(Transport.socket(), binary(), iodata(), integer()) :: :ok | {:error, term()}
  def cast(socket, method, args_iodata, seq_id \\ 1) do
    payload = encode_message(method, @msg_call, seq_id, args_iodata)
    Transport.send_frame(socket, IO.iodata_to_binary(payload))
  end

  # ─── Encoding helpers ────────────────────────────────────────────────────────

  @doc "Encodes a Thrift binary message header + body."
  def encode_message(method, msg_type, seq_id, body) do
    name_bytes = byte_size(method)

    [
      # Version 1 + message type
      <<0x80, 0x01, 0x00::8, msg_type::8>>,
      <<name_bytes::32-big>>,
      method,
      <<seq_id::32-big>>,
      body,
      <<@type_stop>>
    ]
  end

  @doc "Encodes a field header (type byte + field id)."
  def field(type, id), do: <<type::8, id::16-big>>

  @doc "Encodes the STOP byte that terminates a struct."
  def stop, do: <<@type_stop>>

  @doc "Encodes an `i64`."
  def i64(v), do: <<v::64-big-signed>>

  @doc "Encodes an `i32`."
  def i32(v), do: <<v::32-big-signed>>

  @doc "Encodes an `i16`."
  def i16(v), do: <<v::16-big-signed>>

  @doc "Encodes a `byte`."
  def byte_val(v), do: <<v::8>>

  @doc "Encodes a `bool`."
  def bool(true), do: <<1::8>>
  def bool(false), do: <<0::8>>

  @doc "Encodes a `double`."
  def double(v), do: <<v::64-big-float>>

  @doc "Encodes a length-prefixed `binary` / `string`."
  def binary_val(v) when is_binary(v), do: [<<byte_size(v)::32-big>>, v]

  @doc "Encodes a map field (ktype, vtype, count, then pairs)."
  def map(ktype, vtype, items) do
    count = length(items)
    [<<ktype::8, vtype::8, count::32-big>> | items]
  end

  @doc "Encodes a list field (etype, count, then elements)."
  def list(etype, items) do
    count = length(items)
    [<<etype::8, count::32-big>> | items]
  end

  # ─── Decoding helpers ─────────────────────────────────────────────────────

  @doc false
  def decode_reply(<<0x80, 0x01, 0x00, msg_type::8, rest::binary>>) do
    with {:ok, _name, rest2} <- decode_string(rest),
         <<_seq_id::32-big, body::binary>> <- rest2 do
      case msg_type do
        @msg_reply -> {:ok, body}
        @msg_exception -> {:error, {:thrift_exception, body}}
        other -> {:error, {:unexpected_message_type, other}}
      end
    end
  end

  def decode_reply(data) do
    {:error, {:bad_thrift_message, data}}
  end

  @doc "Skips field headers and returns `{field_id, type, rest}`."
  def next_field(<<type::8, field_id::16-big, rest::binary>>) when type != @type_stop do
    {:ok, field_id, type, rest}
  end

  def next_field(<<@type_stop, rest::binary>>), do: {:stop, rest}
  def next_field(<<>>), do: {:stop, <<>>}

  @doc "Decodes a length-prefixed binary/string."
  def decode_string(<<len::32-big, value::binary-size(len), rest::binary>>),
    do: {:ok, value, rest}

  def decode_string(data), do: {:error, {:truncated_string, data}}

  @doc "Decodes an i64."
  def decode_i64(<<v::64-big-signed, rest::binary>>), do: {:ok, v, rest}
  def decode_i64(data), do: {:error, {:truncated_i64, data}}

  @doc "Decodes an i32."
  def decode_i32(<<v::32-big-signed, rest::binary>>), do: {:ok, v, rest}
  def decode_i32(data), do: {:error, {:truncated_i32, data}}

  @doc "Decodes an i16."
  def decode_i16(<<v::16-big-signed, rest::binary>>), do: {:ok, v, rest}
  def decode_i16(data), do: {:error, {:truncated_i16, data}}

  @doc "Decodes a byte."
  def decode_byte(<<v::8, rest::binary>>), do: {:ok, v, rest}
  def decode_byte(data), do: {:error, {:truncated_byte, data}}

  @doc "Decodes a bool."
  def decode_bool(<<1::8, rest::binary>>), do: {:ok, true, rest}
  def decode_bool(<<0::8, rest::binary>>), do: {:ok, false, rest}
  def decode_bool(data), do: {:error, {:truncated_bool, data}}

  @doc "Decodes a double."
  def decode_double(<<v::64-big-float, rest::binary>>), do: {:ok, v, rest}
  def decode_double(data), do: {:error, {:truncated_double, data}}

  # ─── Type constants (exported for sub-modules) ────────────────────────────

  def type_stop, do: @type_stop
  def type_bool, do: @type_bool
  def type_byte, do: @type_byte
  def type_i16, do: @type_i16
  def type_i32, do: @type_i32
  def type_i64, do: @type_i64
  def type_double, do: @type_double
  def type_binary, do: @type_binary
  def type_struct, do: @type_struct
  def type_map, do: @type_map
  def type_set, do: @type_set
  def type_list, do: @type_list
  def msg_call, do: @msg_call
  def msg_reply, do: @msg_reply
  def msg_exception, do: @msg_exception
end