polynode 0.10.0

//! PolyNode client with REST methods and WebSocket stream creation.

use std::time::Duration;
use reqwest::header::{HeaderMap, HeaderValue};
use serde::de::DeserializeOwned;

use crate::error::{Error, Result};
use crate::types::rest::*;
use crate::types::common::CandleResolution;
use crate::ws::{WsStream, StreamOptions};
use crate::orderbook::{ObStream, ObStreamOptions};
use crate::short_form::{ShortFormInterval, ShortFormBuilder};

/// Main PolyNode client.
pub struct PolyNodeClient {
    pub(crate) http: reqwest::Client,
    pub(crate) api_key: String,
    pub(crate) base_url: String,
    pub(crate) ws_url: String,
    pub(crate) ob_url: String,
    pub(crate) rpc_url: String,
}

/// Builder for configuring a PolyNodeClient.
pub struct ClientBuilder {
    api_key: String,
    base_url: String,
    ws_url: String,
    ob_url: String,
    rpc_url: String,
    timeout: Duration,
}

impl ClientBuilder {
    pub fn new(api_key: impl Into<String>) -> Self {
        Self {
            api_key: api_key.into(),
            base_url: "https://api.polynode.dev".into(),
            ws_url: "wss://ws.polynode.dev/ws".into(),
            ob_url: "wss://ob.polynode.dev/ws".into(),
            rpc_url: "https://rpc.polynode.dev".into(),
            timeout: Duration::from_secs(10),
        }
    }

    pub fn base_url(mut self, url: impl Into<String>) -> Self {
        self.base_url = url.into();
        self
    }

    pub fn ws_url(mut self, url: impl Into<String>) -> Self {
        self.ws_url = url.into();
        self
    }

    pub fn ob_url(mut self, url: impl Into<String>) -> Self {
        self.ob_url = url.into();
        self
    }

    pub fn rpc_url(mut self, url: impl Into<String>) -> Self {
        self.rpc_url = url.into();
        self
    }

    pub fn timeout(mut self, timeout: Duration) -> Self {
        self.timeout = timeout;
        self
    }

    pub fn build(self) -> Result<PolyNodeClient> {
        let mut headers = HeaderMap::new();
        headers.insert(
            "x-api-key",
            HeaderValue::from_str(&self.api_key).map_err(|_| Error::Auth("Invalid API key format".into()))?,
        );

        let http = reqwest::Client::builder()
            .default_headers(headers)
            .timeout(self.timeout)
            .build()?;

        Ok(PolyNodeClient {
            http,
            api_key: self.api_key,
            base_url: self.base_url.trim_end_matches('/').to_string(),
            ws_url: self.ws_url.trim_end_matches('/').to_string(),
            ob_url: self.ob_url.trim_end_matches('/').to_string(),
            rpc_url: self.rpc_url.trim_end_matches('/').to_string(),
        })
    }
}

impl PolyNodeClient {
    /// Create a client with defaults.
    pub fn new(api_key: impl Into<String>) -> Result<Self> {
        ClientBuilder::new(api_key).build()
    }

    /// Create a builder for full configuration.
    pub fn builder(api_key: impl Into<String>) -> ClientBuilder {
        ClientBuilder::new(api_key)
    }

    // ── Internal HTTP helpers ──

    pub(crate) async fn get<T: DeserializeOwned>(&self, path: &str) -> Result<T> {
        let url = format!("{}{}", self.base_url, path);
        let resp = self.http.get(&url).send().await?;
        Self::handle_response(resp).await
    }

    pub(crate) async fn get_no_auth<T: DeserializeOwned>(&self, path: &str) -> Result<T> {
        let url = format!("{}{}", self.base_url, path);
        let client = reqwest::Client::new();
        let resp = client.get(&url).send().await?;
        Self::handle_response(resp).await
    }

    pub(crate) async fn get_text(&self, path: &str) -> Result<String> {
        let url = format!("{}{}", self.base_url, path);
        let client = reqwest::Client::new();
        let resp = client.get(&url).send().await?;
        if !resp.status().is_success() {
            return Err(Self::parse_error(resp).await);
        }
        Ok(resp.text().await?)
    }

    async fn handle_response<T: DeserializeOwned>(resp: reqwest::Response) -> Result<T> {
        if !resp.status().is_success() {
            return Err(Self::parse_error(resp).await);
        }
        Ok(resp.json().await?)
    }

    async fn parse_error(resp: reqwest::Response) -> Error {
        let status = resp.status().as_u16();
        let message = resp.text().await.unwrap_or_default();

        // Try to parse as JSON error
        let msg = if let Ok(v) = serde_json::from_str::<serde_json::Value>(&message) {
            v.get("error")
                .and_then(|e| e.as_str())
                .unwrap_or(&message)
                .to_string()
        } else {
            message
        };

        match status {
            401 => Error::Auth(msg),
            403 => Error::Auth(msg),
            404 => Error::NotFound(msg),
            429 => Error::RateLimited(msg),
            _ => Error::Api { status, message: msg },
        }
    }

    // ── System ──

    /// Liveness probe. No auth required.
    pub async fn healthz(&self) -> Result<String> {
        self.get_text("/healthz").await
    }

    /// Readiness check. No auth required.
    pub async fn readyz(&self) -> Result<serde_json::Value> {
        self.get_no_auth("/readyz").await
    }

    /// System status with metrics.
    pub async fn status(&self) -> Result<StatusResponse> {
        self.get("/v1/status").await
    }

    /// Generate a new API key. No auth required.
    pub async fn create_key(&self, name: Option<&str>) -> Result<ApiKeyResponse> {
        let url = format!("{}/v1/keys", self.base_url);
        let body = serde_json::json!({ "name": name.unwrap_or("unnamed") });
        let client = reqwest::Client::new();
        let resp = client
            .post(&url)
            .json(&body)
            .send()
            .await?;
        Self::handle_response(resp).await
    }

    // ── Markets ──

    /// Top markets sorted by 24h volume.
    pub async fn markets(&self, count: Option<u64>) -> Result<MarketsResponse> {
        let mut path = "/v1/markets".to_string();
        if let Some(c) = count {
            path = format!("{}?count={}", path, c);
        }
        self.get(&path).await
    }

    /// Market detail by token ID.
    pub async fn market(&self, token_id: &str) -> Result<serde_json::Value> {
        self.get(&format!("/v1/markets/{}", token_id)).await
    }

    /// Market detail by URL slug.
    pub async fn market_by_slug(&self, slug: &str) -> Result<serde_json::Value> {
        self.get(&format!("/v1/markets/slug/{}", slug)).await
    }

    /// Market detail by condition ID.
    pub async fn market_by_condition(&self, condition_id: &str) -> Result<serde_json::Value> {
        self.get(&format!("/v1/markets/condition/{}", condition_id)).await
    }

    /// Filtered, paginated market listing.
    pub async fn list_markets(&self, params: &ListMarketsParams) -> Result<MarketsListResponse> {
        let mut parts = vec!["/v1/markets/list".to_string()];
        let mut query = Vec::new();
        if let Some(c) = params.count { query.push(format!("count={}", c)); }
        if let Some(ref s) = params.sort { query.push(format!("sort={}", s)); }
        if let Some(ref c) = params.category { query.push(format!("category={}", c)); }
        if let Some(v) = params.min_volume { query.push(format!("min_volume={}", v)); }
        if let Some(a) = params.active_only { query.push(format!("active_only={}", a)); }
        if let Some(c) = params.cursor { query.push(format!("cursor={}", c)); }
        if !query.is_empty() {
            parts.push(format!("?{}", query.join("&")));
        }
        self.get(&parts.join("")).await
    }

    /// Full-text search across market questions.
    pub async fn search(&self, query: &str, limit: Option<u64>, include_inactive: Option<bool>) -> Result<SearchResponse> {
        let mut params = vec![format!("q={}", query)];
        if let Some(l) = limit { params.push(format!("limit={}", l)); }
        if let Some(i) = include_inactive { params.push(format!("include_inactive={}", i)); }
        self.get(&format!("/v1/search?{}", params.join("&"))).await
    }

    // ── Pricing ──

    /// OHLCV candles for a token.
    pub async fn candles(&self, token_id: &str, resolution: Option<CandleResolution>, limit: Option<u64>) -> Result<CandlesResponse> {
        let mut params = Vec::new();
        if let Some(r) = resolution { params.push(format!("resolution={}", r)); }
        if let Some(l) = limit { params.push(format!("limit={}", l)); }
        let qs = if params.is_empty() { String::new() } else { format!("?{}", params.join("&")) };
        self.get(&format!("/v1/candles/{}{}", token_id, qs)).await
    }

    /// Market statistics.
    pub async fn stats(&self, token_id: &str) -> Result<serde_json::Value> {
        self.get(&format!("/v1/stats/{}", token_id)).await
    }

    // ── Settlements ──

    /// Most recent decoded settlements.
    pub async fn recent_settlements(&self, count: Option<u64>) -> Result<SettlementsResponse> {
        let mut path = "/v1/settlements/recent".to_string();
        if let Some(c) = count { path = format!("{}?count={}", path, c); }
        self.get(&path).await
    }

    /// Settlements for a specific token ID.
    pub async fn token_settlements(&self, token_id: &str, count: Option<u64>) -> Result<SettlementsResponse> {
        let mut path = format!("/v1/settlements/token/{}", token_id);
        if let Some(c) = count { path = format!("{}?count={}", path, c); }
        self.get(&path).await
    }

    /// Settlements for a specific wallet address.
    pub async fn wallet_settlements(&self, address: &str, count: Option<u64>) -> Result<SettlementsResponse> {
        let mut path = format!("/v1/settlements/wallet/{}", address);
        if let Some(c) = count { path = format!("{}?count={}", path, c); }
        self.get(&path).await
    }

    // ── Wallets ──

    /// Wallet activity summary.
    pub async fn wallet(&self, address: &str) -> Result<serde_json::Value> {
        self.get(&format!("/v1/wallets/{}", address)).await
    }

    /// Positions for a wallet (proxied from Polymarket data-api). Returns full P&L, current values.
    pub async fn wallet_positions_data(&self, address: &str, limit: Option<u64>, offset: Option<u64>) -> Result<WalletPositionsResponse> {
        let mut params = Vec::new();
        if let Some(l) = limit { params.push(format!("limit={l}")); }
        if let Some(o) = offset { params.push(format!("offset={o}")); }
        let qs = if params.is_empty() { String::new() } else { format!("?{}", params.join("&")) };
        self.get(&format!("/v1/wallets/{}/positions{}", address, qs)).await
    }

    /// Onchain positions from the PNL subgraph. Returns ALL positions (open + closed) with accurate realized P&L.
    pub async fn wallet_onchain_positions(&self, address: &str) -> Result<WalletOnchainPositionsResponse> {
        self.get(&format!("/v2/wallets/{}/positions/onchain", address)).await
    }

    /// Trades for a wallet. Used by the cache for backfill.
    pub async fn wallet_trades(&self, address: &str, limit: Option<u64>, offset: Option<u64>) -> Result<WalletTradesResponse> {
        let mut params = Vec::new();
        if let Some(l) = limit { params.push(format!("limit={l}")); }
        if let Some(o) = offset { params.push(format!("offset={o}")); }
        let qs = if params.is_empty() { String::new() } else { format!("?{}", params.join("&")) };
        self.get(&format!("/v1/wallets/{}/trades{}", address, qs)).await
    }

    /// Trades for a market (by condition ID or slug).
    pub async fn market_trades(&self, id: &str, limit: Option<u64>, offset: Option<u64>, side: Option<&str>, user: Option<&str>) -> Result<MarketTradesResponse> {
        let mut params = Vec::new();
        if let Some(l) = limit { params.push(format!("limit={l}")); }
        if let Some(o) = offset { params.push(format!("offset={o}")); }
        if let Some(s) = side { params.push(format!("side={s}")); }
        if let Some(u) = user { params.push(format!("user={u}")); }
        let qs = if params.is_empty() { String::new() } else { format!("?{}", params.join("&")) };
        self.get(&format!("/v1/markets/{}/trades{}", id, qs)).await
    }

    // ── Orderbook (REST) ──

    /// Full orderbook snapshot from the CLOB.
    pub async fn orderbook_rest(&self, token_id: &str) -> Result<OrderbookRestResponse> {
        self.get(&format!("/v1/orderbook/{}", token_id)).await
    }

    /// Midpoint price for a token.
    pub async fn midpoint(&self, token_id: &str) -> Result<MidpointResponse> {
        self.get(&format!("/v1/midpoint/{}", token_id)).await
    }

    /// Bid-ask spread for a token.
    pub async fn spread(&self, token_id: &str) -> Result<SpreadResponse> {
        self.get(&format!("/v1/spread/{}", token_id)).await
    }

    // ── Enriched Data ──

    /// Top traders leaderboard.
    pub async fn leaderboard(&self, period: Option<&str>, sort: Option<&str>) -> Result<LeaderboardResponse> {
        let mut params = Vec::new();
        if let Some(p) = period { params.push(format!("period={}", p)); }
        if let Some(s) = sort { params.push(format!("sort={}", s)); }
        let qs = if params.is_empty() { String::new() } else { format!("?{}", params.join("&")) };
        self.get(&format!("/v1/leaderboard{}", qs)).await
    }

    /// Trending markets (carousel, breaking, hot topics, featured, movers).
    pub async fn trending(&self) -> Result<TrendingResponse> {
        self.get("/v1/trending").await
    }

    /// Recent global trading activity.
    pub async fn activity(&self) -> Result<ActivityResponse> {
        self.get("/v1/activity").await
    }

    /// Markets with largest 24h price moves.
    pub async fn movers(&self) -> Result<MoversResponse> {
        self.get("/v1/movers").await
    }

    /// Trader profile with stats.
    pub async fn trader_profile(&self, wallet: &str) -> Result<TraderProfile> {
        self.get(&format!("/v1/trader/{}", wallet)).await
    }

    /// Trader P&L time series.
    pub async fn trader_pnl(&self, wallet: &str, period: Option<&str>) -> Result<TraderPnlResponse> {
        let qs = period.map(|p| format!("?period={}", p)).unwrap_or_default();
        self.get(&format!("/v1/trader/{}/pnl{}", wallet, qs)).await
    }

    /// Event detail by slug (includes all markets within the event).
    pub async fn event(&self, slug: &str) -> Result<EventDetailResponse> {
        self.get(&format!("/v1/event/{}", slug)).await
    }

    /// Search events by query string.
    pub async fn search_events(&self, query: &str, limit: Option<u64>) -> Result<EventSearchResponse> {
        let mut params = vec![format!("q={}", query)];
        if let Some(l) = limit { params.push(format!("limit={}", l)); }
        self.get(&format!("/v1/events/search?{}", params.join("&"))).await
    }

    /// Markets filtered by category. Delegates to `/v1/markets/list` with a category filter.
    pub async fn markets_by_category(&self, category: &str) -> Result<MarketsListResponse> {
        self.list_markets(&ListMarketsParams {
            category: Some(category.to_string()),
            ..Default::default()
        }).await
    }

    // ── RPC ──

    /// Send a JSON-RPC request through the PolyNode RPC endpoint (rpc.polynode.dev).
    pub async fn rpc_call(&self, method: &str, params: serde_json::Value) -> Result<serde_json::Value> {
        let body = serde_json::json!({
            "jsonrpc": "2.0",
            "method": method,
            "params": params,
            "id": 1
        });
        let resp = self.http
            .post(&self.rpc_url)
            .json(&body)
            .send()
            .await?;

        if !resp.status().is_success() {
            return Err(Self::parse_error(resp).await);
        }

        let data: JsonRpcResponse = resp.json().await?;
        if let Some(err) = data.error {
            return Err(Error::Api {
                status: err.code as u16,
                message: err.message,
            });
        }
        Ok(data.result.unwrap_or(serde_json::Value::Null))
    }

    // ── WebSocket ──

    /// Open a WebSocket connection, returning a stream of messages.
    pub async fn stream(&self, options: StreamOptions) -> Result<WsStream> {
        WsStream::connect(&self.api_key, &self.ws_url, options).await
    }

    // ── Short-Form Markets ──

    /// Subscribe to short-form crypto markets (5m, 15m, 1h) with auto-rotation.
    ///
    /// Returns a builder to configure coins and options before starting.
    ///
    /// # Example
    /// ```rust,no_run
    /// # async fn example(client: &polynode::PolyNodeClient) -> polynode::Result<()> {
    /// let mut stream = client
    ///     .short_form(polynode::ShortFormInterval::FifteenMin)
    ///     .coins(&[polynode::Coin::Btc, polynode::Coin::Eth])
    ///     .start()
    ///     .await?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn short_form(&self, interval: ShortFormInterval) -> ShortFormBuilder<'_> {
        ShortFormBuilder {
            client: self,
            interval,
            coins: None,
            rotation_buffer: 3,
        }
    }

    // ── Orderbook ──

    /// Open an orderbook WebSocket stream from ob.polynode.dev.
    pub async fn orderbook_stream(&self, options: ObStreamOptions) -> Result<ObStream> {
        ObStream::connect(&self.api_key, &self.ob_url, options).await
    }
}