nebula_graph_ex

defmodule NebulaGraphEx.Transport do
  @moduledoc """
  Low-level TCP/SSL socket abstraction with Thrift framed transport.

  NebulaGraph uses 4-byte big-endian length-prefixed framing over plain TCP
  (or TLS). This module owns the socket lifecycle and provides `send_frame/2`
  and `recv_frame/2` which handle the framing transparently.

  Callers (i.e. `NebulaGraphEx.Connection`) should not use raw socket calls —
  always go through this module so framing and SSL are handled uniformly.
  """

  @type socket :: :gen_tcp.socket() | :ssl.sslsocket()
  @type opts :: keyword()

  # Frame length header is a 4-byte unsigned big-endian integer.
  @frame_header_bytes 4

  @doc """
  Opens a TCP (or TLS) connection to `host:port`.

  ## Options

  * `:ssl` — `boolean`, enable TLS. Default: `false`.
  * `:ssl_opts` — keyword list passed to `:ssl.connect/4`. Default: `[]`.
  * `:connect_timeout` — timeout in ms. Default: `5_000`.
  * `:send_timeout` — socket-level send timeout. Default: `15_000`.

  Returns `{:ok, socket}` or `{:error, reason}`.
  """
  @spec connect(String.t(), :inet.port_number(), opts()) ::
          {:ok, socket()} | {:error, term()}
  def connect(host, port, opts \\ []) do
    ssl? = Keyword.get(opts, :ssl, false)
    connect_timeout = Keyword.get(opts, :connect_timeout, 5_000)
    send_timeout = Keyword.get(opts, :send_timeout, 15_000)
    host_charlist = String.to_charlist(host)

    tcp_opts = [
      :binary,
      active: false,
      packet: :raw,
      nodelay: true,
      send_timeout: send_timeout,
      send_timeout_close: true
    ]

    if ssl? do
      ssl_opts = Keyword.get(opts, :ssl_opts, [])

      merged_ssl_opts =
        Keyword.merge(
          [verify: :verify_peer, depth: 3],
          ssl_opts
        )

      :ssl.connect(host_charlist, port, tcp_opts ++ merged_ssl_opts, connect_timeout)
    else
      :gen_tcp.connect(host_charlist, port, tcp_opts, connect_timeout)
    end
  end

  @doc """
  Sends a binary payload as a single Thrift frame.

  Prepends the 4-byte big-endian length header automatically.
  """
  @spec send_frame(socket(), binary()) :: :ok | {:error, term()}
  def send_frame(socket, payload) when is_binary(payload) do
    frame = <<byte_size(payload)::unsigned-big-integer-size(32)>> <> payload
    raw_send(socket, frame)
  end

  @doc """
  Receives one complete Thrift frame from the socket.

  Reads the 4-byte length header first, then reads exactly that many bytes.
  Returns `{:ok, payload}` or `{:error, reason}`.
  """
  @spec recv_frame(socket(), timeout()) :: {:ok, binary()} | {:error, term()}
  def recv_frame(socket, timeout \\ 15_000) do
    with {:ok, <<len::unsigned-big-integer-size(32)>>} <-
           raw_recv(socket, @frame_header_bytes, timeout),
         {:ok, payload} <- raw_recv(socket, len, timeout) do
      {:ok, payload}
    end
  end

  @doc """
  Closes the socket. Works for both plain TCP and SSL sockets.
  """
  @spec close(socket()) :: :ok
  def close({:sslsocket, _, _} = socket), do: :ssl.close(socket)
  def close(socket), do: :gen_tcp.close(socket)

  @doc """
  Returns `true` if the socket is an SSL socket.
  """
  @spec ssl?(socket()) :: boolean()
  def ssl?({:sslsocket, _, _}), do: true
  def ssl?(_), do: false

  # ─── Private helpers ──────────────────────────────────────────────────────

  defp raw_send({:sslsocket, _, _} = socket, data), do: :ssl.send(socket, data)
  defp raw_send(socket, data), do: :gen_tcp.send(socket, data)

  defp raw_recv({:sslsocket, _, _} = socket, length, timeout),
    do: :ssl.recv(socket, length, timeout)

  defp raw_recv(socket, length, timeout),
    do: :gen_tcp.recv(socket, length, timeout)
end