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