mod cache_fast_path_tests {
    use super::*;
    use crate::{auth::Auth, config::{self, Config, Update, ValidationPolicy}, db::Store};
    use std::net::SocketAddr;
    use tempfile::TempDir;
    use tower::ServiceExt;

    const CSS: &[u8] = b"/* origin comment */\nbody { color: red; margin: 0px; }\n";

    struct Fixture { app: Arc<App>, _directory: TempDir }
    struct Received { status: StatusCode, headers: HeaderMap, body: Bytes }

    impl Fixture {
        async fn new(config: Config) -> Self {
            let directory = TempDir::new().unwrap();
            let policy = ValidationPolicy {
                tls_addr: None, cert_dir: None, data_dir: None,
                proxy_addr: "127.0.0.1:8080".parse().unwrap(),
                admin_addr: "127.0.0.1:9090".parse().unwrap(),
            };
            let store = Store::open(&directory.path().join("db"), &config).await.unwrap();
            let auth = Auth::new(store.clone(), directory.path().to_owned()).await.unwrap();
            let snapshot = config::compile(config, 0, &policy).unwrap();
            let app = App::new(store, snapshot, policy, auth, 32,
                Duration::from_secs(5), Duration::from_secs(5), Duration::from_secs(30));
            Self { app, _directory: directory }
        }

        async fn send(&self, method: Method, path: &str, headers: &[(&str, &str)], body: Body) -> Received {
            send(self.app.clone(), method, path, headers, body).await
        }

        async fn get(&self, path: &str, headers: &[(&str, &str)]) -> Received {
            self.send(Method::GET, path, headers, Body::empty()).await
        }

        async fn change(&self, edit: impl FnOnce(&mut Config)) {
            let current = self.app.config.load_full();
            let mut config = current.envelope.config.clone();
            edit(&mut config);
            self.app.update(Update { expected_revision: current.envelope.revision, config }).await.unwrap();
        }

        async fn stop(self) {
            self.app.shutdown.cancel();
            self.app.tunnels.close();
            self.app.tunnels.wait().await;
            self.app.store.close().await;
        }
    }

    async fn send(app: Arc<App>, method: Method, path: &str, headers: &[(&str, &str)], body: Body) -> Received {
        let mut builder = Request::builder().method(method).uri(path)
            .header(HOST, "localhost").header(USER_AGENT, "cache-plan-regression/1.0");
        for (name, value) in headers { builder = builder.header(*name, *value); }
        let mut request = builder.body(body).unwrap();
        request.extensions_mut().insert(PeerAddr("192.0.2.50:45678".parse().unwrap()));
        request.extensions_mut().insert(LocalAddr("127.0.0.1:8080".parse().unwrap()));
        let response = router(app).oneshot(request).await.unwrap();
        let (parts, body) = response.into_parts();
        let body = body.collect().await.unwrap().to_bytes();
        Received { status: parts.status, headers: parts.headers, body }
    }

    fn config_for(origin: SocketAddr, enabled: bool) -> Config {
        let mut config = config::demo_config(origin);
        config.cache.enabled = enabled;
        config.cache.expose_status = true;
        config.cache.expose_hits = true;
        config.cache.expose_cache_status = true;
        config.cache.expose_only_hits = false;
        config.cache.static_ignore_cookies = false;
        config
    }

    fn web_config(directory: &TempDir, enabled: bool) -> Config {
        std::fs::write(directory.path().join("style.css"), CSS).unwrap();
        let mut config = config_for("127.0.0.1:9000".parse().unwrap(), enabled);
        config.web_services.push(crate::web_server::Service {
            id: "site".into(), name: "Site".into(), document_root: directory.path().to_owned(),
            cache_control: Some("public, max-age=600".into()), ..Default::default()
        });
        config.vhosts[0].source = crate::vhost::Source::WebServer { web_service_id: "site".into() };
        config.vhosts[0].backends.clear();
        config.vhosts[0].routes.clear();
        config
    }

    #[tokio::test]
    async fn content_type_policy_denies_before_local_web_server_and_cache_and_updates_live() {
        let directory = TempDir::new().unwrap();
        let mut config = web_config(&directory, true);
        config.vhosts[0].content_type_policy = crate::vhost::ContentTypePolicy {
            enabled: true,
            blocked: vec!["multipart/form-data".into(), "image/*".into()],
        };
        let fixture = Fixture::new(config).await;
        // Warm the cache before testing a header-denied cache hit.
        assert_eq!(fixture.get("/style.css", &[]).await.status, StatusCode::OK);
        assert_eq!(fixture.get("/style.css", &[]).await.status, StatusCode::OK);
        assert_eq!(fixture.get("/style.css", &[("content-type", "multipart/form-data; boundary=example")]).await.status, StatusCode::FORBIDDEN);
        assert_eq!(fixture.get("/style.css", &[("content-type", "IMAGE/PNG")]).await.status, StatusCode::FORBIDDEN);
        assert_eq!(fixture.send(Method::POST, "/style.css", &[("content-type", "multipart/form-data; boundary=x")], Body::from("sample")).await.status, StatusCode::FORBIDDEN);
        assert_eq!(fixture.get("/style.css", &[("content-type", "text/plain")]).await.status, StatusCode::OK);
        assert_eq!(fixture.get("/style.css", &[("content-type", "application/json, multipart/form-data")]).await.status, StatusCode::FORBIDDEN);
        fixture.change(|config| config.vhosts[0].content_type_policy.enabled = false).await;
        assert_eq!(fixture.get("/style.css", &[("content-type", "multipart/form-data")]).await.status, StatusCode::OK);
        fixture.stop().await;
    }

    #[tokio::test]
    async fn content_type_policy_denies_before_reverse_proxy_upstream_request() {
        let origin = origin().await;
        let mut config = config_for(origin.addr, false);
        config.vhosts[0].content_type_policy = crate::vhost::ContentTypePolicy {
            enabled: true, blocked: vec!["multipart/form-data".into()],
        };
        let fixture = Fixture::new(config).await;
        let blocked = fixture.send(Method::POST, "/", &[("content-type", "Multipart/Form-Data; boundary=abc")], Body::from("--abc\r\n")).await;
        assert_eq!(blocked.status, StatusCode::FORBIDDEN);
        assert_eq!(origin.requests.load(Ordering::Relaxed), 0);
        let allowed = fixture.get("/", &[("content-type", "application/json")]).await;
        assert_eq!(allowed.status, StatusCode::OK);
        assert_eq!(origin.requests.load(Ordering::Relaxed), 1);
        fixture.stop().await;
    }

    #[tokio::test]
    async fn web_cache_disabled_keeps_get_head_range_conditionals_and_observability() {
        let directory = TempDir::new().unwrap();
        let fixture = Fixture::new(web_config(&directory, false)).await;
        let first = fixture.get("/style.css", &[]).await;
        assert_eq!(first.status, StatusCode::OK);
        assert_eq!(first.body, CSS);
        assert!(!first.headers.contains_key("x-cache"));
        let etag = first.headers["etag"].to_str().unwrap();
        let head = fixture.send(Method::HEAD, "/style.css", &[], Body::empty()).await;
        assert_eq!(head.status, StatusCode::OK);
        assert_eq!(head.headers["content-length"], CSS.len().to_string());
        assert!(head.body.is_empty());
        let partial = fixture.get("/style.css", &[("range", "bytes=0-3"), ("if-range", etag)]).await;
        assert_eq!(partial.status, StatusCode::PARTIAL_CONTENT);
        assert_eq!(partial.body, &CSS[..4]);
        let mismatch = fixture.get("/style.css", &[("range", "bytes=0-3"), ("if-range", "\"different\"")]).await;
        assert_eq!(mismatch.status, StatusCode::OK);
        assert_eq!(mismatch.body, CSS);
        let conditional = fixture.get("/style.css", &[("if-none-match", etag)]).await;
        assert_eq!(conditional.status, StatusCode::NOT_MODIFIED);
        assert!(conditional.body.is_empty());
        let invalid = fixture.get("/style.css", &[("range", "bytes=99999-")]).await;
        assert_eq!(invalid.status, StatusCode::RANGE_NOT_SATISFIABLE);
        assert_eq!(invalid.headers["content-range"], format!("bytes */{}", CSS.len()));
        let only_cached = fixture.get("/style.css", &[("cache-control", "only-if-cached")]).await;
        assert_eq!(only_cached.body, CSS);
        let logs = fixture.app.logs.recent();
        assert_eq!(logs.len(), 7);
        assert!(logs.iter().all(|entry| entry.cache_reason == "cache_disabled"));
        assert!(logs.iter().all(|entry| entry.request_id.len() == 64));
        assert_eq!(fixture.app.metrics.completed.load(Ordering::Relaxed), 7);
        assert_eq!(fixture.app.metrics.active.load(Ordering::Relaxed), 0);
        assert_eq!(fixture.app.logs.stats().queue_dropped, 0);
        fixture.stop().await;
    }

    #[tokio::test]
    async fn web_cache_hot_toggles_keep_miss_hit_and_bypass_diagnostics() {
        let directory = TempDir::new().unwrap();
        let fixture = Fixture::new(web_config(&directory, false)).await;
        assert!(!fixture.get("/style.css", &[]).await.headers.contains_key("x-cache"));
        fixture.change(|config| config.cache.enabled = true).await;
        assert_eq!(fixture.get("/style.css", &[]).await.headers["x-cache"], "MISS");
        let hit = fixture.get("/style.css", &[]).await;
        assert_eq!(hit.headers["x-cache"], "HIT");
        assert_eq!(hit.body, CSS);
        fixture.change(|config| config.vhosts[0].routes.push(crate::vhost::PathRoute {
            id: "uncached".into(), path_prefix: "/style.css".into(), cache: Some(false),
            source: crate::vhost::Source::WebServer { web_service_id: "site".into() },
            ..Default::default()
        })).await;
        let bypass = fixture.get("/style.css", &[]).await;
        assert_eq!(bypass.headers["x-cache"], "BYPASS");
        assert_eq!(bypass.headers["x-cache-reason"], "cache_disabled");
        fixture.change(|config| {
            config.vhosts[0].routes.clear();
            config.vhosts[0].cache.enabled = Some(false);
        }).await;
        assert!(!fixture.get("/style.css", &[]).await.headers.contains_key("x-cache"));
        fixture.change(|config| config.vhosts[0].cache.enabled = None).await;
        assert_eq!(fixture.get("/style.css", &[]).await.headers["x-cache"], "MISS");
        assert_eq!(fixture.get("/style.css", &[]).await.headers["x-cache"], "HIT");
        fixture.stop().await;
    }

    #[tokio::test]
    async fn web_transform_remains_independent_of_disabled_cache_and_range() {
        let directory = TempDir::new().unwrap();
        let mut config = web_config(&directory, false);
        config.transform.enabled = true;
        config.vhosts[0].transform.enabled = true;
        let fixture = Fixture::new(config).await;
        let normal = fixture.get("/style.css", &[]).await;
        assert_eq!(normal.status, StatusCode::OK);
        assert!(normal.body.len() < CSS.len(), "transform must remain active with cache off");
        let range = fixture.get("/style.css", &[("range", "bytes=0-3"), ("accept-encoding", "gzip")]).await;
        assert_eq!(range.status, StatusCode::PARTIAL_CONTENT);
        assert_eq!(range.body, &CSS[..4]);
        assert!(!range.headers.contains_key("content-encoding"));
        fixture.change(|config| config.vhosts[0].transform.enabled = false).await;
        assert_eq!(fixture.get("/style.css", &[]).await.body, CSS);
        fixture.stop().await;
    }

    #[tokio::test]
    async fn cache_fast_paths_keep_global_and_location_access_control() {
        let directory = TempDir::new().unwrap();
        for enabled in [false, true] {
            let mut config = web_config(&directory, enabled);
            let deny = crate::acl::Acl { default: crate::acl::Action::Deny, rules: vec![] };
            config.host_policies.push(config::HostPolicy { hosts: vec!["localhost".into()], acl: deny.clone() });
            let fixture = Fixture::new(config).await;
            assert_eq!(fixture.get("/style.css", &[]).await.status, StatusCode::FORBIDDEN);
            fixture.change(|config| {
                config.host_policies.clear();
                config.vhosts[0].routes.push(crate::vhost::PathRoute {
                    id: "private".into(), path_prefix: "/style.css".into(), acl: Some(deny),
                    source: crate::vhost::Source::WebServer { web_service_id: "site".into() },
                    ..Default::default()
                });
            }).await;
            assert_eq!(fixture.get("/style.css", &[]).await.status, StatusCode::FORBIDDEN);
            fixture.stop().await;
        }
    }

    struct Origin { addr: SocketAddr, requests: Arc<AtomicU64>, task: tokio::task::JoinHandle<()> }
    impl Drop for Origin { fn drop(&mut self) { self.task.abort(); } }

    async fn origin() -> Origin {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let requests = Arc::new(AtomicU64::new(0));
        let count = requests.clone();
        let app = Router::new().fallback(move |request: Request| {
            let count = count.clone();
            async move {
                count.fetch_add(1, Ordering::Relaxed);
                let (parts, body) = request.into_parts();
                if parts.uri.path() == "/slow" { tokio::time::sleep(Duration::from_millis(25)).await; }
                let input = body.collect().await.unwrap().to_bytes();
                let variant = parts.headers.get("x-variant").and_then(|v| v.to_str().ok()).unwrap_or("default");
                let output = if parts.method == Method::POST { input } else { Bytes::from(format!("origin-{variant}")) };
                let mut response = Response::new(Body::from(output.clone()));
                response.headers_mut().insert("etag", HeaderValue::from_static("\"origin-v1\""));
                response.headers_mut().insert("cache-control", HeaderValue::from_static(
                    if parts.uri.path() == "/private" { "private, max-age=600" } else { "public, max-age=600" }));
                response.headers_mut().insert("vary", HeaderValue::from_static("X-Variant"));
                response.headers_mut().insert("content-type", HeaderValue::from_static("text/plain"));
                for name in ["range", "if-range", "if-none-match", "cookie", "authorization", "x-request-id"] {
                    if let Some(value) = parts.headers.get(name) {
                        let echoed: http::HeaderName = format!("x-seen-{name}").parse().unwrap();
                        response.headers_mut().insert(echoed, value.clone());
                    }
                }
                if parts.headers.get("if-none-match").is_some_and(|v| v == "\"origin-v1\"") {
                    *response.status_mut() = StatusCode::NOT_MODIFIED;
                    *response.body_mut() = Body::empty();
                } else if parts.headers.get("range").is_some_and(|v| v == "bytes=0-2")
                    && parts.headers.get("if-range").is_none_or(|v| v == "\"origin-v1\"") {
                    *response.status_mut() = StatusCode::PARTIAL_CONTENT;
                    response.headers_mut().insert("content-range", format!("bytes 0-2/{}", output.len()).parse().unwrap());
                    *response.body_mut() = Body::from(output.slice(..3));
                }
                response
            }
        });
        let task = tokio::spawn(async move { axum::serve(listener, app).await.unwrap(); });
        Origin { addr, requests, task }
    }

    #[tokio::test]
    async fn proxy_disabled_cache_preserves_headers_post_and_conditionals() {
        let origin = origin().await;
        let fixture = Fixture::new(config_for(origin.addr, false)).await;
        let range = fixture.get("/inspect", &[("range", "bytes=0-2"), ("if-range", "\"origin-v1\""),
            ("cookie", "session=one"), ("authorization", "Bearer example")]).await;
        assert_eq!(range.status, StatusCode::PARTIAL_CONTENT);
        assert_eq!(range.body, "ori");
        assert_eq!(range.headers["x-seen-range"], "bytes=0-2");
        assert_eq!(range.headers["x-seen-if-range"], "\"origin-v1\"");
        assert_eq!(range.headers["x-seen-cookie"], "session=one");
        assert_eq!(range.headers["x-seen-authorization"], "Bearer example");
        assert_eq!(range.headers["x-seen-x-request-id"].as_bytes().len(), 64);
        assert!(!range.headers.contains_key("x-cache"));
        let conditional = fixture.get("/inspect", &[("if-none-match", "\"origin-v1\"")]).await;
        assert_eq!(conditional.status, StatusCode::NOT_MODIFIED);
        assert!(conditional.body.is_empty());
        let input = futures_util::stream::iter([
            Ok::<_, std::io::Error>(Bytes::from_static(b"streamed ")),
            Ok(Bytes::from_static(b"content")),
        ]);
        let post = fixture.send(Method::POST, "/echo", &[], Body::from_stream(input)).await;
        assert_eq!(post.status, StatusCode::OK);
        assert_eq!(post.body, "streamed content");
        assert_eq!(origin.requests.load(Ordering::Relaxed), 3);
        assert!(fixture.app.logs.recent().iter().all(|entry| entry.cache_reason == "cache_disabled"));
        fixture.stop().await;
    }

    #[tokio::test]
    async fn proxy_cache_on_preserves_vary_private_revalidation_and_ranges() {
        let origin = origin().await;
        let fixture = Fixture::new(config_for(origin.addr, true)).await;
        for variant in ["a", "b"] {
            assert_eq!(fixture.get("/vary", &[("x-variant", variant)]).await.headers["x-cache"], "MISS");
        }
        let hit = fixture.get("/vary", &[("x-variant", "a")]).await;
        assert_eq!(hit.headers["x-cache"], "HIT");
        assert_eq!(hit.body, "origin-a");
        let range = fixture.get("/vary", &[("x-variant", "a"), ("range", "bytes=0-2"), ("if-range", "\"origin-v1\"")]).await;
        assert_eq!(range.status, StatusCode::PARTIAL_CONTENT);
        assert_eq!(range.headers["x-cache"], "HIT");
        assert_eq!(range.body, "ori");
        let validated = fixture.get("/vary", &[("x-variant", "a"), ("cache-control", "no-cache")]).await;
        assert_eq!(validated.status, StatusCode::OK);
        assert_eq!(validated.headers["x-cache"], "HIT");
        assert_eq!(validated.headers["x-cache-reason"], "cache_revalidated");
        assert_eq!(validated.body, "origin-a");
        let conditional = fixture.get("/vary", &[("x-variant", "a"), ("if-none-match", "\"origin-v1\"")]).await;
        assert_eq!(conditional.status, StatusCode::NOT_MODIFIED);
        let bypass = fixture.get("/vary", &[("authorization", "Bearer example")]).await;
        assert_eq!(bypass.headers["x-cache-reason"], "request_authorization");
        for _ in 0..2 {
            assert_eq!(fixture.get("/private", &[]).await.headers["x-cache"], "MISS");
        }
        assert_eq!(origin.requests.load(Ordering::Relaxed), 6);
        fixture.stop().await;
    }

    #[tokio::test]
    async fn proxy_cache_enabled_concurrent_fill_stays_single_flight() {
        let origin = origin().await;
        let fixture = Fixture::new(config_for(origin.addr, true)).await;
        let mut requests = tokio::task::JoinSet::new();
        for _ in 0..12 {
            let app = fixture.app.clone();
            requests.spawn(async move { send(app, Method::GET, "/slow", &[], Body::empty()).await });
        }
        while let Some(result) = requests.join_next().await {
            let response = result.unwrap();
            assert_eq!(response.status, StatusCode::OK);
            assert_eq!(response.body, "origin-default");
        }
        assert_eq!(origin.requests.load(Ordering::Relaxed), 1);
        assert_eq!(fixture.app.metrics.completed.load(Ordering::Relaxed), 12);
        assert_eq!(fixture.app.metrics.active.load(Ordering::Relaxed), 0);
        fixture.stop().await;
    }
}