wickra-exchange 0.2.1

//! Real-socket transport adapters.
//!
//! These implement the core's [`HttpTransport`] (and, later, `WsTransport`) over
//! actual sockets. They are the only part of the library that touches the
//! network, so they are deliberately thin and are **excluded from coverage** —
//! exercised only by gated `#[ignore]` integration tests against live testnets,
//! never by the offline unit suite that drives the mock transports.

use flate2::read::GzDecoder;
use futures_util::{SinkExt, StreamExt};
use std::io::Read as _;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::mpsc::{self, Receiver, TryRecvError};
use std::sync::Arc;
use tokio::sync::mpsc::{unbounded_channel, UnboundedSender};
use tokio_tungstenite::tungstenite::Message;
use wickra_exchange_core::{
    Error, ExchangeOptions, HttpMethod, HttpRequest, HttpResponse, HttpTransport, Result,
    WsConnection, WsTransport,
};

/// A synchronous HTTP transport backed by `reqwest`'s blocking client (rustls
/// TLS). The blocking client keeps the public surface synchronous with no tokio
/// runtime for callers to manage.
#[derive(Debug)]
pub struct ReqwestHttpTransport {
    client: reqwest::blocking::Client,
}

impl ReqwestHttpTransport {
    /// Build a transport from the connection options (timeout, user agent, proxy).
    ///
    /// # Errors
    ///
    /// Returns [`Error::Network`] if the underlying client cannot be constructed.
    pub fn new(options: &ExchangeOptions) -> Result<Self> {
        let mut builder = reqwest::blocking::Client::builder().timeout(options.timeout);
        if let Some(user_agent) = &options.user_agent {
            builder = builder.user_agent(user_agent);
        }
        if let Some(proxy) = &options.proxy {
            let proxy = reqwest::Proxy::all(proxy).map_err(|e| Error::Network(e.to_string()))?;
            builder = builder.proxy(proxy);
        }
        let client = builder.build().map_err(|e| Error::Network(e.to_string()))?;
        Ok(Self { client })
    }
}

fn reqwest_method(method: HttpMethod) -> reqwest::Method {
    match method {
        HttpMethod::Get => reqwest::Method::GET,
        HttpMethod::Post => reqwest::Method::POST,
        HttpMethod::Put => reqwest::Method::PUT,
        HttpMethod::Patch => reqwest::Method::PATCH,
        HttpMethod::Delete => reqwest::Method::DELETE,
    }
}

impl HttpTransport for ReqwestHttpTransport {
    fn execute(&self, request: &HttpRequest) -> Result<HttpResponse> {
        let mut builder = self
            .client
            .request(reqwest_method(request.method), &request.url);
        for (name, value) in &request.headers {
            builder = builder.header(name, value);
        }
        if let Some(body) = &request.body {
            builder = builder.body(body.clone());
        }
        let response = builder.send().map_err(|e| {
            if e.is_timeout() {
                Error::Timeout
            } else {
                Error::Network(e.to_string())
            }
        })?;
        let status = response.status().as_u16();
        // Carried through because the body is not the whole answer: a venue
        // that rate-limits says how long to wait in `Retry-After`, and
        // dropping the headers here is what left `Error::RateLimited` with
        // `retry_after: None` at every site that raises it. Values that are
        // not valid UTF-8 are skipped rather than failing the response --
        // a malformed header is not a reason to lose a body that parsed.
        let headers: Vec<(String, String)> = response
            .headers()
            .iter()
            .filter_map(|(name, value)| {
                value
                    .to_str()
                    .ok()
                    .map(|value| (name.as_str().to_string(), value.to_string()))
            })
            .collect();
        let body = response.text().map_err(|e| Error::Network(e.to_string()))?;
        Ok(HttpResponse {
            status,
            headers,
            body,
        })
    }
}

/// A pull-based WebSocket transport backed by tokio-tungstenite.
///
/// Each connection spawns a dedicated thread running a current-thread tokio
/// runtime that owns the socket: it reads frames into a channel, answers pings,
/// and forwards outbound messages. The synchronous [`WsConnection`] drains the
/// inbound channel non-blockingly (`recv` returns `Ok(None)` when nothing is
/// pending), so the pull model needs no runtime on the caller's side. Dropping
/// the connection closes the outbound channel, which shuts the reader down.
#[derive(Debug, Default)]
pub struct TungsteniteWsTransport;

impl TungsteniteWsTransport {
    /// A new transport.
    #[must_use]
    pub fn new() -> Self {
        Self
    }
}

/// Turn a binary frame into text, decompressing it if it is gzip.
///
/// HTX sends **every** frame gzip-compressed, over spot and futures alike, and
/// gzip is never valid UTF-8 -- its first two bytes are `1f 8b`. This arm used
/// to forward a binary frame only when it decoded as UTF-8, so an HTX stream
/// delivered nothing: not one trade, not one book update, no error either. The
/// venue's unit tests passed throughout, because the mock transport hands the
/// parser plain JSON and never a real frame.
///
/// A frame that is neither gzip nor UTF-8 is still dropped: there is nothing to
/// hand a parser that speaks text.
fn decode_binary(bytes: &[u8]) -> Option<String> {
    if bytes.starts_with(&[0x1f, 0x8b]) {
        let mut text = String::new();
        return GzDecoder::new(bytes)
            .read_to_string(&mut text)
            .ok()
            .map(|_| text);
    }
    String::from_utf8(bytes.to_vec()).ok()
}

impl WsTransport for TungsteniteWsTransport {
    fn connect(&self, url: &str) -> Result<Box<dyn WsConnection>> {
        let (inbound_tx, inbound_rx) = mpsc::channel::<Result<Option<String>>>();
        let (outbound_tx, mut outbound_rx) = unbounded_channel::<String>();
        let closed = Arc::new(AtomicBool::new(false));
        let closed_task = Arc::clone(&closed);
        let url = url.to_string();

        std::thread::spawn(move || {
            let runtime = match tokio::runtime::Builder::new_current_thread()
                .enable_all()
                .build()
            {
                Ok(runtime) => runtime,
                Err(e) => {
                    let _ = inbound_tx.send(Err(Error::Network(e.to_string())));
                    return;
                }
            };
            runtime.block_on(async move {
                let stream = match tokio_tungstenite::connect_async(url.as_str()).await {
                    Ok((stream, _response)) => stream,
                    Err(e) => {
                        let _ = inbound_tx.send(Err(Error::Network(e.to_string())));
                        return;
                    }
                };
                let (mut sink, mut source) = stream.split();
                loop {
                    if closed_task.load(Ordering::Relaxed) {
                        let _ = sink.close().await;
                        break;
                    }
                    tokio::select! {
                        incoming = source.next() => match incoming {
                            Some(Ok(Message::Text(text))) => {
                                // tungstenite 0.29 carries text as `Utf8Bytes`; the
                                // channel speaks owned `String`.
                                if inbound_tx.send(Ok(Some(text.to_string()))).is_err() {
                                    break;
                                }
                            }
                            Some(Ok(Message::Binary(bytes))) => {
                                if let Some(text) = decode_binary(&bytes) {
                                    if inbound_tx.send(Ok(Some(text))).is_err() {
                                        break;
                                    }
                                }
                            }
                            Some(Ok(Message::Ping(payload))) => {
                                let _ = sink.send(Message::Pong(payload)).await;
                            }
                            Some(Ok(Message::Close(_))) | None => {
                                let _ = inbound_tx.send(Ok(None));
                                break;
                            }
                            Some(Ok(_)) => {}
                            Some(Err(e)) => {
                                let _ = inbound_tx.send(Err(Error::Network(e.to_string())));
                                break;
                            }
                        },
                        outgoing = outbound_rx.recv() => {
                            if let Some(text) = outgoing {
                                if sink.send(Message::Text(text.into())).await.is_err() {
                                    break;
                                }
                            } else {
                                let _ = sink.close().await;
                                break;
                            }
                        }
                    }
                }
            });
        });

        Ok(Box::new(RealWsConnection {
            inbound: inbound_rx,
            outbound: outbound_tx,
            closed,
            peer_closed: false,
        }))
    }
}

struct RealWsConnection {
    inbound: Receiver<Result<Option<String>>>,
    outbound: UnboundedSender<String>,
    closed: Arc<AtomicBool>,
    /// Set once the peer closes the stream (a `None`/`Err` item, or the reader
    /// thread ending), so `is_connected` can report it for reconnect.
    peer_closed: bool,
}

impl WsConnection for RealWsConnection {
    fn send(&mut self, text: &str) -> Result<()> {
        self.outbound
            .send(text.to_string())
            .map_err(|_| Error::NotConnected)
    }

    fn recv(&mut self) -> Result<Option<String>> {
        match self.inbound.try_recv() {
            Ok(Ok(Some(text))) => Ok(Some(text)),
            Ok(Err(e)) => {
                self.peer_closed = true;
                Err(e)
            }
            Ok(Ok(None)) | Err(TryRecvError::Disconnected) => {
                self.peer_closed = true;
                Ok(None)
            }
            Err(TryRecvError::Empty) => Ok(None),
        }
    }

    fn is_connected(&self) -> bool {
        !self.peer_closed && !self.closed.load(Ordering::Relaxed)
    }

    fn close(&mut self) -> Result<()> {
        self.closed.store(true, Ordering::Relaxed);
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use wickra_exchange_core::MarketType;

    #[test]
    fn builds_from_options() {
        // Construction is deterministic and offline; the actual socket path is
        // covered by the gated integration test below.
        let opts = ExchangeOptions::mainnet(MarketType::Spot);
        assert!(ReqwestHttpTransport::new(&opts).is_ok());
    }

    /// A gzip frame becomes text; plain UTF-8 still passes; noise is dropped.
    ///
    /// HTX sends every frame gzip-compressed, and this arm forwarded a binary
    /// frame only when it decoded as UTF-8 -- which gzip never does. An HTX
    /// stream therefore delivered nothing at all, with no error, while the
    /// venue's unit tests passed on plain JSON through the mock transport.
    #[test]
    fn a_gzip_frame_is_decoded_and_plain_text_still_passes() {
        use flate2::write::GzEncoder;
        use flate2::Compression;
        use std::io::Write as _;

        let payload = r#"{"ch":"market.BTC-USDT.trade.detail","ts":1788462493057}"#;
        let mut encoder = GzEncoder::new(Vec::new(), Compression::default());
        encoder.write_all(payload.as_bytes()).unwrap();
        let compressed = encoder.finish().unwrap();

        assert_eq!(
            &compressed[..2],
            &[0x1f, 0x8b],
            "the fixture must actually be gzip, or this proves nothing"
        );
        assert!(
            String::from_utf8(compressed.clone()).is_err(),
            "gzip is not valid UTF-8, which is why the old arm dropped it"
        );

        assert_eq!(decode_binary(&compressed).as_deref(), Some(payload));
        assert_eq!(decode_binary(payload.as_bytes()).as_deref(), Some(payload));
        // Neither gzip nor UTF-8: there is nothing to hand a parser.
        assert!(decode_binary(&[0xff, 0xfe, 0xfd]).is_none());
    }

    #[test]
    #[ignore = "hits the network; run explicitly with --ignored"]
    fn live_request_reaches_binance() {
        let opts = ExchangeOptions::mainnet(MarketType::Spot);
        let transport = ReqwestHttpTransport::new(&opts).unwrap();
        let request = HttpRequest::get("https://api.binance.com/api/v3/time");
        let response = transport.execute(&request).unwrap();
        // Binance geo-restricts data-centre / CI-runner IP ranges (HTTP 451, or
        // 403 on some endpoints). When the venue is unreachable from this
        // network, skip rather than fail — this test checks for upstream API
        // drift, not the runner's location.
        if matches!(response.status, 451 | 403) {
            eprintln!(
                "skipping live_request_reaches_binance: venue restricted from this location (HTTP {})",
                response.status
            );
            return;
        }
        assert!(response.is_success());
        assert!(response.body.contains("serverTime"));
    }

    #[test]
    fn ws_transport_constructs() {
        let _transport = TungsteniteWsTransport::new();
    }

    #[test]
    #[ignore = "opens a live WebSocket; run explicitly with --ignored"]
    fn live_ws_receives_binance_trades() {
        use std::time::Duration;

        let transport = TungsteniteWsTransport::new();
        let mut connection = transport
            .connect("wss://stream.binance.com:9443/ws/btcusdt@trade")
            .unwrap();
        // The connect is lazy — a geo-restricted runner (Binance blocks
        // data-centre / CI IP ranges) surfaces the rejected handshake as an
        // error from recv(). Treat "venue unreachable" as a skip, not a
        // failure, so the nightly job stays green where Binance is blocked.
        let mut got = None;
        for _ in 0..50 {
            let frame = match connection.recv() {
                Ok(frame) => frame,
                Err(err) => {
                    eprintln!("skipping live_ws_receives_binance_trades: WS unreachable ({err})");
                    return;
                }
            };
            if let Some(frame) = frame {
                got = Some(frame);
                break;
            }
            std::thread::sleep(Duration::from_millis(200));
        }
        connection.close().unwrap();
        assert!(got.is_some_and(|f| f.contains("btcusdt") || f.contains("\"e\"")));
    }

    /// HTX's socket, over the real transport, delivering a real frame.
    ///
    /// This is the test that would have failed before the gzip decoder: every
    /// frame HTX sends is gzip-compressed, so `recv` returned nothing at all,
    /// for as long as anyone waited. The venue's own unit tests could not notice
    /// -- the mock transport hands the parser plain JSON, and a real frame never
    /// reached it.
    ///
    /// Unreachable is a skip, for the reason the Binance one gives.
    #[test]
    #[ignore = "opens a live WebSocket; run explicitly with --ignored"]
    fn live_ws_receives_htx_frames_that_arrive_gzipped() {
        use std::time::Duration;

        let transport = TungsteniteWsTransport::new();
        let mut connection = match transport.connect("wss://api.huobi.pro/ws") {
            Ok(connection) => connection,
            Err(err) => {
                eprintln!("skipping live HTX ws test: unreachable ({err})");
                return;
            }
        };
        if connection
            .send(r#"{"sub":"market.btcusdt.trade.detail","id":"1"}"#)
            .is_err()
        {
            eprintln!("skipping live HTX ws test: subscribe failed");
            return;
        }

        let mut got = None;
        for _ in 0..60 {
            let frame = match connection.recv() {
                Ok(frame) => frame,
                Err(err) => {
                    eprintln!("skipping live HTX ws test: WS unreachable ({err})");
                    return;
                }
            };
            if let Some(frame) = frame {
                got = Some(frame);
                break;
            }
            std::thread::sleep(Duration::from_millis(200));
        }
        connection.close().ok();

        let frame = got.expect(
            "HTX sent nothing readable in 12 seconds -- every frame it sends is \
             gzip, so this is what a missing decoder looks like",
        );
        assert!(
            frame.contains("\"ping\"") || frame.contains("market.btcusdt"),
            "unexpected frame: {frame}"
        );
    }
}