offline-cli 0.3.3

//! Localhost callback listener for the developer portal login redirect.

use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};

use anyhow::{anyhow, bail, Context, Result};
use base64::Engine;
use serde::Deserialize;
use tokio::sync::mpsc;

/// Fixed port used by CLI releases up to 0.2.3. Only a last-resort fallback now:
/// the listener binds an OS-assigned ephemeral port so two logins, or another
/// local program holding a well-known port, cannot collide with or intercept it.
pub const LEGACY_CALLBACK_PORT: u16 = 9876;
pub const ENV_CALLBACK_PORT: &str = "OFFLINE_CALLBACK_PORT";
pub const CALLBACK_PATH: &str = "/callback";

/// Largest callback POST body the listener reads. The real payload (a key,
/// a few ids and an email) is well under 1 KiB.
pub const MAX_CALLBACK_BODY: usize = 16 * 1024;
/// How long the listener waits for a callback POST body before giving up on
/// that request and serving the next one.
pub const CALLBACK_BODY_TIMEOUT: Duration = Duration::from_secs(5);

/// Validated credentials delivered by the portal on the CLI callback URL.
#[derive(Debug, Clone, PartialEq)]
pub struct CallbackPayload {
    pub api_key: String,
    pub app_id: String,
    pub dev_id: i32,
    pub email: String,
}

/// Developer authenticated in the portal but API key not yet selected.
#[derive(Debug, Clone, PartialEq)]
pub struct AuthCallbackPayload {
    pub dev_id: i32,
    pub email: String,
}

#[derive(Debug, Clone, PartialEq)]
pub enum CallbackMessage {
    DeveloperAuthenticated(AuthCallbackPayload),
    AuthorizationComplete(CallbackPayload),
}

#[derive(Debug, Deserialize)]
struct RawCallbackPayload {
    success: bool,
    #[serde(deserialize_with = "deserialize_api_key")]
    api_key: String,
    #[serde(default)]
    state: Option<String>,
    #[serde(default)]
    dev_id: Option<i32>,
    #[serde(
        default,
        alias = "projectId",
        alias = "applicationId",
        alias = "application_id"
    )]
    app_id: Option<String>,
    #[serde(default)]
    email: Option<String>,
    user: Option<RawCallbackUser>,
}

#[derive(Debug, Deserialize)]
struct RawCallbackUser {
    #[serde(default)]
    dev_id: Option<i32>,
    email: String,
}

#[derive(Deserialize)]
#[serde(untagged)]
enum ApiKeyField {
    Token(String),
    Object { token: String },
}

fn deserialize_api_key<'de, D>(deserializer: D) -> Result<String, D::Error>
where
    D: serde::Deserializer<'de>,
{
    match ApiKeyField::deserialize(deserializer)? {
        ApiKeyField::Token(token) => Ok(token),
        ApiKeyField::Object { token } => Ok(token),
    }
}

pub fn generate_callback_state() -> String {
    use rand::RngCore;
    let mut bytes = [0u8; 32];
    rand::thread_rng().fill_bytes(&mut bytes);
    base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(bytes)
}

/// Callback URL with `state` query param the portal must echo.
pub fn callback_url(port: u16, state: &str) -> String {
    format!(
        "http://127.0.0.1:{port}{CALLBACK_PATH}?state={}",
        urlencoding::encode(state)
    )
}

/// Port pinned with `OFFLINE_CALLBACK_PORT`, if any (for firewalled setups).
pub fn configured_callback_port() -> Option<u16> {
    std::env::var(ENV_CALLBACK_PORT)
        .ok()
        .and_then(|v| v.trim().parse::<u16>().ok())
        .filter(|p| *p != 0)
}

/// Ports to try in order. `0` asks the OS for a free ephemeral port.
fn bind_candidates(configured: Option<u16>) -> Vec<u16> {
    match configured {
        Some(port) => vec![port, 0],
        None => vec![0, LEGACY_CALLBACK_PORT],
    }
}

/// A bound (not yet serving) loopback callback listener.
pub struct BoundCallback {
    server: tiny_http::Server,
    port: u16,
    /// Set when the pinned `OFFLINE_CALLBACK_PORT` was busy and another port was used.
    pub fallback_from: Option<u16>,
}

impl BoundCallback {
    pub fn port(&self) -> u16 {
        self.port
    }

    pub fn url(&self, state: &str) -> String {
        callback_url(self.port, state)
    }
}

/// Bind the loopback callback listener before the login session is created,
/// so the callback URL sent to the portal names a port this process owns.
pub fn bind_callback() -> Result<BoundCallback> {
    bind_callback_with(configured_callback_port())
}

fn bind_callback_with(configured: Option<u16>) -> Result<BoundCallback> {
    let mut errors = Vec::new();
    for candidate in bind_candidates(configured) {
        match tiny_http::Server::http(("127.0.0.1", candidate)) {
            Ok(server) => {
                let port = server
                    .server_addr()
                    .to_ip()
                    .map(|a| a.port())
                    .ok_or_else(|| anyhow!("callback listener has no IP address"))?;
                let fallback_from = configured.filter(|p| *p != port);
                return Ok(BoundCallback {
                    server,
                    port,
                    fallback_from,
                });
            }
            Err(e) => {
                let label = if candidate == 0 {
                    "an ephemeral port".to_string()
                } else {
                    format!("port {candidate}")
                };
                errors.push(format!("{label}: {e}"));
            }
        }
    }
    bail!("binding callback listener failed ({})", errors.join("; "))
}

fn assert_callback_state(got: Option<&str>, expected: &str) -> Result<()> {
    match got {
        Some(s) if s == expected => Ok(()),
        Some(_) => bail!("callback state mismatch (possible CSRF)"),
        None => bail!("callback missing state"),
    }
}

pub fn parse_callback_request(
    url: &str,
    body: Option<&str>,
    expected_state: &str,
) -> Result<CallbackMessage> {
    if let Some(body) = body.filter(|b| !b.trim().is_empty()) {
        return parse_callback_body(body, expected_state);
    }
    parse_callback_query(url, expected_state)
}

pub fn parse_callback_body(body: &str, expected_state: &str) -> Result<CallbackMessage> {
    let raw: RawCallbackPayload =
        serde_json::from_str(body).context("parsing callback JSON payload")?;

    assert_callback_state(raw.state.as_deref(), expected_state)?;

    if !raw.success {
        bail!("authorization callback reported failure");
    }

    let email = raw
        .email
        .or_else(|| raw.user.as_ref().map(|u| u.email.clone()))
        .filter(|e| !e.trim().is_empty())
        .ok_or_else(|| anyhow!("callback missing email"))?;
    let dev_id = raw
        .dev_id
        .or_else(|| raw.user.as_ref().and_then(|u| u.dev_id))
        .filter(|id| *id > 0)
        .ok_or_else(|| anyhow!("callback missing dev_id"))?;
    let app_id = raw
        .app_id
        .filter(|id| !id.trim().is_empty())
        .ok_or_else(|| anyhow!("callback missing app_id"))?;

    validate_complete_fields(&raw.api_key, &app_id, dev_id, &email)?;

    Ok(CallbackMessage::AuthorizationComplete(CallbackPayload {
        api_key: raw.api_key,
        app_id,
        dev_id,
        email,
    }))
}

fn validate_complete_fields(api_key: &str, app_id: &str, dev_id: i32, email: &str) -> Result<()> {
    if api_key.trim().is_empty() {
        bail!("callback missing api_key");
    }
    if app_id.trim().is_empty() {
        bail!("callback missing app_id");
    }
    if dev_id <= 0 {
        bail!("callback missing dev_id");
    }
    if email.trim().is_empty() {
        bail!("callback missing email");
    }
    Ok(())
}

fn validate_auth_fields(dev_id: i32, email: &str) -> Result<()> {
    if dev_id <= 0 {
        bail!("callback missing dev_id");
    }
    if email.trim().is_empty() {
        bail!("callback missing email");
    }
    Ok(())
}

fn request_path(url: &str) -> &str {
    let without_query = url.split('?').next().unwrap_or(url);
    if let Some(scheme_end) = without_query.find("://") {
        let after_scheme = &without_query[scheme_end + 3..];
        if let Some(path_start) = after_scheme.find('/') {
            return &after_scheme[path_start..];
        }
        return "/";
    }
    without_query
}

fn matches_callback_path(url: &str) -> bool {
    request_path(url) == CALLBACK_PATH
}

fn handle_callback_get(url: &str, expected_state: &str) -> Result<CallbackMessage> {
    parse_callback_query(url, expected_state)
}

fn parse_callback_query(url: &str, expected_state: &str) -> Result<CallbackMessage> {
    let (_, query) = url.split_once('?').unwrap_or((url, ""));
    assert_callback_state(query_param_value(query, "state").as_deref(), expected_state)?;

    if let Some(encoded) = query_param_value(query, "payload") {
        return decode_payload_param(&encoded, expected_state);
    }

    let phase = query_param_value(query, "phase");

    if phase.as_deref() == Some("auth") {
        let dev_id = query_param_value(query, "dev_id")
            .and_then(|v| v.parse().ok())
            .ok_or_else(|| anyhow!("callback missing dev_id"))?;
        let email =
            query_param_value(query, "email").ok_or_else(|| anyhow!("callback missing email"))?;
        validate_auth_fields(dev_id, &email)?;
        return Ok(CallbackMessage::DeveloperAuthenticated(
            AuthCallbackPayload { dev_id, email },
        ));
    }

    let api_key =
        query_param_value(query, "api_key").ok_or_else(|| anyhow!("callback missing api_key"))?;
    let app_id = query_param_value(query, "app_id")
        .or_else(|| query_param_value(query, "application_id"))
        .or_else(|| query_param_value(query, "project_id"))
        .ok_or_else(|| anyhow!("callback missing app_id"))?;
    let email =
        query_param_value(query, "email").ok_or_else(|| anyhow!("callback missing email"))?;
    let dev_id = query_param_value(query, "dev_id")
        .and_then(|v| v.parse().ok())
        .ok_or_else(|| anyhow!("callback missing dev_id"))?;

    validate_complete_fields(&api_key, &app_id, dev_id, &email)?;
    Ok(CallbackMessage::AuthorizationComplete(CallbackPayload {
        api_key,
        app_id,
        dev_id,
        email,
    }))
}

fn query_param_value(query: &str, name: &str) -> Option<String> {
    query.split('&').find_map(|pair| {
        let (key, value) = pair.split_once('=')?;
        if key != name {
            return None;
        }
        urlencoding::decode(value).ok().map(|s| s.into_owned())
    })
}

fn decode_payload_param(encoded: &str, expected_state: &str) -> Result<CallbackMessage> {
    let bytes = base64::engine::general_purpose::URL_SAFE_NO_PAD
        .decode(encoded.as_bytes())
        .or_else(|_| {
            let padded = match encoded.len() % 4 {
                0 => encoded.to_string(),
                n => format!("{encoded}{}", "=".repeat(4 - n)),
            };
            base64::engine::general_purpose::URL_SAFE
                .decode(padded.as_bytes())
                .or_else(|_| base64::engine::general_purpose::STANDARD.decode(padded.as_bytes()))
        })
        .context("decoding callback payload")?;
    let body = String::from_utf8(bytes).context("callback payload is not utf-8")?;
    // Reuse outer `state` when the JSON body omits it.
    if body.contains("\"state\"") {
        parse_callback_body(&body, expected_state)
    } else {
        // Inject validated `state` before re-parsing body.
        let mut value: serde_json::Value =
            serde_json::from_str(&body).context("parsing callback payload JSON")?;
        if let Some(obj) = value.as_object_mut() {
            obj.insert(
                "state".to_string(),
                serde_json::Value::String(expected_state.to_string()),
            );
        }
        parse_callback_body(&value.to_string(), expected_state)
    }
}

pub struct CallbackListener {
    cancel: Arc<AtomicBool>,
}

impl Drop for CallbackListener {
    fn drop(&mut self) {
        self.cancel.store(true, Ordering::Relaxed);
    }
}

pub struct CallbackReceiver {
    rx: mpsc::UnboundedReceiver<CallbackMessage>,
}

impl CallbackReceiver {
    pub async fn recv(&mut self) -> Option<CallbackMessage> {
        self.rx.recv().await
    }
}

pub fn start_listener(
    bound: BoundCallback,
    deadline: Instant,
    expected_state: String,
) -> (CallbackListener, CallbackReceiver) {
    let cancel = Arc::new(AtomicBool::new(false));
    let server = bound.server;
    let (tx, rx) = mpsc::unbounded_channel();
    let cancel_for_task = cancel.clone();
    tokio::task::spawn_blocking(move || {
        listen_until(
            deadline,
            server,
            expected_state,
            cancel_for_task,
            tx,
            BodyLimits::default(),
        )
    });
    (CallbackListener { cancel }, CallbackReceiver { rx })
}

#[derive(Clone, Copy, Debug)]
struct BodyLimits {
    max_bytes: usize,
    timeout: Duration,
}

impl Default for BodyLimits {
    fn default() -> Self {
        Self {
            max_bytes: MAX_CALLBACK_BODY,
            timeout: CALLBACK_BODY_TIMEOUT,
        }
    }
}

/// Outcome of reading a callback POST body.
enum BodyRead {
    Body(Box<tiny_http::Request>, String),
    /// Answer the request with this status and message.
    Rejected(Box<tiny_http::Request>, u16, &'static str),
    /// The client did not send the body in time. The request stays with the
    /// reader thread, which ends when the client closes the connection.
    TimedOut,
}

/// Read a callback POST body with a size cap and a deadline. The read runs on
/// a helper thread so a client that stalls mid-body cannot block the
/// listener; after `limits.timeout` the listener moves on without it.
fn read_body_limited(request: tiny_http::Request, limits: BodyLimits) -> BodyRead {
    use std::io::Read;

    if request.body_length().is_some_and(|n| n > limits.max_bytes) {
        return BodyRead::Rejected(Box::new(request), 413, "callback body too large");
    }
    let (done_tx, done_rx) = std::sync::mpsc::channel();
    std::thread::spawn(move || {
        let mut request = request;
        let mut buf = Vec::new();
        let result = request
            .as_reader()
            .take(limits.max_bytes as u64 + 1)
            .read_to_end(&mut buf)
            .map(|_| buf);
        let _ = done_tx.send((Box::new(request), result));
    });
    match done_rx.recv_timeout(limits.timeout) {
        Ok((request, Ok(buf))) if buf.len() > limits.max_bytes => {
            BodyRead::Rejected(request, 413, "callback body too large")
        }
        Ok((request, Ok(buf))) => match String::from_utf8(buf) {
            Ok(body) => BodyRead::Body(request, body),
            Err(_) => BodyRead::Rejected(request, 400, "callback body is not UTF-8"),
        },
        Ok((request, Err(_))) => BodyRead::Rejected(request, 400, "could not read callback body"),
        Err(_) => BodyRead::TimedOut,
    }
}

fn respond_text(request: tiny_http::Request, status: u16, text: &str) {
    let response = tiny_http::Response::from_string(text)
        .with_status_code(status)
        .with_header(
            tiny_http::Header::from_bytes(&b"Content-Type"[..], &b"text/plain"[..]).unwrap(),
        );
    let _ = request.respond(response);
}

fn listen_until(
    deadline: Instant,
    server: tiny_http::Server,
    expected_state: String,
    cancel: Arc<AtomicBool>,
    tx: mpsc::UnboundedSender<CallbackMessage>,
    limits: BodyLimits,
) -> Result<()> {
    while Instant::now() < deadline && !cancel.load(Ordering::Relaxed) {
        let remaining = deadline
            .checked_duration_since(Instant::now())
            .unwrap_or(Duration::ZERO);
        let timeout = remaining.min(Duration::from_millis(500));

        let mut request: tiny_http::Request = match server.recv_timeout(timeout) {
            Ok(Some(request)) => request,
            Ok(None) => continue,
            Err(e) => bail!("callback listener error: {e}"),
        };

        let full_url = request.url().to_string();

        if !matches_callback_path(&full_url) {
            let response = tiny_http::Response::from_string("not found")
                .with_status_code(404)
                .with_header(
                    tiny_http::Header::from_bytes(&b"Content-Type"[..], &b"text/plain"[..])
                        .unwrap(),
                );
            let _ = request.respond(response);
            continue;
        }

        let payload_result = match *request.method() {
            tiny_http::Method::Get | tiny_http::Method::Head => {
                handle_callback_get(&full_url, &expected_state)
            }
            tiny_http::Method::Post => {
                // A bad body is answered and skipped; it never stops the
                // listener, so a stray local request cannot end a login.
                let body = match read_body_limited(request, limits) {
                    BodyRead::Body(req, body) => {
                        request = *req;
                        body
                    }
                    BodyRead::Rejected(req, status, text) => {
                        respond_text(*req, status, text);
                        continue;
                    }
                    BodyRead::TimedOut => continue,
                };
                if body.trim().is_empty() {
                    parse_callback_query(&full_url, &expected_state)
                } else {
                    let (_, query) = full_url.split_once('?').unwrap_or((&full_url, ""));
                    if query_param_value(query, "state").is_some() {
                        parse_callback_query(&full_url, &expected_state).or_else(|_| {
                            parse_callback_request(&full_url, Some(&body), &expected_state)
                        })
                    } else {
                        parse_callback_request(&full_url, Some(&body), &expected_state)
                    }
                }
            }
            _ => {
                let response =
                    tiny_http::Response::from_string("method not allowed — expected GET or POST")
                        .with_status_code(405)
                        .with_header(
                            tiny_http::Header::from_bytes(&b"Content-Type"[..], &b"text/plain"[..])
                                .unwrap(),
                        );
                let _ = request.respond(response);
                continue;
            }
        };

        match payload_result {
            Ok(message) => {
                let is_complete = matches!(message, CallbackMessage::AuthorizationComplete(_));
                if *request.method() == tiny_http::Method::Head {
                    let _ = request.respond(tiny_http::Response::empty(200));
                } else {
                    let html = if is_complete {
                        "<html><body><p>Successfully logged in. \
                         You can close this tab and return to the terminal.</p></body></html>"
                    } else {
                        "<html><body><p>Developer authenticated. \
                         Select an API key in the browser to finish.</p></body></html>"
                    };
                    let response = tiny_http::Response::from_string(html)
                        .with_status_code(200)
                        .with_header(
                            tiny_http::Header::from_bytes(&b"Content-Type"[..], &b"text/html"[..])
                                .unwrap(),
                        );
                    let _ = request.respond(response);
                }

                let _ = tx.send(message);
                if is_complete {
                    cancel.store(true, Ordering::Relaxed);
                    return Ok(());
                }
            }
            Err(e) => {
                let response = tiny_http::Response::from_string(format!("bad request: {e:#}"))
                    .with_status_code(400)
                    .with_header(
                        tiny_http::Header::from_bytes(&b"Content-Type"[..], &b"text/plain"[..])
                            .unwrap(),
                    );
                let _ = request.respond(response);
            }
        }
    }

    if cancel.load(Ordering::Relaxed) {
        bail!("callback listener stopped");
    }
    bail!("callback timed out waiting for portal response")
}

#[cfg(test)]
#[allow(clippy::await_holding_lock)]
mod tests {
    use super::*;
    use crate::config::storage::test_sync::env_lock;
    use std::net::TcpListener;
    use std::sync::{Mutex, MutexGuard};

    /// Loopback listener tests must not run in parallel: they share fixed fallback
    /// port 9876 and tiny_http can stall on Windows when connections overlap.
    static LISTENER_TEST_LOCK: Mutex<()> = Mutex::new(());

    fn listener_test_lock() -> MutexGuard<'static, ()> {
        LISTENER_TEST_LOCK.lock().unwrap_or_else(|e| e.into_inner())
    }

    #[test]
    fn parses_auth_phase_query() {
        let url = "/callback?state=csrf-token&phase=auth&dev_id=42&email=dev%40example.com";
        let payload = match parse_callback_query(url, "csrf-token").unwrap() {
            CallbackMessage::DeveloperAuthenticated(payload) => payload,
            _ => panic!("expected auth payload"),
        };
        assert_eq!(payload.dev_id, 42);
        assert_eq!(payload.email, "dev@example.com");
    }

    #[test]
    fn parses_complete_query_without_user_id() {
        let url = "/callback?state=csrf-token&phase=complete&api_key=550e8400-e29b-41d4-a716-446655440000&project_id=proj_abc&dev_id=42&email=dev%40example.com";
        let payload = match parse_callback_query(url, "csrf-token").unwrap() {
            CallbackMessage::AuthorizationComplete(payload) => payload,
            _ => panic!("expected complete payload"),
        };
        assert_eq!(payload.dev_id, 42);
        assert_eq!(payload.app_id, "proj_abc");
        assert_eq!(payload.email, "dev@example.com");
    }

    #[test]
    fn rejects_missing_or_mismatched_state() {
        let url = "/callback?phase=auth&dev_id=42&email=dev%40example.com";
        assert!(parse_callback_query(url, "csrf-token").is_err());
        let bad = "/callback?state=wrong&phase=auth&dev_id=42&email=dev%40example.com";
        assert!(parse_callback_query(bad, "csrf-token").is_err());
    }

    fn wait_for_listener(port: u16) {
        for _ in 0..100 {
            if std::net::TcpStream::connect(("127.0.0.1", port)).is_ok() {
                return;
            }
            std::thread::sleep(Duration::from_millis(25));
        }
        panic!("callback listener did not start on port {port}");
    }

    #[test]
    fn binds_an_ephemeral_port_by_default() {
        let _guard = listener_test_lock();
        let a = bind_callback_with(None).unwrap();
        let b = bind_callback_with(None).unwrap();
        assert_ne!(a.port(), 0);
        assert_ne!(a.port(), b.port(), "two logins must not share a port");
        assert!(a.fallback_from.is_none());
        assert!(a
            .url("s")
            .starts_with(&format!("http://127.0.0.1:{}/callback?state=", a.port())));
    }

    #[test]
    fn honours_pinned_port_and_falls_back_when_busy() {
        let _guard = listener_test_lock();
        let free = {
            let listener = TcpListener::bind("127.0.0.1:0").unwrap();
            listener.local_addr().unwrap().port()
        };
        let pinned = bind_callback_with(Some(free)).unwrap();
        assert_eq!(pinned.port(), free);
        assert!(pinned.fallback_from.is_none());

        // The pinned port is now held by `pinned`, so a second bind falls back.
        let fallback = bind_callback_with(Some(free)).unwrap();
        assert_ne!(fallback.port(), free);
        assert_eq!(fallback.fallback_from, Some(free));
    }

    #[test]
    fn configured_port_ignores_invalid_values() {
        let _lock = env_lock();
        std::env::set_var(ENV_CALLBACK_PORT, "not-a-port");
        assert_eq!(configured_callback_port(), None);
        std::env::set_var(ENV_CALLBACK_PORT, "0");
        assert_eq!(configured_callback_port(), None);
        std::env::set_var(ENV_CALLBACK_PORT, " 43123 ");
        assert_eq!(configured_callback_port(), Some(43123));
        std::env::remove_var(ENV_CALLBACK_PORT);
        assert_eq!(configured_callback_port(), None);
        assert_eq!(bind_candidates(None), vec![0, LEGACY_CALLBACK_PORT]);
    }

    async fn recv_callback(rx: &mut CallbackReceiver) -> CallbackMessage {
        tokio::time::timeout(Duration::from_secs(10), rx.recv())
            .await
            .expect("timed out waiting for callback message")
            .expect("callback channel closed before message")
    }

    #[tokio::test]
    async fn receives_auth_then_complete_callbacks() {
        let port;
        let state = "test-csrf-state".to_string();
        let (_listener, mut receiver) = {
            let _serial = listener_test_lock();
            let bound = bind_callback_with(None).unwrap();
            port = bound.port();
            let deadline = Instant::now() + Duration::from_secs(5);
            let handles = start_listener(bound, deadline, state.clone());
            wait_for_listener(port);
            handles
        };

        let client = reqwest::Client::builder()
            .timeout(Duration::from_secs(5))
            .build()
            .unwrap();
        let base = format!("http://127.0.0.1:{port}");
        let auth = client
            .get(format!(
                "{base}{CALLBACK_PATH}?state={state}&phase=auth&dev_id=42&email=dev%40example.com"
            ))
            .send()
            .await
            .unwrap();
        assert!(auth.status().is_success());
        assert!(matches!(
            recv_callback(&mut receiver).await,
            CallbackMessage::DeveloperAuthenticated(_)
        ));

        let complete = client
            .get(format!(
                "{base}{CALLBACK_PATH}?state={state}&api_key=550e8400-e29b-41d4-a716-446655440000&project_id=proj_abc&dev_id=42&email=dev%40example.com"
            ))
            .send()
            .await
            .unwrap();
        assert!(complete.status().is_success());

        let payload = match recv_callback(&mut receiver).await {
            CallbackMessage::AuthorizationComplete(payload) => payload,
            _ => panic!("expected complete payload"),
        };
        assert_eq!(payload.dev_id, 42);
        assert_eq!(payload.app_id, "proj_abc");
        assert_eq!(payload.email, "dev@example.com");
    }

    fn raw_request(port: u16, request: &[u8]) -> String {
        use std::io::{Read, Write};
        let mut stream = std::net::TcpStream::connect(("127.0.0.1", port)).unwrap();
        stream
            .set_read_timeout(Some(Duration::from_secs(5)))
            .unwrap();
        stream.write_all(request).unwrap();
        let mut response = String::new();
        let _ = stream.read_to_string(&mut response);
        response
    }

    fn start_test_listener(
        limits: BodyLimits,
    ) -> (
        u16,
        String,
        Arc<AtomicBool>,
        mpsc::UnboundedReceiver<CallbackMessage>,
    ) {
        let bound = bind_callback_with(None).unwrap();
        let port = bound.port();
        let state = "limit-test-state".to_string();
        let cancel = Arc::new(AtomicBool::new(false));
        let (tx, rx) = mpsc::unbounded_channel();
        let (server, st, c) = (bound.server, state.clone(), cancel.clone());
        std::thread::spawn(move || {
            let _ = listen_until(
                Instant::now() + Duration::from_secs(20),
                server,
                st,
                c,
                tx,
                limits,
            );
        });
        wait_for_listener(port);
        (port, state, cancel, rx)
    }

    fn complete_query(state: &str) -> String {
        format!(
            "{CALLBACK_PATH}?state={state}&api_key=550e8400-e29b-41d4-a716-446655440000&project_id=proj_abc&dev_id=42&email=dev%40example.com"
        )
    }

    /// The listener runs on a helper thread; on Windows the channel message can
    /// arrive slightly after the HTTP response is read.
    fn pop_complete_message(rx: &mut mpsc::UnboundedReceiver<CallbackMessage>) -> CallbackMessage {
        for _ in 0..200 {
            if let Ok(msg) = rx.try_recv() {
                return msg;
            }
            std::thread::sleep(Duration::from_millis(10));
        }
        panic!("expected callback message on channel after successful GET");
    }

    #[test]
    fn oversized_callback_body_is_refused_and_listener_keeps_serving() {
        let _guard = listener_test_lock();
        let limits = BodyLimits {
            max_bytes: 1024,
            timeout: Duration::from_secs(2),
        };
        let (port, state, cancel, mut rx) = start_test_listener(limits);

        // Declared too large: refused from the header alone.
        let declared = format!(
            "POST {CALLBACK_PATH} HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Length: 100000\r\nConnection: close\r\n\r\n"
        );
        assert!(raw_request(port, declared.as_bytes()).starts_with("HTTP/1.1 413"));

        // Chunked with no length: refused once the cap is passed.
        let chunk = "x".repeat(4096);
        let chunked = format!(
            "POST {CALLBACK_PATH} HTTP/1.1\r\nHost: 127.0.0.1\r\nTransfer-Encoding: chunked\r\nConnection: close\r\n\r\n{:x}\r\n{chunk}\r\n0\r\n\r\n",
            chunk.len()
        );
        assert!(raw_request(port, chunked.as_bytes()).starts_with("HTTP/1.1 413"));

        // The login still completes afterwards.
        let get = format!(
            "GET {} HTTP/1.1\r\nHost: 127.0.0.1\r\nConnection: close\r\n\r\n",
            complete_query(&state)
        );
        assert!(raw_request(port, get.as_bytes()).starts_with("HTTP/1.1 200"));
        assert!(matches!(
            pop_complete_message(&mut rx),
            CallbackMessage::AuthorizationComplete(_)
        ));
        cancel.store(true, Ordering::Relaxed);
    }

    #[test]
    fn stalled_callback_body_times_out_and_listener_keeps_serving() {
        use std::io::Write;
        let _guard = listener_test_lock();
        let limits = BodyLimits {
            max_bytes: 8 * 1024,
            timeout: Duration::from_millis(300),
        };
        let (port, state, cancel, mut rx) = start_test_listener(limits);

        // Promise 4000 bytes (above the size tiny_http buffers itself, below
        // the cap), send 10, then stall with the connection open.
        let mut stalled = std::net::TcpStream::connect(("127.0.0.1", port)).unwrap();
        let _ = stalled.set_read_timeout(Some(Duration::from_secs(2)));
        let _ = stalled.set_write_timeout(Some(Duration::from_secs(2)));
        stalled
            .write_all(
                format!(
                    "POST {CALLBACK_PATH} HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Length: 4000\r\n\r\n0123456789"
                )
                .as_bytes(),
            )
            .unwrap();
        // Let the listener pick up the stalled request first.
        std::thread::sleep(Duration::from_millis(150));

        let started = Instant::now();
        let get = format!(
            "GET {} HTTP/1.1\r\nHost: 127.0.0.1\r\nConnection: close\r\n\r\n",
            complete_query(&state)
        );
        assert!(raw_request(port, get.as_bytes()).starts_with("HTTP/1.1 200"));
        let waited = started.elapsed();
        assert!(
            waited >= Duration::from_millis(100),
            "the stalled request should have been read first (waited {waited:?})"
        );
        assert!(
            waited < Duration::from_secs(4),
            "a stalled body must not hold the listener (waited {waited:?})"
        );
        assert!(matches!(
            pop_complete_message(&mut rx),
            CallbackMessage::AuthorizationComplete(_)
        ));
        drop(stalled);
        cancel.store(true, Ordering::Relaxed);
    }
}