use serde::Serialize;
use std::collections::HashMap;
use std::env;
use std::fmt;
use std::sync::{Arc, OnceLock, RwLock};
use std::time::Duration;
use wreq::Client;
use wreq::Proxy;
use wreq::Response;
use wreq::Version;
use wreq::Uri;
use wreq::cookie::{Cookies, CookieStore, Jar};
use wreq::header::{HeaderMap, HeaderValue, SET_COOKIE, USER_AGENT};
use wreq::multipart::Form;
use wreq::redirect::Policy;
use wreq_util::Emulation;

use crate::util::flaresolverr::FlareSolverrRequest;
use crate::util::flaresolverr::Flaresolverr;
use crate::util::proxy;

// A Send + Sync error type for all async paths
type AnyErr = Box<dyn std::error::Error + Send + Sync + 'static>;

#[derive(serde::Serialize, serde::Deserialize, Clone)]
pub struct Requester {
    #[serde(skip)]
    pub(crate) client: Client,
    #[serde(skip)]
    pub(crate) cookie_jar: Arc<Jar>,
    #[serde(skip)]
    pub(crate) debug_trace_id: Option<String>,
    pub(crate) proxy: bool,
    flaresolverr_session: Option<String>,
    pub(crate) user_agent: Option<String>,
}

impl fmt::Debug for Requester {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Requester")
            .field("proxy", &self.proxy)
            .field("debug_trace_id", &self.debug_trace_id)
            .field("flaresolverr_session", &self.flaresolverr_session)
            .field("user_agent", &self.user_agent)
            .finish()
    }
}

impl Requester {
    fn shared_cookie_jar() -> Arc<Jar> {
        static SHARED_COOKIE_JAR: OnceLock<Arc<Jar>> = OnceLock::new();
        SHARED_COOKIE_JAR
            .get_or_init(|| Arc::new(Jar::default()))
            .clone()
    }

    /// User agents recovered from FlareSolverr solves, keyed by host.
    ///
    /// The cookie jar is already process-wide, but the solved user agent used to
    /// live only on the `Requester` that ran the solve. Providers clone their
    /// requester for every concurrent detail fetch, so those clones went out with
    /// the default user agent and Cloudflare rejected them — forcing another
    /// (~15s) solve per page. Keeping the identity here lets every later request
    /// to that host replay the solved user agent alongside the shared cookies and
    /// stay on the direct path.
    fn flaresolverr_user_agents() -> &'static RwLock<HashMap<String, String>> {
        static USER_AGENTS: OnceLock<RwLock<HashMap<String, String>>> = OnceLock::new();
        USER_AGENTS.get_or_init(|| RwLock::new(HashMap::new()))
    }

    fn host_key(url: &str) -> Option<String> {
        url::Url::parse(url)
            .ok()?
            .host_str()
            .map(|host| host.trim_start_matches("www.").to_ascii_lowercase())
    }

    pub(crate) fn remember_flaresolverr_user_agent(url: &str, user_agent: &str) {
        if user_agent.is_empty() {
            return;
        }
        let Some(host) = Self::host_key(url) else {
            return;
        };
        if let Ok(mut agents) = Self::flaresolverr_user_agents().write() {
            agents.insert(host, user_agent.to_string());
        }
    }

    fn remembered_flaresolverr_user_agent(url: &str) -> Option<String> {
        let host = Self::host_key(url)?;
        Self::flaresolverr_user_agents()
            .read()
            .ok()?
            .get(&host)
            .cloned()
    }

    /// Adopt a previously solved user agent for `url` when this requester does not
    /// already carry one, so the direct attempt reuses the FlareSolverr identity.
    /// Returns true when the client was rebuilt.
    fn adopt_remembered_identity(&mut self, url: &str) -> bool {
        if self.user_agent.is_some() {
            return false;
        }
        let Some(user_agent) = Self::remembered_flaresolverr_user_agent(url) else {
            return false;
        };
        self.user_agent = Some(user_agent);
        self.client = Self::build_client(self.cookie_jar.clone(), self.user_agent.as_deref());
        true
    }

    fn origin_url_for_cookie_scope(url: &str) -> Option<url::Url> {
        let parsed = url::Url::parse(url).ok()?;
        let host = parsed.host_str()?;
        let scheme = parsed.scheme();
        url::Url::parse(&format!("{scheme}://{host}/")).ok()
    }

    fn store_response_cookies(&self, url: &str, response: &Response) {
        let Some(default_origin) = Self::origin_url_for_cookie_scope(url) else {
            return;
        };

        for value in response.headers().get_all(SET_COOKIE).iter() {
            let Ok(cookie) = value.to_str() else {
                continue;
            };
            // Honour the Domain attribute: if Set-Cookie specifies Domain=.foo.com,
            // register the cookie against https://foo.com/ so the Jar returns it for
            // sub.foo.com requests as well.
            let origin = Self::cookie_domain_origin(cookie, &default_origin);
            self.cookie_jar.add(cookie, origin.as_str());
        }
    }

    /// Derive the registration origin for a cookie string.
    /// If the string contains `Domain=<d>`, strips the leading dot and builds a
    /// URL from `<d>` so the jar scopes the cookie to the whole domain tree.
    /// Falls back to `default` when no Domain attribute is present.
    fn cookie_domain_origin(cookie_str: &str, default: &url::Url) -> url::Url {
        for attr in cookie_str.split(';').skip(1) {
            let attr = attr.trim();
            if attr.len() >= 7 && attr[..7].eq_ignore_ascii_case("domain=") {
                let domain = attr[7..].trim().trim_start_matches('.');
                if !domain.is_empty() {
                    let scheme = default.scheme();
                    if let Ok(u) = url::Url::parse(&format!("{scheme}://{domain}/")) {
                        return u;
                    }
                }
            }
        }
        default.clone()
    }

    pub(crate) fn store_flaresolverr_cookies(
        &mut self,
        request_url: &str,
        cookies: &[crate::util::flaresolverr::FlaresolverrCookie],
    ) {
        let fallback_origin = Self::origin_url_for_cookie_scope(request_url);

        for cookie in cookies {
            let origin = if !cookie.domain.is_empty() {
                let scheme = fallback_origin
                    .as_ref()
                    .map(|url| url.scheme())
                    .unwrap_or("https");
                let host = cookie.domain.trim_start_matches('.');
                url::Url::parse(&format!("{scheme}://{host}/"))
                    .ok()
                    .or_else(|| fallback_origin.clone())
            } else {
                fallback_origin.clone()
            };

            let Some(origin) = origin else {
                continue;
            };

            let mut cookie_string =
                format!("{}={}; Path={}", cookie.name, cookie.value, cookie.path);
            if !cookie.domain.is_empty() {
                cookie_string.push_str(&format!("; Domain={}", cookie.domain));
            }
            if cookie.secure {
                cookie_string.push_str("; Secure");
            }
            if cookie.httpOnly {
                cookie_string.push_str("; HttpOnly");
            }
            if let Some(same_site) = cookie.sameSite.as_deref() {
                if !same_site.is_empty() {
                    cookie_string.push_str(&format!("; SameSite={same_site}"));
                }
            }

            self.cookie_jar.add(cookie_string.as_str(), origin.as_str());
        }
    }

    fn debug_cookie_preview_from_owned_headers(
        &self,
        url: &str,
        headers: &[(String, String)],
    ) -> String {
        if let Some((_, value)) = headers
            .iter()
            .find(|(key, _)| key.eq_ignore_ascii_case("cookie"))
        {
            return crate::util::flow_debug::preview(value, 160);
        }

        self.cookie_header_for_url(url)
            .map(|cookie| crate::util::flow_debug::preview(&cookie, 160))
            .unwrap_or_else(|| "none".to_string())
    }

    fn looks_like_cloudflare_challenge(body: &str) -> bool {
        body.contains("Just a moment...")
            || body.contains("cf-turnstile-response")
            || body.contains("Enable JavaScript and cookies to continue")
    }

    fn jina_mirror_url(url: &str) -> Option<String> {
        if url.trim().is_empty() {
            return None;
        }
        if url.starts_with("https://r.jina.ai/") || url.starts_with("http://r.jina.ai/") {
            return Some(url.to_string());
        }
        let stripped = url
            .strip_prefix("https://")
            .or_else(|| url.strip_prefix("http://"))
            .unwrap_or(url);
        Some(format!("https://r.jina.ai/http://{stripped}"))
    }

    async fn fetch_jina_mirror_body(
        cookie_jar: Arc<Jar>,
        user_agent: Option<String>,
        proxy_enabled: bool,
        url: &str,
    ) -> Result<String, AnyErr> {
        let mirror_url = Self::jina_mirror_url(url).ok_or("invalid mirror url")?;
        let client = Self::build_client(cookie_jar, user_agent.as_deref());
        let mut request = client
            .get(&mirror_url)
            .version(Version::HTTP_11)
            .header("X-Return-Format", "html");
        if proxy_enabled {
            if let Ok(proxy_url) = env::var("BURP_URL") {
                let proxy = Proxy::all(&proxy_url)?;
                request = request.proxy(proxy);
            }
        }
        let response = request.send().await?;
        if !response.status().is_success() {
            return Err(format!("jina mirror returned status {}", response.status()).into());
        }
        Ok(response.text().await?)
    }

    fn debug_cookie_preview_from_borrowed_headers(
        &self,
        url: &str,
        headers: &[(&str, &str)],
    ) -> String {
        if let Some((_, value)) = headers
            .iter()
            .find(|(key, _)| key.eq_ignore_ascii_case("cookie"))
        {
            return crate::util::flow_debug::preview(value, 160);
        }

        self.cookie_header_for_url(url)
            .map(|cookie| crate::util::flow_debug::preview(&cookie, 160))
            .unwrap_or_else(|| "none".to_string())
    }

    pub(crate) fn build_client(cookie_jar: Arc<Jar>, user_agent: Option<&str>) -> Client {
        let mut builder = Client::builder()
            .tls_cert_verification(false)
            .emulation(Emulation::Firefox151)
            .cookie_provider(cookie_jar)
            .redirect(Policy::default());

        if let Some(user_agent) = user_agent {
            let mut headers = HeaderMap::new();
            if let Ok(value) = HeaderValue::from_str(user_agent) {
                headers.insert(USER_AGENT, value);
                builder = builder.default_headers(headers);
            }
        }

        builder.build().expect("Failed to create HTTP client")
    }

    pub fn new() -> Self {
        let cookie_jar = Self::shared_cookie_jar();
        let client = Self::build_client(cookie_jar.clone(), None);

        let requester = Requester {
            client,
            cookie_jar,
            debug_trace_id: None,
            proxy: false,
            flaresolverr_session: None,
            user_agent: None,
        };

        proxy::init_all_proxies_background(requester.clone());

        requester
    }

    pub fn set_proxy(&mut self, proxy: bool) {
        if proxy {
            println!("Proxy enabled");
        }
        self.proxy = proxy;
    }

    pub fn set_debug_trace_id(&mut self, debug_trace_id: Option<String>) {
        self.debug_trace_id = debug_trace_id;
    }

    pub fn debug_trace_id(&self) -> Option<&str> {
        self.debug_trace_id.as_deref()
    }

    pub fn cookie_header_for_url(&self, url: &str) -> Option<String> {
        let uri = url.parse::<Uri>().ok()?;
        // HTTP/1.1 asks the store for the combined `a=1; b=2` form; HTTP/2 would
        // hand back one field value per cookie instead.
        let joined = match self.cookie_jar.cookies(&uri, Version::HTTP_11) {
            Cookies::Compressed(value) => value.to_str().ok()?.to_string(),
            Cookies::Uncompressed(values) => values
                .iter()
                .filter_map(|value| value.to_str().ok())
                .collect::<Vec<_>>()
                .join("; "),
            _ => return None,
        };
        (!joined.is_empty()).then_some(joined)
    }

    pub async fn get_raw(&mut self, url: &str) -> Result<Response, wreq::Error> {
        self.adopt_remembered_identity(url);
        let cookie_preview = self
            .cookie_header_for_url(url)
            .map(|cookie| crate::util::flow_debug::preview(&cookie, 160))
            .unwrap_or_else(|| "none".to_string());
        #[cfg(not(feature = "debug"))]
        let _ = &cookie_preview;
        crate::flow_debug!(
            "trace={} requester get_raw url={} cookies={} proxy={}",
            self.debug_trace_id().unwrap_or("none"),
            crate::util::flow_debug::preview(url, 120),
            cookie_preview,
            self.proxy
        );
        let client = Self::build_client(self.cookie_jar.clone(), self.user_agent.as_deref());

        let mut request = client.get(url).version(Version::HTTP_11);

        if self.proxy {
            if let Ok(proxy_url) = env::var("BURP_URL") {
                let proxy = Proxy::all(&proxy_url).unwrap();
                request = request.proxy(proxy);
            }
        }

        let response = request.send().await?;
        self.store_response_cookies(url, &response);
        Ok(response)
    }

    /// Like [`get_raw`], but disables redirect-following so a 3xx response's
    /// `Location` header can be inspected directly. Uses the same
    /// Cloudflare-resilient client (Firefox emulation, shared cookie jar,
    /// FlareSolverr-derived user agent) as every other request path — unlike
    /// a bare `wreq::Client`, which Cloudflare is far more likely to flag,
    /// especially from datacenter/production IPs.
    pub async fn get_no_redirect_raw(&mut self, url: &str) -> Result<Response, wreq::Error> {
        self.adopt_remembered_identity(url);
        let cookie_preview = self
            .cookie_header_for_url(url)
            .map(|cookie| crate::util::flow_debug::preview(&cookie, 160))
            .unwrap_or_else(|| "none".to_string());
        #[cfg(not(feature = "debug"))]
        let _ = &cookie_preview;
        crate::flow_debug!(
            "trace={} requester get_no_redirect_raw url={} cookies={} proxy={}",
            self.debug_trace_id().unwrap_or("none"),
            crate::util::flow_debug::preview(url, 120),
            cookie_preview,
            self.proxy
        );
        let mut builder = Client::builder()
            .tls_cert_verification(false)
            .emulation(Emulation::Firefox151)
            .cookie_provider(self.cookie_jar.clone())
            .redirect(Policy::none());
        if let Some(user_agent) = self.user_agent.as_deref() {
            let mut headers = HeaderMap::new();
            if let Ok(value) = HeaderValue::from_str(user_agent) {
                headers.insert(USER_AGENT, value);
                builder = builder.default_headers(headers);
            }
        }
        let client = builder.build().expect("Failed to create HTTP client");

        let mut request = client.get(url).version(Version::HTTP_11);

        if self.proxy {
            if let Ok(proxy_url) = env::var("BURP_URL") {
                let proxy = Proxy::all(&proxy_url).unwrap();
                request = request.proxy(proxy);
            }
        }

        let response = request.send().await?;
        self.store_response_cookies(url, &response);
        Ok(response)
    }

    /// Like [`get_raw_with_headers`], but disables redirect-following so a 3xx response's
    /// `Location` header can be inspected directly.
    pub async fn get_no_redirect_raw_with_headers(
        &mut self,
        url: &str,
        headers: Vec<(String, String)>,
    ) -> Result<Response, wreq::Error> {
        self.adopt_remembered_identity(url);
        let cookie_preview = self
            .cookie_header_for_url(url)
            .map(|cookie| crate::util::flow_debug::preview(&cookie, 160))
            .unwrap_or_else(|| "none".to_string());
        #[cfg(not(feature = "debug"))]
        let _ = &cookie_preview;
        crate::flow_debug!(
            "trace={} requester get_no_redirect_raw_with_headers url={} headers={} cookies={} proxy={}",
            self.debug_trace_id().unwrap_or("none"),
            crate::util::flow_debug::preview(url, 120),
            headers.len(),
            cookie_preview,
            self.proxy
        );
        let mut builder = Client::builder()
            .tls_cert_verification(false)
            .emulation(Emulation::Firefox151)
            .cookie_provider(self.cookie_jar.clone())
            .redirect(Policy::none());
        if let Some(user_agent) = self.user_agent.as_deref() {
            let mut hdrs = HeaderMap::new();
            if let Ok(value) = HeaderValue::from_str(user_agent) {
                hdrs.insert(USER_AGENT, value);
                builder = builder.default_headers(hdrs);
            }
        }
        let client = builder.build().expect("Failed to create HTTP client");

        let mut request = client.get(url).version(Version::HTTP_11);

        if self.proxy {
            if let Ok(proxy_url) = env::var("BURP_URL") {
                let proxy = Proxy::all(&proxy_url).unwrap();
                request = request.proxy(proxy);
            }
        }
        // Set custom headers
        for (key, value) in headers.iter() {
            request = request.header(key, value);
        }

        let response = request.send().await?;
        self.store_response_cookies(url, &response);
        Ok(response)
    }

    pub async fn get_raw_with_headers(
        &mut self,
        url: &str,
        headers: Vec<(String, String)>,
    ) -> Result<Response, wreq::Error> {
        self.get_raw_with_headers_timeout(url, headers, None, None).await
    }

    pub async fn get_raw_with_headers_timeout(
        &mut self,
        url: &str,
        headers: Vec<(String, String)>,
        timeout: Option<Duration>,
        http_version: Option<Version>,
    ) -> Result<Response, wreq::Error> {
        // Replay the solved FlareSolverr identity here too: a token minted
        // through `get_with_headers` under that user agent is rejected (401)
        // when the follow-up raw request goes out with the default one.
        self.adopt_remembered_identity(url);
        let cookie_preview = self.debug_cookie_preview_from_owned_headers(url, &headers);
        #[cfg(not(feature = "debug"))]
        let _ = &cookie_preview;
        crate::flow_debug!(
            "trace={} requester get_raw_with_headers url={} headers={} cookies={} http_version={:?} proxy={}",
            self.debug_trace_id().unwrap_or("none"),
            crate::util::flow_debug::preview(url, 120),
            headers.len(),
            cookie_preview,
            http_version,
            self.proxy
        );
        let client = Self::build_client(self.cookie_jar.clone(), self.user_agent.as_deref());

        let mut request = client.get(url);
        if let Some(version) = http_version {
            request = request.version(version);
        } else {
            request = request.version(Version::HTTP_11);
        }
        if let Some(timeout) = timeout {
            request = request.timeout(timeout);
        }

        if self.proxy {
            if let Ok(proxy_url) = env::var("BURP_URL") {
                let proxy = Proxy::all(&proxy_url).unwrap();
                request = request.proxy(proxy);
            }
        }
        // Set custom headers
        for (key, value) in headers.iter() {
            request = request.header(key, value);
        }
        let response = request.send().await?;
        self.store_response_cookies(url, &response);
        Ok(response)
    }

    pub async fn post_json<S>(
        &mut self,
        url: &str,
        data: &S,
        headers: Vec<(String, String)>,
    ) -> Result<Response, wreq::Error>
    where
        S: Serialize + ?Sized,
    {
        let cookie_preview = self.debug_cookie_preview_from_owned_headers(url, &headers);
        #[cfg(not(feature = "debug"))]
        let _ = &cookie_preview;
        crate::flow_debug!(
            "trace={} requester post_json url={} headers={} cookies={} proxy={}",
            self.debug_trace_id().unwrap_or("none"),
            crate::util::flow_debug::preview(url, 120),
            headers.len(),
            cookie_preview,
            self.proxy
        );
        let mut request = self.client.post(url).version(Version::HTTP_11).json(data);

        // Set custom headers
        for (key, value) in headers.iter() {
            request = request.header(key, value);
        }

        if self.proxy {
            if let Ok(proxy_url) = env::var("BURP_URL") {
                let proxy = Proxy::all(&proxy_url).unwrap();
                request = request.proxy(proxy);
            }
        }

        let response = request.send().await?;
        self.store_response_cookies(url, &response);
        Ok(response)
    }

    pub async fn post(
        &mut self,
        url: &str,
        data: &str,
        headers: Vec<(&str, &str)>,
    ) -> Result<Response, wreq::Error> {
        let cookie_preview = self.debug_cookie_preview_from_borrowed_headers(url, &headers);
        #[cfg(not(feature = "debug"))]
        let _ = &cookie_preview;
        crate::flow_debug!(
            "trace={} requester post url={} headers={} cookies={} body_len={} proxy={}",
            self.debug_trace_id().unwrap_or("none"),
            crate::util::flow_debug::preview(url, 120),
            headers.len(),
            cookie_preview,
            data.len(),
            self.proxy
        );
        let mut request = self
            .client
            .post(url)
            .version(Version::HTTP_11)
            .body(data.to_string());

        // Set custom headers
        for (key, value) in headers.iter() {
            request = request.header(key.to_string(), value.to_string());
        }

        if self.proxy {
            if let Ok(proxy_url) = env::var("BURP_URL") {
                let proxy = Proxy::all(&proxy_url).unwrap();
                request = request.proxy(proxy);
            }
        }

        let response = request.send().await?;
        self.store_response_cookies(url, &response);
        Ok(response)
    }

    pub async fn post_multipart(
        &mut self,
        url: &str,
        form: Form,
        headers: Vec<(String, String)>,
        _http_version: Option<Version>,
    ) -> Result<Response, wreq::Error> {
        let cookie_preview = self.debug_cookie_preview_from_owned_headers(url, &headers);
        #[cfg(not(feature = "debug"))]
        let _ = &cookie_preview;
        crate::flow_debug!(
            "trace={} requester post_multipart url={} headers={} cookies={} proxy={}",
            self.debug_trace_id().unwrap_or("none"),
            crate::util::flow_debug::preview(url, 120),
            headers.len(),
            cookie_preview,
            self.proxy
        );
        let http_version = match _http_version {
            Some(v) => v,
            None => Version::HTTP_11,
        };

        let mut request = self.client.post(url).multipart(form).version(http_version);

        // Set custom headers
        for (key, value) in headers.iter() {
            request = request.header(key, value);
        }

        if self.proxy {
            if let Ok(proxy_url) = env::var("BURP_URL") {
                let proxy = Proxy::all(&proxy_url).unwrap();
                request = request.proxy(proxy);
            }
        }

        let response = request.send().await?;
        self.store_response_cookies(url, &response);
        Ok(response)
    }

    pub async fn get(
        &mut self,
        url: &str,
        _http_version: Option<Version>,
    ) -> Result<String, AnyErr> {
        self.get_with_headers(url, Vec::new(), _http_version).await
    }

    pub async fn get_with_headers(
        &mut self,
        url: &str,
        headers: Vec<(String, String)>,
        _http_version: Option<Version>,
    ) -> Result<String, AnyErr> {
        if self.adopt_remembered_identity(url) {
            crate::flow_debug!(
                "trace={} requester reusing flaresolverr identity url={} user_agent={}",
                self.debug_trace_id().unwrap_or("none"),
                crate::util::flow_debug::preview(url, 120),
                crate::util::flow_debug::preview(self.user_agent.as_deref().unwrap_or("unknown"), 96)
            );
        }
        let cookie_preview = self.debug_cookie_preview_from_owned_headers(url, &headers);
        #[cfg(not(feature = "debug"))]
        let _ = &cookie_preview;
        crate::flow_debug!(
            "trace={} requester get_with_headers start url={} headers={} cookies={} http_version={:?} proxy={}",
            self.debug_trace_id().unwrap_or("none"),
            crate::util::flow_debug::preview(url, 120),
            headers.len(),
            cookie_preview,
            _http_version,
            self.proxy
        );
        // Only pin the protocol when the caller asked for one. Defaulting to
        // HTTP/1.1 used to force every request onto h1 while the TLS handshake
        // still advertised the browser emulation's h2 ALPN — a mismatch
        // Cloudflare scores as a bot. On jable.tv it is a deterministic 403:
        // the identical client returns 200 over the negotiated h2 and 403 the
        // moment the version is pinned to HTTP/1.1.
        let cookie_jar = self.cookie_jar.clone();
        let user_agent = self.user_agent.clone();
        let proxy_enabled = self.proxy;
        let _trace_id = self.debug_trace_id.as_deref().unwrap_or("none").to_string();
        let url_owned = url.to_string();
        let jina_handle = tokio::spawn(async move {
            Self::fetch_jina_mirror_body(cookie_jar, user_agent, proxy_enabled, &url_owned).await
        });
        // A persistent 429 (e.g. a site's IP-level rate limit kicking in) used
        // to retry forever with no cap and no fallback, hanging until an
        // unrelated caller-side timeout cut it off. Cap the retries and fall
        // through to the same Jina/FlareSolverr fallback every other failure
        // status uses once they're exhausted.
        const MAX_RATE_LIMIT_RETRIES: u8 = 3;
        let mut rate_limit_retries: u8 = 0;
        loop {
            let mut request = self.client.get(url);
            if let Some(version) = _http_version {
                request = request.version(version);
            }
            for (key, value) in headers.iter() {
                request = request.header(key, value);
            }
            if self.proxy {
                if let Ok(proxy_url) = env::var("BURP_URL") {
                    let proxy = Proxy::all(&proxy_url).unwrap();
                    request = request.proxy(proxy);
                }
            }
            let response = match request.send().await {
                Ok(response) => response,
                Err(_error) => {
                    crate::flow_debug!(
                        "trace={} requester direct transport failed url={} error={} using_jina_fallback=true",
                        _trace_id,
                        crate::util::flow_debug::preview(url, 120),
                        crate::util::flow_debug::preview(&_error.to_string(), 160)
                    );
                    match jina_handle.await {
                        Ok(Ok(body)) => {
                            if Self::looks_like_cloudflare_challenge(&body) {
                                crate::flow_debug!(
                                    "trace={} requester jina fallback returned cloudflare challenge after transport error url={} using_flaresolverr=true",
                                    _trace_id,
                                    crate::util::flow_debug::preview(url, 120)
                                );
                            } else {
                                return Ok(body);
                            }
                        }
                        Ok(Err(_jina_error)) => {
                            crate::flow_debug!(
                                "trace={} requester jina fallback failed after transport error url={} error={}",
                                _trace_id,
                                crate::util::flow_debug::preview(url, 120),
                                crate::util::flow_debug::preview(&_jina_error.to_string(), 160)
                            );
                        }
                        Err(_join_error) => {
                            crate::flow_debug!(
                                "trace={} requester jina task join failed after transport error url={} error={}",
                                _trace_id,
                                crate::util::flow_debug::preview(url, 120),
                                crate::util::flow_debug::preview(&_join_error.to_string(), 160)
                            );
                        }
                    }
                    break;
                }
            };
            self.store_response_cookies(url, &response);
            crate::flow_debug!(
                "trace={} requester direct response url={} status={}",
                self.debug_trace_id().unwrap_or("none"),
                crate::util::flow_debug::preview(url, 120),
                response.status()
            );
            if response.status().is_success() || response.status().as_u16() == 404 {
                let status = response.status();
                let body = response.text().await?;
                if status.is_success() && Self::looks_like_cloudflare_challenge(&body) {
                    crate::flow_debug!(
                        "trace={} requester direct response looks like cloudflare challenge url={} using_jina_fallback=true",
                        _trace_id,
                        crate::util::flow_debug::preview(url, 120)
                    );
                    match jina_handle.await {
                        Ok(Ok(jina_body)) => {
                            if Self::looks_like_cloudflare_challenge(&jina_body) {
                                crate::flow_debug!(
                                    "trace={} requester jina fallback also returned cloudflare challenge url={}",
                                    _trace_id,
                                    crate::util::flow_debug::preview(url, 120)
                                );
                            } else {
                                return Ok(jina_body);
                            }
                        }
                        Ok(Err(_jina_error)) => {
                            crate::flow_debug!(
                                "trace={} requester jina fallback failed after cloudflare challenge url={} error={}",
                                _trace_id,
                                crate::util::flow_debug::preview(url, 120),
                                crate::util::flow_debug::preview(&_jina_error.to_string(), 160)
                            );
                        }
                        Err(_join_error) => {
                            crate::flow_debug!(
                                "trace={} requester jina task join failed after cloudflare challenge url={} error={}",
                                _trace_id,
                                crate::util::flow_debug::preview(url, 120),
                                crate::util::flow_debug::preview(&_join_error.to_string(), 160)
                            );
                        }
                    }
                    break;
                }
                jina_handle.abort();
                return Ok(body);
            }
            if response.status().as_u16() == 429 && rate_limit_retries < MAX_RATE_LIMIT_RETRIES {
                rate_limit_retries += 1;
                crate::flow_debug!(
                    "trace={} requester direct retry url={} status=429 attempt={}/{}",
                    self.debug_trace_id().unwrap_or("none"),
                    crate::util::flow_debug::preview(url, 120),
                    rate_limit_retries,
                    MAX_RATE_LIMIT_RETRIES
                );
                tokio::time::sleep(std::time::Duration::from_secs(1)).await;
                continue;
            } else {
                crate::flow_debug!(
                    "trace={} requester direct failed url={} status={} using_jina_fallback=true",
                    _trace_id,
                    crate::util::flow_debug::preview(url, 120),
                    response.status()
                );
                match jina_handle.await {
                    Ok(Ok(body)) => {
                        if Self::looks_like_cloudflare_challenge(&body) {
                            crate::flow_debug!(
                                "trace={} requester jina fallback returned cloudflare challenge url={}",
                                _trace_id,
                                crate::util::flow_debug::preview(url, 120)
                            );
                        } else {
                            return Ok(body);
                        }
                    }
                    Ok(Err(_error)) => {
                        crate::flow_debug!(
                            "trace={} requester jina fallback failed url={} error={}",
                            _trace_id,
                            crate::util::flow_debug::preview(url, 120),
                            crate::util::flow_debug::preview(&_error.to_string(), 160)
                        );
                    }
                    Err(_error) => {
                        crate::flow_debug!(
                            "trace={} requester jina task join failed url={} error={}",
                            _trace_id,
                            crate::util::flow_debug::preview(url, 120),
                            crate::util::flow_debug::preview(&_error.to_string(), 160)
                        );
                    }
                }
                break;
            }
        }

        // If direct request failed, try FlareSolverr. Map its error to a Send+Sync error immediately,
        // so no non-Send error value lives across later `.await`s.
        let flare_url = match env::var("FLARE_URL") {
            Ok(url) => url,
            Err(e) => return Err(format!("FLARE_URL not set: {e}").into()),
        };
        let mut flare = Flaresolverr::new(flare_url);
        if self.proxy && env::var("BURP_URL").is_ok() {
            flare.set_proxy(true);
        }
        crate::flow_debug!(
            "trace={} requester flaresolverr url={} proxy={}",
            self.debug_trace_id().unwrap_or("none"),
            crate::util::flow_debug::preview(url, 120),
            self.proxy
        );

        let res = flare
            .solve(FlareSolverrRequest {
                cmd: "request.get".to_string(),
                url: url.to_string(),
                maxTimeout: 60000,
            })
            .await
            .map_err(|e| -> AnyErr { format!("Failed to solve FlareSolverr: {e}").into() })?;

        // Rebuild client and apply UA/cookies from FlareSolverr. Both are also
        // published process-wide (cookies via the shared jar, user agent by host)
        // so later requests can go direct instead of solving again.
        let useragent = res.solution.userAgent;
        Self::remember_flaresolverr_user_agent(url, &useragent);
        self.user_agent = Some(useragent);
        self.store_flaresolverr_cookies(url, &res.solution.cookies);

        self.client = Self::build_client(self.cookie_jar.clone(), self.user_agent.as_deref());
        crate::flow_debug!(
            "trace={} requester flaresolverr solved url={} user_agent={}",
            self.debug_trace_id().unwrap_or("none"),
            crate::util::flow_debug::preview(url, 120),
            crate::util::flow_debug::preview(self.user_agent.as_deref().unwrap_or("unknown"), 96)
        );

        // Retry the original URL with the updated client & (optional) proxy.
        // Left unpinned so the protocol matches the emulated ALPN, same as the
        // direct attempt.
        let mut request = self.client.get(url);
        for (key, value) in headers.iter() {
            request = request.header(key, value);
        }
        if self.proxy {
            if let Ok(proxy_url) = env::var("BURP_URL") {
                let proxy = Proxy::all(&proxy_url).unwrap();
                request = request.proxy(proxy);
            }
        }

        let response = match request.send().await {
            Ok(r) => r,
            Err(_retry_err) => {
                crate::flow_debug!(
                    "trace={} requester post-flaresolverr retry transport failed url={} using_flaresolverr_body=true",
                    self.debug_trace_id().unwrap_or("none"),
                    crate::util::flow_debug::preview(url, 120)
                );
                return Ok(res.solution.response);
            }
        };
        self.store_response_cookies(url, &response);
        crate::flow_debug!(
            "trace={} requester retry response url={} status={}",
            self.debug_trace_id().unwrap_or("none"),
            crate::util::flow_debug::preview(url, 120),
            response.status()
        );
        if response.status().is_success() {
            return Ok(response.text().await?);
        }

        // Fall back to FlareSolverr-provided body
        crate::flow_debug!(
            "trace={} requester fallback body url={}",
            self.debug_trace_id().unwrap_or("none"),
            crate::util::flow_debug::preview(url, 120)
        );
        Ok(res.solution.response)
    }
}

#[cfg(test)]
mod tests {
    use super::Requester;

    #[test]
    fn new_requesters_share_cookie_jar() {
        let a = Requester::new();
        let b = Requester::new();
        let origin = "https://shared-cookie-requester-test.invalid/";

        a.cookie_jar
            .add("shared_cookie=1; Path=/; SameSite=Lax", origin);

        let cookie_header = b
            .cookie_header_for_url("https://shared-cookie-requester-test.invalid/path")
            .unwrap_or_default();

        assert!(cookie_header.contains("shared_cookie=1"));
    }
}