ex_polymarket

defmodule Polymarket.WebSocket do
  @moduledoc """
  A WebSocket client for the Polymarket CLOB market feed.

  Wraps `Mint.WebSocket` in a `GenServer`, handling the HTTP upgrade,
  periodic pings, and decoding of incoming text frames.
  """

  use GenServer
  use TypedStruct

  alias Polymarket.WebSocket.Handlers.Noop
  alias Polymarket.WebSocket.MessageHandler

  require Logger
  require Mint.HTTP

  # The CLOB market feed lives at a single fixed endpoint.
  @ws_host "ws-subscriptions-clob.polymarket.com"
  @ws_port 443
  @ws_path "/ws/market"

  # `Mint.WebSocket.new/4`'s success typing is inferred as error-only (a known
  # Dialyzer limitation: the `with` flow inside its private `do_new/4` hides the
  # `{:ok, conn, websocket}` return), so the matching clause in `handle_responses/2`
  # is flagged as unreachable even though it fires on every successful upgrade.
  @dialyzer {:no_match, handle_responses: 2}

  typedstruct do
    field(:conn, Mint.HTTP.t())
    field(:websocket, Mint.WebSocket.t())
    field(:request_ref, Mint.Types.request_ref())
    field(:caller, GenServer.from())
    field(:status, pos_integer())
    field(:resp_headers, Mint.Types.headers())
    field(:closing?, boolean())
    field(:timer_ref, :timer.tref())
    field(:last_pong, DateTime.t())
    field(:handler, module(), default: Noop)
  end

  # ---------------------------------------------------------------------------#
  #                                API                                         #
  # ---------------------------------------------------------------------------#

  @doc """
  Starts a websocket connection.

  ## Options

    * `:handler` — module implementing `Polymarket.WebSocket.Handler` that
      receives every decoded event. Defaults to `Polymarket.WebSocket.Handlers.Noop`.

  Any other options are ignored.
  """
  @spec start_link(keyword()) :: {:ok, pid()} | {:error, term()}
  def start_link(opts) do
    handler = Keyword.get(opts, :handler, Noop)

    # The socket process decodes every inbound frame, so it churns short-lived
    # maps/structs/floats at a high rate. A larger starting heap avoids constant
    # young-gen GC (measured ~99% fewer collections, ~1.5x faster parse); an
    # off-heap mailbox keeps GC pauses bounded under burst backlog. Costs ~1.6 MB
    # resident heap per connection — size `min_heap_size` against your shard count.
    spawn_opt = [min_heap_size: 64_000, message_queue_data: :off_heap, fullsweep_after: 20]

    case GenServer.start_link(__MODULE__, handler, spawn_opt: spawn_opt) do
      {:ok, socket} ->
        case GenServer.call(socket, :connect) do
          {:ok, :connected} ->
            message = ~S"""
            {
              "assets_ids": [],
              "type": "market",
              "custom_feature_enabled": true
            }
            """

            send_message(socket, message)
            {:ok, socket}

          {:error, reason} ->
            # The process started but never connected; stop it so a failed
            # connect does not leave it lingering under the caller/supervisor.
            GenServer.stop(socket)
            {:error, reason}
        end

      error ->
        error
    end
  end

  @doc """
  Sends a string as a message over the websocket.
  """
  @spec send_message(pid(), String.t()) :: :ok
  def send_message(pid, text) do
    GenServer.call(pid, {:send_text, text})
  end

  # ---------------------------------------------------------------------------#
  #                                GenServer                                   #
  # ---------------------------------------------------------------------------#

  @impl GenServer
  def init(handler) do
    {:ok, timer} = :timer.send_interval(to_timeout(second: 3), :send_ping)
    {:ok, %__MODULE__{timer_ref: timer, handler: handler}}
  end

  @impl GenServer
  def handle_call({:send_text, text}, _from, state) do
    {:ok, state} = send_frame(state, {:text, text})
    {:reply, :ok, state}
  end

  @impl GenServer
  def handle_call(:connect, from, state) do
    # `nodelay: true` disables Nagle's algorithm so small outbound frames (pings
    # and, critically, order submissions sent in response to market data) are not
    # held back by the TCP stack. Mint does not set this by default, so Nagle is
    # otherwise on. User `transport_opts` survive Mint's option merge.
    with {:ok, conn} <-
           Mint.HTTP.connect(:https, @ws_host, @ws_port,
             protocols: [:http1],
             transport_opts: [nodelay: true]
           ),
         {:ok, conn, ref} <- Mint.WebSocket.upgrade(:wss, conn, @ws_path, []) do
      state = %{state | conn: conn, request_ref: ref, caller: from}
      {:noreply, state}
    else
      {:error, reason} ->
        {:reply, {:error, reason}, state}

      {:error, conn, reason} ->
        {:reply, {:error, reason}, put_in(state.conn, conn)}
    end
  end

  @impl GenServer
  def handle_info(:send_ping, state) do
    {:ok, state} = send_frame(state, {:text, "PING"})
    {:noreply, state}
  end

  def handle_info(message, state) do
    case Mint.WebSocket.stream(state.conn, message) do
      {:ok, conn, responses} ->
        state =
          put_in(state.conn, conn)
          |> handle_responses(responses)

        if state.closing? do
          do_close(state)
        else
          {:noreply, state}
        end

      {:error, conn, reason, _responses} ->
        Logger.debug("Error: #{inspect(binding())}")

        state =
          put_in(state.conn, conn)
          |> reply({:error, reason})

        {:stop, :server_disconnected, state}

      :unknown ->
        {:noreply, state}
    end
  end

  # ---------------------------------------------------------------------------#
  #                                Helpers                                     #
  # ---------------------------------------------------------------------------#

  # ---------------------------------------------------------------------------
  # Handle Response

  @spec handle_responses(t(), [Mint.Types.response()]) :: t()
  defp handle_responses(state, responses)

  defp handle_responses(%{request_ref: ref} = state, [{:status, ref, status} | rest]) do
    Logger.debug("Status: #{inspect(status)}")

    put_in(state.status, status)
    |> handle_responses(rest)
  end

  defp handle_responses(%{request_ref: ref} = state, [{:headers, ref, resp_headers} | rest]) do
    Logger.debug("Headers: #{inspect(resp_headers)}")

    put_in(state.resp_headers, resp_headers)
    |> handle_responses(rest)
  end

  # executes
  defp handle_responses(%{request_ref: ref, status: status, resp_headers: resp_headers} = state, [{:done, ref} | rest])
       when is_integer(status) and is_list(resp_headers) do
    Logger.debug("Done")

    case Mint.WebSocket.new(state.conn, ref, status, resp_headers) do
      {:ok, conn, websocket} ->
        %{state | conn: conn, websocket: websocket, status: nil, resp_headers: nil}
        |> reply({:ok, :connected})
        |> handle_responses(rest)

      {:error, conn, reason} ->
        Logger.error("Error in response #{inspect(binding())}")

        put_in(state.conn, conn)
        |> reply({:error, reason})
    end
  end

  defp handle_responses(%{request_ref: ref, websocket: websocket} = state, [{:data, ref, data} | rest])
       when websocket != nil do
    case Mint.WebSocket.decode(websocket, data) do
      {:ok, websocket, frames} ->
        put_in(state.websocket, websocket)
        |> handle_frames(frames)
        |> handle_responses(rest)

      {:error, websocket, reason} ->
        put_in(state.websocket, websocket)
        |> reply({:error, reason})
    end
  end

  defp handle_responses(state, [_response | rest]) do
    handle_responses(state, rest)
  end

  defp handle_responses(state, []), do: state

  # ---------------------------------------------------------------------------
  # Send Frame

  @spec send_frame(t(), Mint.WebSocket.frame() | Mint.WebSocket.shorthand_frame()) ::
          {:ok, t()} | {:error, t(), term()}
  defp send_frame(state, frame) do
    case Mint.WebSocket.encode(state.websocket, frame) do
      {:ok, websocket, data} ->
        state = put_in(state.websocket, websocket)

        case Mint.WebSocket.stream_request_body(state.conn, state.request_ref, data) do
          {:ok, conn} -> {:ok, put_in(state.conn, conn)}
          {:error, conn, reason} -> {:error, put_in(state.conn, conn), reason}
        end

      {:error, websocket, reason} ->
        {:error, put_in(state.websocket, websocket), reason}
    end
  end

  # ---------------------------------------------------------------------------
  # Receive Frame

  @spec handle_frames(t(), [Mint.WebSocket.frame()]) :: t()
  defp handle_frames(state, frames) do
    Enum.reduce(frames, state, fn
      # reply to pings with pongs
      {:ping, data}, state ->
        Logger.debug("Received ping")
        {:ok, state} = send_frame(state, {:pong, data})
        state

      {:close, _code, reason}, state ->
        Logger.debug("Closing connection: #{inspect(reason)}")
        %{state | closing?: true}

      {:text, "PONG"}, state ->
        Logger.debug("Received pong")
        put_in(state.last_pong, DateTime.utc_now())

      {:text, text}, state ->
        handle_text(text, state)

      frame, state ->
        Logger.debug("Unexpected frame received: #{inspect(frame)}")
        state
    end)
  end

  @spec handle_text(String.t(), t()) :: t()
  defp handle_text(text, state) do
    # try and decode the message to json. the message can be a list of
    # events, so deal with those too.
    case decode_message(text) do
      {:ok, messages} ->
        messages
        # Can be a single or a list of messages, so always turn it into a list.
        |> List.wrap()
        |> Enum.reduce(state, fn message, state ->
          MessageHandler.handle_message(message, state)
          |> process_result()
        end)

      {:error, err} ->
        Logger.warning("failed to decode message #{text} #{inspect(err)}")
        state
    end
  end

  # ---------------------------------------------------------------------------
  # Closing

  @spec do_close(t()) :: {:stop, :normal, t()}
  defp do_close(state) do
    Logger.debug("Closing websocket #{inspect(state)}")
    # Streaming a close frame may fail if the server has already closed
    # for writing.
    _ = send_frame(state, :close)
    Mint.HTTP.close(state.conn)
    {:stop, :normal, state}
  end

  # ---------------------------------------------------------------------------
  # Reply

  @spec reply(t(), term()) :: t()
  defp reply(state, response) do
    if state.caller, do: GenServer.reply(state.caller, response)
    put_in(state.caller, nil)
  end

  # ---------------------------------------------------------------------------
  # Decoding

  @spec decode_message(String.t()) :: {:ok, term()} | {:error, Jason.DecodeError.t()}
  defp decode_message(message), do: Jason.decode(message, keys: :atoms)

  # ---------------------------------------------------------------------------
  # Process response from handler

  @spec process_result(
          {:reply, {:text, String.t()}, t()}
          | {:noreply, t()}
        ) :: t()
  defp process_result({:reply, frame, state}) do
    {:ok, state} = send_frame(state, frame)
    state
  end

  defp process_result({:noreply, state}) do
    state
  end
end