billdogeng 1.0.0-beta.1

//! A minimal local HTTP stub server for resilience / round-trip tests
//! (permitted by the spec for these tests only). Mirrors node's `stub-server.ts`.
//!
//! Uses only the standard library: a background thread accepts connections,
//! parses the request line + headers + body (gunzipping when needed), records
//! it, and hands it to a user-supplied handler that returns a response body.

use std::io::{BufRead, BufReader, Read, Write};
use std::net::{TcpListener, TcpStream};
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::thread;

use flate2::read::GzDecoder;
use serde_json::Value;

/// A captured request.
#[derive(Debug, Clone)]
pub struct CapturedRequest {
    #[allow(dead_code)]
    pub method: String,
    pub path: String,
    pub headers: Vec<(String, String)>,
    pub body: Option<Value>,
}

impl CapturedRequest {
    /// Lowercased header lookup.
    pub fn header(&self, name: &str) -> Option<&str> {
        let name = name.to_ascii_lowercase();
        self.headers
            .iter()
            .find(|(k, _)| k.to_ascii_lowercase() == name)
            .map(|(_, v)| v.as_str())
    }
}

/// A stub response.
pub struct StubResponse {
    pub status: u16,
    pub body: String,
}

impl StubResponse {
    /// `{ success, data, meta }` envelope (status 200 by default).
    pub fn envelope(data: Value) -> Self {
        Self::envelope_status(data, 200)
    }

    /// `{ success, data, meta }` envelope at a specific status.
    pub fn envelope_status(data: Value, status: u16) -> Self {
        let body = serde_json::json!({
            "success": status < 400,
            "data": data,
            "meta": { "request_id": "stub" },
        });
        Self {
            status,
            body: body.to_string(),
        }
    }

    /// Raw (legacy) JSON body.
    pub fn raw(body: Value) -> Self {
        Self::raw_status(body, 200)
    }

    /// Raw JSON body at a specific status.
    pub fn raw_status(body: Value, status: u16) -> Self {
        Self {
            status,
            body: body.to_string(),
        }
    }
}

/// Handle to a running stub server.
pub struct StubServer {
    pub url: String,
    requests: Arc<Mutex<Vec<CapturedRequest>>>,
    running: Arc<AtomicBool>,
    handle: Option<thread::JoinHandle<()>>,
}

impl StubServer {
    /// All captured requests so far.
    pub fn requests(&self) -> Vec<CapturedRequest> {
        self.requests.lock().unwrap().clone()
    }

    /// Captured requests filtered by exact path.
    pub fn requests_for(&self, path: &str) -> Vec<CapturedRequest> {
        self.requests()
            .into_iter()
            .filter(|r| r.path == path)
            .collect()
    }
}

impl Drop for StubServer {
    fn drop(&mut self) {
        self.running.store(false, Ordering::SeqCst);
        // Nudge the accept loop with a throwaway connection.
        if let Ok(addr) = self.url.trim_start_matches("http://").parse::<std::net::SocketAddr>() {
            let _ = TcpStream::connect(addr);
        }
        if let Some(h) = self.handle.take() {
            let _ = h.join();
        }
    }
}

/// Start a stub server. The handler receives the parsed request and a 0-based
/// hit counter (so a test can vary behavior per attempt, e.g. 503 then 200).
pub fn start_stub<F>(handler: F) -> StubServer
where
    F: Fn(&CapturedRequest, usize) -> StubResponse + Send + Sync + 'static,
{
    let listener = TcpListener::bind("127.0.0.1:0").expect("bind stub server");
    let addr = listener.local_addr().unwrap();
    let url = format!("http://127.0.0.1:{}", addr.port());

    let requests = Arc::new(Mutex::new(Vec::new()));
    let running = Arc::new(AtomicBool::new(true));
    let hits = Arc::new(AtomicUsize::new(0));
    let handler = Arc::new(handler);

    let requests_t = requests.clone();
    let running_t = running.clone();

    let handle = thread::spawn(move || {
        for stream in listener.incoming() {
            if !running_t.load(Ordering::SeqCst) {
                break;
            }
            let stream = match stream {
                Ok(s) => s,
                Err(_) => continue,
            };
            let hit = hits.fetch_add(1, Ordering::SeqCst);
            if let Some(req) = handle_connection(&stream, &requests_t) {
                let resp = handler(&req, hit);
                write_response(&stream, &resp);
            }
        }
    });

    StubServer {
        url,
        requests,
        running,
        handle: Some(handle),
    }
}

fn handle_connection(
    stream: &TcpStream,
    requests: &Arc<Mutex<Vec<CapturedRequest>>>,
) -> Option<CapturedRequest> {
    let mut reader = BufReader::new(stream.try_clone().ok()?);

    // Request line.
    let mut line = String::new();
    if reader.read_line(&mut line).ok()? == 0 {
        return None;
    }
    let mut parts = line.split_whitespace();
    let method = parts.next().unwrap_or("GET").to_string();
    let path = parts.next().unwrap_or("/").to_string();

    // Headers.
    let mut headers: Vec<(String, String)> = Vec::new();
    let mut content_length = 0usize;
    let mut gzipped = false;
    loop {
        let mut h = String::new();
        if reader.read_line(&mut h).ok()? == 0 {
            break;
        }
        let trimmed = h.trim_end();
        if trimmed.is_empty() {
            break;
        }
        if let Some((k, v)) = trimmed.split_once(':') {
            let k = k.trim().to_string();
            let v = v.trim().to_string();
            let kl = k.to_ascii_lowercase();
            if kl == "content-length" {
                content_length = v.parse().unwrap_or(0);
            } else if kl == "content-encoding" && v.to_ascii_lowercase().contains("gzip") {
                gzipped = true;
            }
            headers.push((k, v));
        }
    }

    // Body.
    let mut raw = vec![0u8; content_length];
    if content_length > 0 {
        reader.read_exact(&mut raw).ok()?;
    }
    if gzipped && raw.len() > 1 && raw[0] == 0x1f && raw[1] == 0x8b {
        let mut dec = GzDecoder::new(&raw[..]);
        let mut out = Vec::new();
        if dec.read_to_end(&mut out).is_ok() {
            raw = out;
        }
    }

    let body = if raw.is_empty() {
        None
    } else {
        let text = String::from_utf8_lossy(&raw);
        serde_json::from_str::<Value>(&text)
            .ok()
            .or_else(|| Some(Value::String(text.to_string())))
    };

    let captured = CapturedRequest {
        method,
        path,
        headers,
        body,
    };
    requests.lock().unwrap().push(captured.clone());
    Some(captured)
}

fn write_response(mut stream: &TcpStream, resp: &StubResponse) {
    let reason = match resp.status {
        200 => "OK",
        400 => "Bad Request",
        404 => "Not Found",
        500 => "Internal Server Error",
        503 => "Service Unavailable",
        _ => "Status",
    };
    let out = format!(
        "HTTP/1.1 {} {}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
        resp.status,
        reason,
        resp.body.as_bytes().len(),
        resp.body
    );
    let _ = stream.write_all(out.as_bytes());
    let _ = stream.flush();
}