polymarket_sdk

defmodule Polymarket.RTDS do
  @moduledoc """
  Real-time trade feed client for Polymarket (RTDS — Real-Time Data Socket).

  Connects to `wss://ws-live-data.polymarket.com` and forwards every trade
  on Polymarket to a callback PID as
  `{:polymarket_rtds, :trade, payload}`. The payload is the raw map sent
  by Polymarket; normalization (typed structs, snake_case keys) is left to
  the caller — typically by feeding the same map shape into
  `Polymarket.Activity.normalize/1`.

  The surface is deliberately narrow:

    * **No backpressure / batching / debouncing.** Every trade event is
      forwarded as a separate Erlang message. Busy markets can produce
      5–20 events/second and a slow callback can flood. If this is a
      problem, debounce in the callback or migrate to a future
      demand-driven (`GenStage`/`Broadway`) shape.
    * **Reconnects on transient drops, stays closed on deliberate
      close.** A network drop reconnects (with exponential backoff so a
      hard-down endpoint does not storm); a deliberate close — e.g. the
      callback PID dying — stays closed. This module is not embedded in
      a `Supervisor` here — callers that want crash-and-restart
      guarantees should add their own supervisor.
    * **No topic generalization.** Only the global trades feed
      (`{"topic": "activity", "type": "trades"}`) is subscribed. New
      Polymarket RTDS topics (resolutions, liquidity events) are not
      modeled.
    * **No payload normalization.** Forwarded payloads are the raw maps
      from Polymarket. Pair with `Polymarket.Activity.normalize/1` if you
      need typed structs.
    * **Callback is PID-only.** The `{module, function}` callback form is
      not supported.

  ## Usage

      {:ok, _pid} = Polymarket.RTDS.start_link(callback: self())

      receive do
        {:polymarket_rtds, :trade, payload} ->
          IO.inspect(payload, label: "rtds trade")
      end

  ## Trade payload fields (raw, from Polymarket)

    * `proxyWallet` — trader's wallet address.
    * `pseudonym` — trader's display name.
    * `side` — `"BUY"` or `"SELL"`.
    * `outcome` — `"Yes"`, `"No"`, or named outcome.
    * `price` — execution price (number).
    * `size` — trade size (number).
    * `asset` — outcome token id (decimal string).
    * `conditionId` — market condition id (hex).
    * `eventSlug` — market slug.
    * `title` — human-readable market name.
    * `transactionHash` — on-chain tx hash.
    * `timestamp` — Unix timestamp.

  ## Options

    * `:callback` *(required)* — PID to receive trade events.
    * `:url` — Override the WebSocket URL (default `#{inspect("wss://ws-live-data.polymarket.com")}`).
    * `:ping_interval` — PING frame interval in milliseconds (default `5_000`).
    * `:name` — Optional process name passed to `WebSockex.start_link/4`.
  """

  use WebSockex

  require Logger

  @rtds_url "wss://ws-live-data.polymarket.com"
  @default_ping_interval_ms 5_000
  @max_backoff_ms 30_000

  @doc """
  Starts the RTDS client and connects.

  Raises `ArgumentError` if `:callback` is missing or not a PID.
  """
  @spec start_link(keyword()) :: {:ok, pid()} | {:error, term()}
  def start_link(opts) do
    callback = validate_callback!(opts)
    url = Keyword.get(opts, :url, @rtds_url)
    ping_interval = validate_ping_interval!(opts)

    state = %{
      callback: callback,
      callback_monitor: nil,
      ping_interval_ms: ping_interval,
      ping_timer: nil,
      stop: false
    }

    ws_opts =
      case Keyword.get(opts, :name) do
        nil -> []
        name -> [name: name]
      end

    WebSockex.start_link(url, __MODULE__, state, ws_opts)
  end

  @doc """
  Default RTDS production URL. Exposed so tests can compare against it.
  """
  @spec default_url() :: String.t()
  def default_url, do: @rtds_url

  @doc """
  Default ping interval in milliseconds.
  """
  @spec default_ping_interval_ms() :: pos_integer()
  def default_ping_interval_ms, do: @default_ping_interval_ms

  # ── WebSockex Callbacks ──

  @impl true
  def handle_connect(_conn, state) do
    Logger.info("[Polymarket.RTDS] Connected to trade feed")

    # Cancel any ping timer left over from a previous connection before
    # scheduling a fresh one — otherwise each reconnect leaks a timer
    # and PING frames pile up.
    state = reschedule_ping(state)

    state = maybe_monitor_callback(state)

    # Subscribe to the global trades topic on the next message turn so
    # the connect phase finishes cleanly before we send a frame.
    Process.send_after(self(), :subscribe, 0)

    {:ok, state}
  end

  @impl true
  def handle_info(:subscribe, state) do
    msg =
      Jason.encode!(%{
        "action" => "subscribe",
        "subscriptions" => [%{"topic" => "activity", "type" => "trades"}]
      })

    Logger.info("[Polymarket.RTDS] Subscribing to global trade feed")
    {:reply, {:text, msg}, state}
  end

  @impl true
  def handle_info(:ping, state) do
    state = reschedule_ping(state)
    {:reply, {:text, "PING"}, state}
  end

  @impl true
  def handle_info({:DOWN, _ref, :process, pid, reason}, %{callback: pid} = state) do
    Logger.warning("[Polymarket.RTDS] Callback PID died (#{inspect(reason)}), closing")

    # Deliberate close: cancel the ping timer and flag the disconnect as
    # intentional so `handle_disconnect/2` does NOT reconnect. Without
    # the flag the process would zombie-reconnect forever and keep
    # `send/2`-ing to a dead callback PID.
    state = cancel_ping(state)
    {:close, %{state | stop: true}}
  end

  @impl true
  def handle_info(_msg, state), do: {:ok, state}

  @impl true
  def handle_frame({:text, "PONG"}, state), do: {:ok, state}

  @impl true
  def handle_frame({:text, msg}, state) do
    case Jason.decode(msg) do
      {:ok, %{"topic" => "activity", "type" => "trades", "payload" => payload}} ->
        emit(state.callback, :trade, payload)

      {:ok, %{"message" => error_msg}} ->
        Logger.warning("[Polymarket.RTDS] Server error: #{inspect(error_msg)}")

      {:ok, _other} ->
        :ok

      {:error, _} ->
        :ok
    end

    {:ok, state}
  end

  @impl true
  def handle_frame(_frame, state), do: {:ok, state}

  @impl true
  def handle_disconnect(_status, %{stop: true} = state) do
    # Intentional close (e.g. callback PID died). Honor it — returning
    # {:ok, state} lets the WebSockex process terminate instead of
    # reconnecting into a zombie loop.
    Logger.info("[Polymarket.RTDS] Closed intentionally; not reconnecting")
    {:ok, state}
  end

  @impl true
  def handle_disconnect(status, state) do
    reason = Map.get(status, :reason)
    attempt = Map.get(status, :attempt_number, 1)
    delay = backoff_ms(attempt)

    Logger.warning(
      "[Polymarket.RTDS] Disconnected: #{inspect(reason)}, reconnecting in #{delay}ms " <>
        "(attempt #{attempt})"
    )

    # Exponential backoff between reconnect attempts so a hard-down
    # endpoint does not storm. The first attempt is immediate; each
    # subsequent failed attempt (WebSockex increments `attempt_number`
    # and resets it to 1 on a successful connect) doubles the delay up
    # to the cap. WebSockex reconnects synchronously from this callback,
    # so the wait is a bounded `Process.sleep`; it only runs while the
    # socket is already down and has no book/ping work to do.
    if delay > 0, do: Process.sleep(delay)

    {:reconnect, state}
  end

  # ── Private ──

  defp validate_callback!(opts) do
    case Keyword.get(opts, :callback) do
      nil ->
        raise ArgumentError, "missing required option :callback for Polymarket.RTDS"

      pid when is_pid(pid) ->
        pid

      other ->
        raise ArgumentError,
              ":callback must be a PID, got: #{inspect(other)}. " <>
                "Only the callback-PID form is supported; {module, function} is not."
    end
  end

  defp validate_ping_interval!(opts) do
    case Keyword.get(opts, :ping_interval, @default_ping_interval_ms) do
      n when is_integer(n) and n > 0 ->
        n

      other ->
        raise ArgumentError,
              ":ping_interval must be a positive integer (milliseconds), got: #{inspect(other)}"
    end
  end

  # Cancel any outstanding ping timer, then schedule a fresh one and
  # store its ref in state. Cancelling first is what prevents a timer
  # leak across reconnects.
  defp reschedule_ping(state) do
    state = cancel_ping(state)
    ref = Process.send_after(self(), :ping, state.ping_interval_ms)
    %{state | ping_timer: ref}
  end

  defp cancel_ping(state) do
    case Map.get(state, :ping_timer) do
      ref when is_reference(ref) -> Process.cancel_timer(ref)
      _ -> :ok
    end

    %{state | ping_timer: nil}
  end

  # First reconnect attempt is immediate (0ms); afterwards back off
  # exponentially: 1s, 2s, 4s, … capped at @max_backoff_ms.
  defp backoff_ms(attempt) when is_integer(attempt) and attempt <= 1, do: 0

  defp backoff_ms(attempt) when is_integer(attempt) do
    min(@max_backoff_ms, 1_000 * Bitwise.bsl(1, attempt - 2))
  end

  defp maybe_monitor_callback(%{callback: pid, callback_monitor: nil} = state)
       when is_pid(pid) do
    ref = Process.monitor(pid)
    %{state | callback_monitor: ref}
  end

  defp maybe_monitor_callback(state), do: state

  defp emit(pid, event_type, data) when is_pid(pid) do
    send(pid, {:polymarket_rtds, event_type, data})
  end
end