//! The loopback HTTP plumbing the four mocks sit on.
//!
//! One listener per service on a port the kernel picks, so nothing here
//! collides with another test, another suite running in parallel, or
//! anything else on the machine. Bound before the base URL is handed out, so
//! a command can never race the listener into existence.
//!
//! hyper's server half rather than a hand-rolled socket reader: reqwest —
//! which is what atgc uses for every request — is built on hyper's client
//! half, and the parts of HTTP/1.1 that a hand-rolled server gets wrong are
//! exactly the parts a real client exercises without asking (keep-alive,
//! chunked bodies, `Expect: 100-continue`). It is already in the tree.

use std::convert::Infallible;
use std::net::SocketAddr;
use std::pin::Pin;
use std::sync::{Arc, Mutex};
use std::task::{Context, Poll};

use http_body_util::combinators::BoxBody;
use http_body_util::{BodyExt, Full};
use hyper::body::{Bytes, Frame};
use hyper::service::service_fn;
use hyper::{Request, Response, StatusCode};
use hyper_util::rt::TokioIo;

use super::world::World;

/// Every reply is boxed so one of them can be a stream.
///
/// `Full<Bytes>` knows its own length, and hyper turns that into a
/// `Content-Length` header. That is right for the four ordinary replies and
/// useless for [`Reply::Flood`], whose entire point is to arrive without
/// one — so the body type is erased and hyper decides per response.
type Body = BoxBody<Bytes, Infallible>;

/// A request, flattened into the parts a mock cares about.
pub struct Incoming {
    pub method: hyper::Method,
    /// Path with no leading slash.
    pub path: String,
    pub params: std::collections::BTreeMap<String, String>,
    pub body: Vec<u8>,
    /// The `Authorization` header's credential, with the scheme removed.
    pub token: Option<String>,
}

impl Incoming {
    pub fn param(&self, name: &str) -> Option<&str> {
        self.params.get(name).map(String::as_str)
    }

    /// The request body as JSON, or `Null` when there is none.
    pub fn json(&self) -> serde_json::Value {
        serde_json::from_slice(&self.body).unwrap_or(serde_json::Value::Null)
    }

    /// The XRPC method this path names, if it is an XRPC path at all.
    pub fn nsid(&self) -> Option<&str> {
        self.path.strip_prefix("xrpc/")
    }
}

/// What a mock answers with.
pub enum Reply {
    Json(serde_json::Value),
    /// JSON under a status other than 200. RFC 9126 has a pushed
    /// authorization request answer `201 Created`, and jacquard checks.
    JsonStatus(StatusCode, serde_json::Value),
    Bytes(Vec<u8>),
    /// An XRPC error: status, `error` name, and a message.
    Error(StatusCode, &'static str, String),
    /// This many bytes, chunked, with no `Content-Length` — a host that
    /// answers with more than atgc will read. See [`World::flood`].
    Flood(usize),
    /// A `302` to `location`. The appview publishes the repo-DID-to-owner
    /// mapping this way and nowhere else, so a mock that cannot redirect
    /// cannot answer the one question `resolve::owner_of` asks.
    Redirect(String),
}

impl Reply {
    pub fn ok(value: serde_json::Value) -> Reply {
        Reply::Json(value)
    }

    pub fn not_found(message: impl Into<String>) -> Reply {
        Reply::Error(StatusCode::NOT_FOUND, "RecordNotFound", message.into())
    }

    pub fn bad(error: &'static str, message: impl Into<String>) -> Reply {
        Reply::Error(StatusCode::BAD_REQUEST, error, message.into())
    }

    /// A knot turning down a call it understood, in the knot's own spelling:
    /// 401, tagged `AccessControl`.
    pub fn denied(message: impl Into<String>) -> Reply {
        Reply::Error(StatusCode::UNAUTHORIZED, "AccessControl", message.into())
    }

    /// A knot refusing because the thing is already there — the 409 that
    /// `sh.tangled.repo.create` answers a taken name with. Its own
    /// constructor because `exit::from_status` classifies it apart from every
    /// other refusal, and a test that could not produce one could not check
    /// that classification.
    pub fn conflict(message: impl Into<String>) -> Reply {
        Reply::Error(StatusCode::CONFLICT, "RepoExists", message.into())
    }
}

/// A bound service: the base URL to point atgc at, and the task serving it.
pub struct Service {
    pub base: String,
}

/// The one runtime every mock in this test binary runs on.
///
/// Shared rather than one per scenario, and that is not a micro-optimization:
/// cargo runs integration tests on one thread per test, so a runtime per
/// scenario meant dozens of tokio worker threads competing with dozens of
/// `atgc` child processes on the same machine. Under that load a loopback
/// request could go unanswered long enough for a command to give up, and the
/// suite failed in a way that looked like a bug in the tool — a DID that
/// would not resolve — rather than like the harness running out of room.
///
/// Nothing is lost by sharing it. The listeners are independent, the state
/// behind them is per-scenario, and it never shuts down because the process
/// exits when the last test does.
fn runtime() -> &'static tokio::runtime::Runtime {
    static RUNTIME: std::sync::OnceLock<tokio::runtime::Runtime> = std::sync::OnceLock::new();
    RUNTIME.get_or_init(|| {
        tokio::runtime::Builder::new_multi_thread()
            .worker_threads(2)
            .enable_all()
            .build()
            .expect("a runtime for the mocks")
    })
}

/// Bind a listener and serve `handler` on it for the rest of the process.
///
/// `handler` is synchronous and takes the world's lock for the length of one
/// request, which serializes one scenario's mocks. That is the right trade
/// here: no command atgc runs depends on two of its own requests being in
/// flight at once, and a journal written under one lock is a journal whose
/// order is the order the requests actually arrived in.
pub fn serve<F>(world: Arc<Mutex<World>>, handler: F) -> std::io::Result<Service>
where
    F: Fn(&mut World, &Incoming) -> Reply + Send + Sync + 'static,
{
    let rt = runtime();
    let handler = Arc::new(handler);
    // Bound synchronously so the port is real before this returns.
    let listener = std::net::TcpListener::bind(SocketAddr::from(([127, 0, 0, 1], 0)))?;
    let base = format!("http://{}", listener.local_addr()?);
    listener.set_nonblocking(true)?;

    let _guard = rt.enter();
    let listener = tokio::net::TcpListener::from_std(listener)?;
    rt.spawn(async move {
        loop {
            // A failed accept is not a reason to stop listening. This used
            // to `return`, which turns one transient error into a service
            // that is gone for the rest of the scenario — and its absence
            // then surfaces as whatever the command downstream says when a
            // DID will not resolve, which is a long way from the cause.
            let Ok((stream, _)) = listener.accept().await else {
                tokio::task::yield_now().await;
                continue;
            };
            let world = world.clone();
            let handler = handler.clone();
            tokio::spawn(async move {
                let service = service_fn(move |req: Request<hyper::body::Incoming>| {
                    let world = world.clone();
                    let handler = handler.clone();
                    async move { Ok::<_, Infallible>(answer(&world, handler.as_ref(), req).await) }
                });
                let _ = hyper::server::conn::http1::Builder::new()
                    .serve_connection(TokioIo::new(stream), service)
                    .await;
            });
        }
    });
    Ok(Service { base })
}

async fn answer<F>(
    world: &Mutex<World>,
    handler: &F,
    req: Request<hyper::body::Incoming>,
) -> Response<Body>
where
    F: Fn(&mut World, &Incoming) -> Reply,
{
    let method = req.method().clone();
    let token = req
        .headers()
        .get(hyper::header::AUTHORIZATION)
        .and_then(|v| v.to_str().ok())
        // "DPoP <token>" and "Bearer <token>" both reduce to the credential.
        .map(|v| v.split_whitespace().last().unwrap_or(v).to_string());
    let uri = req.uri().clone();
    let body = req
        .into_body()
        .collect()
        .await
        .map(|b| b.to_bytes().to_vec())
        .unwrap_or_default();

    let incoming = Incoming {
        method,
        // The path arrives percent-encoded — a DID's colons survive, but a
        // strict client may encode them, and `at://` URIs in query strings
        // certainly are.
        path: decode(uri.path().trim_start_matches('/')),
        params: query(uri.query().unwrap_or_default()),
        body,
        token,
    };

    let reply = {
        let mut world = world.lock().expect("the world's lock");
        handler(&mut world, &incoming)
    };

    match reply {
        Reply::JsonStatus(status, value) => Response::builder()
            .status(status)
            .header(hyper::header::CONTENT_TYPE, "application/json")
            .body(Full::new(Bytes::from(value.to_string())).boxed())
            .expect("a response"),
        Reply::Json(value) => Response::builder()
            .status(StatusCode::OK)
            .header(hyper::header::CONTENT_TYPE, "application/json")
            .body(Full::new(Bytes::from(value.to_string())).boxed())
            .expect("a response"),
        Reply::Bytes(bytes) => Response::builder()
            .status(StatusCode::OK)
            .header(hyper::header::CONTENT_TYPE, "application/octet-stream")
            .body(Full::new(Bytes::from(bytes)).boxed())
            .expect("a response"),
        Reply::Error(status, error, message) => {
            let body = serde_json::json!({ "error": error, "message": message });
            Response::builder()
                .status(status)
                .header(hyper::header::CONTENT_TYPE, "application/json")
                .body(Full::new(Bytes::from(body.to_string())).boxed())
                .expect("a response")
        }
        Reply::Redirect(location) => Response::builder()
            .status(StatusCode::FOUND)
            .header(hyper::header::LOCATION, location)
            .body(Full::new(Bytes::new()).boxed())
            .expect("a response"),
        // A 200 with a plausible content type, because the interesting
        // refusal is the one that happens before anything has been parsed
        // or the status has been looked at.
        Reply::Flood(bytes) => Response::builder()
            .status(StatusCode::OK)
            .header(hyper::header::CONTENT_TYPE, "application/json")
            .body(Flood { remaining: bytes }.boxed())
            .expect("a response"),
    }
}

/// A body of `remaining` bytes that never says how many there will be.
///
/// The shape the size bound actually has to survive. Everything built on
/// `Full<Bytes>` carries a `Content-Length`, so a mock made of those
/// exercises only the cheap half of `http::bytes_bounded` — the half a
/// sender can omit for free by answering chunked, which is what this does.
/// What is left holding the line is the running total, which is the thing
/// worth testing end to end.
///
/// It never yields `Poll::Pending`, so there is no waker to store: a real
/// hostile server would trickle, and trickling is precisely what
/// `READ_TIMEOUT` already covers. What it cannot cover is a server that
/// answers *promptly* and forever, and that is this.
struct Flood {
    remaining: usize,
}

impl hyper::body::Body for Flood {
    type Data = Bytes;
    type Error = Infallible;

    fn poll_frame(
        mut self: Pin<&mut Self>,
        _cx: &mut Context<'_>,
    ) -> Poll<Option<Result<Frame<Bytes>, Infallible>>> {
        if self.remaining == 0 {
            return Poll::Ready(None);
        }
        // 64 KiB a frame: few enough frames that nine mebibytes is fast, and
        // small enough that the running total is consulted a hundred-odd
        // times on the way to the limit rather than sailing past it in one.
        let n = self.remaining.min(64 * 1024);
        self.remaining -= n;
        Poll::Ready(Some(Ok(Frame::data(Bytes::from(vec![b'x'; n])))))
    }
}

fn query(raw: &str) -> std::collections::BTreeMap<String, String> {
    raw.split('&')
        .filter(|pair| !pair.is_empty())
        .filter_map(|pair| pair.split_once('='))
        .map(|(k, v)| (decode(k), decode(v)))
        .collect()
}

/// Percent-decoding, enough for the values atgc puts in a query: DIDs,
/// NSIDs, record keys, CIDs and `at://` URIs.
fn decode(raw: &str) -> String {
    let bytes = raw.as_bytes();
    let mut out = Vec::with_capacity(bytes.len());
    let mut i = 0;
    while i < bytes.len() {
        if bytes[i] == b'%' && i + 2 < bytes.len() {
            let hex = std::str::from_utf8(&bytes[i + 1..i + 3]).unwrap_or("");
            if let Ok(byte) = u8::from_str_radix(hex, 16) {
                out.push(byte);
                i += 3;
                continue;
            }
        }
        out.push(bytes[i]);
        i += 1;
    }
    String::from_utf8_lossy(&out).into_owned()
}