bitculator 1.1.0

//! A tiny single-threaded HTTP/1.1 stub server on a std TcpListener — the SDK's
//! transport is reqwest with no injection seam, so tests talk to a real local
//! socket instead of mocks. Zero dev-dependencies.

use std::collections::HashMap;
use std::io::{Read, Write};
use std::net::{TcpListener, TcpStream};
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::Duration;

pub struct Response {
    status: u16,
    body: String,
    headers: Vec<(String, String)>,
    delay: Duration,
}

impl Response {
    pub fn json(status: u16, body: &str) -> Self {
        Response { status, body: body.to_string(), headers: Vec::new(), delay: Duration::ZERO }
    }

    pub fn error(status: u16) -> Self {
        Self::json(
            status,
            &format!(r#"{{"error":{{"code":"err_{status}","message":"stub error {status}","details":null}}}}"#),
        )
    }

    pub fn ok() -> Self {
        Self::json(200, r#"{"data":"ok"}"#)
    }

    pub fn with_header(mut self, name: &str, value: &str) -> Self {
        self.headers.push((name.to_string(), value.to_string()));
        self
    }

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

#[derive(Clone, Debug)]
pub struct Request {
    pub method: String,
    pub target: String,
    pub headers: HashMap<String, String>,
    pub body: String,
}

impl Request {
    pub fn header(&self, name: &str) -> Option<&str> {
        self.headers.get(&name.to_lowercase()).map(String::as_str)
    }

    pub fn path(&self) -> &str {
        self.target.split('?').next().unwrap_or("")
    }

    pub fn query(&self) -> &str {
        self.target.split_once('?').map_or("", |(_, q)| q)
    }

    pub fn has_query_pair(&self, pair: &str) -> bool {
        self.query().split('&').any(|part| part == pair)
    }
}

pub struct Server {
    pub base_url: String,
    requests: Arc<Mutex<Vec<Request>>>,
}

impl Server {
    /// Serve the queued responses, one per connection, recording each request.
    pub fn start(responses: Vec<Response>) -> Server {
        let listener = TcpListener::bind("127.0.0.1:0").expect("bind stub server");
        let addr = listener.local_addr().expect("stub server addr");
        let requests: Arc<Mutex<Vec<Request>>> = Arc::new(Mutex::new(Vec::new()));
        let recorded = requests.clone();
        thread::spawn(move || {
            for response in responses {
                let Ok((mut stream, _)) = listener.accept() else { return };
                if let Some(request) = read_request(&mut stream) {
                    recorded.lock().unwrap().push(request);
                }
                if !response.delay.is_zero() {
                    thread::sleep(response.delay);
                }
                let mut head = format!(
                    "HTTP/1.1 {} Stub\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n",
                    response.status,
                    response.body.len(),
                );
                for (name, value) in &response.headers {
                    head.push_str(&format!("{name}: {value}\r\n"));
                }
                head.push_str("\r\n");
                let _ = stream.write_all(head.as_bytes());
                let _ = stream.write_all(response.body.as_bytes());
            }
        });
        Server { base_url: format!("http://{addr}"), requests }
    }

    pub fn hits(&self) -> usize {
        self.requests.lock().unwrap().len()
    }

    /// Wait until `expected` requests were recorded (the single-threaded stub
    /// may still be sleeping through a delayed response when the client has
    /// already errored out). Panics after 3s.
    pub fn wait_for_hits(&self, expected: usize) {
        let deadline = std::time::Instant::now() + Duration::from_secs(3);
        while self.hits() < expected {
            if std::time::Instant::now() > deadline {
                panic!("expected {expected} hits, saw {}", self.hits());
            }
            thread::sleep(Duration::from_millis(20));
        }
    }

    pub fn request(&self, index: usize) -> Request {
        self.requests.lock().unwrap()[index].clone()
    }
}

fn read_request(stream: &mut TcpStream) -> Option<Request> {
    stream.set_read_timeout(Some(Duration::from_secs(5))).ok()?;
    let mut buf: Vec<u8> = Vec::new();
    let mut chunk = [0u8; 4096];
    let header_end = loop {
        let n = stream.read(&mut chunk).ok()?;
        if n == 0 {
            return None;
        }
        buf.extend_from_slice(&chunk[..n]);
        if let Some(pos) = find(&buf, b"\r\n\r\n") {
            break pos + 4;
        }
    };

    let head = String::from_utf8_lossy(&buf[..header_end]).to_string();
    let mut lines = head.split("\r\n");
    let request_line = lines.next()?;
    let mut parts = request_line.split(' ');
    let method = parts.next()?.to_string();
    let target = parts.next()?.to_string();

    let mut headers = HashMap::new();
    for line in lines {
        if let Some((name, value)) = line.split_once(':') {
            headers.insert(name.trim().to_lowercase(), value.trim().to_string());
        }
    }

    let content_length: usize = headers.get("content-length").and_then(|v| v.parse().ok()).unwrap_or(0);
    while buf.len() < header_end + content_length {
        let n = stream.read(&mut chunk).ok()?;
        if n == 0 {
            break;
        }
        buf.extend_from_slice(&chunk[..n]);
    }
    let body = String::from_utf8_lossy(&buf[header_end..(header_end + content_length).min(buf.len())]).to_string();

    Some(Request { method, target, headers, body })
}

fn find(haystack: &[u8], needle: &[u8]) -> Option<usize> {
    haystack.windows(needle.len()).position(|window| window == needle)
}