ccxt_client

defmodule CCXT.WS do
  @moduledoc """
  WebSocket entry point. Thin wrapper around `ZenWebsocket.Client` that binds
  a `%CCXT.Exchange{}` to a connection so `subscribe/3` can pick the correct
  exchange-native frame builder.

  ## Layer 1+2 (Task 92)

  Pure URL resolution (`CCXT.WS.URLRouting`) + connection lifecycle (this module).
  Layer 3 (auth state machine, custom reconnection) is deliberately deferred —
  `zen_websocket` covers reconnection, backoff, heartbeat, and subscription
  restoration natively.

  ## Usage

      {:ok, ws} = CCXT.WS.connect(exchange, :public)
      :ok = CCXT.WS.subscribe(ws, ["tickers.BTCUSDT"])
      # Messages arrive at the calling process as {:websocket_message, decoded_map}
      CCXT.WS.close(ws)

  For deribit's JSON-RPC subscribe (which carries an `id` field), `subscribe/3`
  returns `{:ok, response}` from zen_websocket's request-correlation; for bybit
  and okx it returns `:ok` and the subscribe-ack arrives asynchronously as a
  `{:websocket_message, _}` message.

  ## Scope

  Three canary exchanges are wired today: `bybit`, `deribit`, `okx`. Other
  priority-tier exchanges land in T93 (auth) + T94 (subscription patterns).
  """

  alias CCXT.Exchange
  alias CCXT.WS.Config
  alias CCXT.WS.Subscription
  alias CCXT.WS.URLRouting
  alias ZenWebsocket.Client, as: ZenClient

  @type section :: :public | :private

  @enforce_keys [:exchange, :zen_client, :url, :section]
  defstruct [:exchange, :zen_client, :url, :section]

  @type t :: %__MODULE__{
          exchange: Exchange.t(),
          zen_client: ZenClient.t(),
          url: String.t(),
          section: section()
        }

  @doc """
  Connects to the exchange's WebSocket endpoint for the given section
  (`:public` or `:private`).

  Extra opts are forwarded to `ZenWebsocket.Client.connect/2`. The connection's
  heartbeat config is resolved from `CCXT.WS.Config` unless the caller overrides
  `heartbeat_config` in opts.

  Returns `{:error, :unsupported_exchange}` if the exchange has no WS config,
  or `{:error, :no_url_configured}` if the requested section is absent.
  """
  @spec connect(Exchange.t(), section(), keyword()) :: {:ok, t()} | {:error, term()}
  def connect(%Exchange{} = exchange, section, opts \\ []) when section in [:public, :private] do
    with {:ok, config} <- fetch_config(exchange),
         {:ok, url} <- fetch_url(exchange, section),
         connect_opts = build_connect_opts(config, opts),
         {:ok, zen_client} <- ZenClient.connect(url, connect_opts) do
      {:ok, %__MODULE__{exchange: exchange, zen_client: zen_client, url: url, section: section}}
    end
  end

  @doc """
  Sends an exchange-native subscribe frame for the given channels.

  The frame is built by the exchange's registered `subscription_pattern` module
  (via `CCXT.WS.Subscription.build_subscribe/3`), encoded as JSON, and sent via
  `ZenWebsocket.Client.send_message/2`.

  Pattern modules return either a single map (most exchanges) or a list of maps
  (`:sub_subscribe`, `:reqtype_sub`, and `:custom` with `array_format` —
  HTX/BingX/Upbit emit one frame per channel). List returns are sent
  sequentially; the function returns `:ok` only if every frame sent successfully.

  For frames with an `id` field (deribit JSON-RPC), `send_message/2` blocks on
  the correlated response and returns `{:ok, response}`. For frames without an
  id (bybit, okx, etc.), returns `:ok`.

  Extra `opts` merge into the exchange's `subscription_config` from
  `CCXT.WS.Config` — used for runtime overrides like a fresh JSON-RPC id or an
  inline auth token. Keys must be atoms to override the atom-keyed base config;
  string-keyed maps coexist rather than override.

  TODO(adapter): `:rest_token` (kraken) and `:inline_subscribe` (coinbase) auth
  patterns require per-frame auth injection via `CCXT.WS.Auth.build_subscribe_auth/5`,
  which this function does not call. Private subscribes on those exchanges ship
  unauthenticated until the adapter layer lands (see CHANGELOG T94).
  """
  @spec subscribe(t(), [String.t() | map()], keyword() | map()) ::
          :ok | {:ok, map()} | {:error, term()}
  def subscribe(%__MODULE__{exchange: exchange, zen_client: zen_client}, channels, opts \\ []) when is_list(channels) do
    with {:ok, config} <- fetch_config(exchange),
         merged_config = merge_subscription_config(config, opts),
         {:ok, payload} <-
           Subscription.build_subscribe(config.subscription_pattern, channels, merged_config) do
      send_payload(zen_client, payload)
    end
  end

  @doc "Sends a raw (already-encoded or map) payload. Delegates to zen_websocket."
  @spec send_message(t(), String.t() | map()) :: :ok | {:ok, map()} | {:error, term()}
  def send_message(%__MODULE__{zen_client: zen_client}, payload) when is_binary(payload) do
    ZenClient.send_message(zen_client, payload)
  end

  def send_message(%__MODULE__{zen_client: zen_client}, %{} = payload) do
    ZenClient.send_message(zen_client, Jason.encode!(payload))
  end

  @doc "Closes the WebSocket connection."
  @spec close(t()) :: :ok
  def close(%__MODULE__{zen_client: zen_client}), do: ZenClient.close(zen_client)

  @doc "Returns the current connection state (`:connecting`, `:connected`, or `:disconnected`)."
  @spec get_state(t()) :: :connecting | :connected | :disconnected
  def get_state(%__MODULE__{zen_client: zen_client}), do: ZenClient.get_state(zen_client)

  @doc "Returns the resolved WS URL this connection is using."
  @spec get_url(t()) :: String.t()
  def get_url(%__MODULE__{url: url}), do: url

  # ---------------------------------------------------------------------------
  # Private
  # ---------------------------------------------------------------------------

  defp fetch_config(%Exchange{id: id}) do
    case Config.for_exchange(id) do
      nil -> {:error, :unsupported_exchange}
      config -> {:ok, config}
    end
  end

  defp fetch_url(%Exchange{} = exchange, :public) do
    case URLRouting.public_url(exchange) do
      nil -> {:error, :no_url_configured}
      url -> {:ok, url}
    end
  end

  defp fetch_url(%Exchange{} = exchange, :private) do
    case URLRouting.private_url(exchange) do
      nil -> {:error, :no_url_configured}
      url -> {:ok, url}
    end
  end

  defp build_connect_opts(config, opts) do
    heartbeat = Keyword.get(opts, :heartbeat_config, config.heartbeat)
    Keyword.put(opts, :heartbeat_config, heartbeat)
  end

  defp merge_subscription_config(%{subscription_config: base}, opts) when is_list(opts) do
    Map.merge(base, Map.new(opts))
  end

  defp merge_subscription_config(%{subscription_config: base}, opts) when is_map(opts) do
    Map.merge(base, opts)
  end

  defp send_payload(zen_client, frames) when is_list(frames) do
    Enum.reduce_while(frames, :ok, fn frame, _acc ->
      case ZenClient.send_message(zen_client, Jason.encode!(frame)) do
        :ok -> {:cont, :ok}
        {:ok, _response} = ok -> {:cont, ok}
        {:error, _} = err -> {:halt, err}
      end
    end)
  end

  defp send_payload(zen_client, %{} = payload) do
    ZenClient.send_message(zen_client, Jason.encode!(payload))
  end
end