#![allow(clippy::too_many_lines)]
#![allow(clippy::cast_possible_truncation)]
use super::cookie::{CookieJar, host_key};
use super::{HttpTransport, PER_HOST_CONCURRENCY_LIMIT};
use crate::camouflage::{HeaderConfig, apply_browser_headers_to_request, random_delay};
use futures::{StreamExt, stream::FuturesUnordered};
use reqwest::header::{CONTENT_LENGTH, COOKIE};
use secir::matcher::ResponseData;
use secir::transport::{RequestPlan, RequestSpec, Response, Transport};
use secir::{Error, Result};
use std::borrow::Cow;
use std::collections::HashMap;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration;
use guise_pacing::capped_pow2_backoff_ms;
use tokio::sync::{Mutex as AsyncMutex, Notify, Semaphore};

#[derive(Clone, Copy)]
pub(super) struct ExecuteRequestConfig<'a> {
    max_response_size: usize,
    allow_internal: bool,
    cookie_jar: &'a CookieJar,
    retries: usize,
    header_config: &'a HeaderConfig,
    request_index: usize,
}

#[derive(Debug, Default)]
pub(super) struct RequestCoordinator {
    states: AsyncMutex<HashMap<SequenceScope, Arc<SequenceState>>>,
}

#[derive(Debug, Clone, PartialEq, Eq, Hash)]
struct SequenceScope {
    host: String,
    template_id: String,
}

#[derive(Debug, Default)]
struct SequenceState {
    completed: AsyncMutex<Option<usize>>,
    notify: Notify,
}

impl RequestCoordinator {
    pub(super) async fn wait_for_predecessors(&self, spec: &RequestSpec) {
        for (template_id, request_index) in &spec.template_refs {
            if *request_index == 0 {
                continue;
            }

            let Some(host) = host_key(&spec.url) else {
                continue;
            };
            let scope = SequenceScope {
                host,
                template_id: template_id.clone(),
            };
            let state = self.state(scope).await;

            loop {
                let notified = state.notify.notified();
                let ready = {
                    let completed = state.completed.lock().await;
                    completed.is_some_and(|index| index + 1 >= *request_index)
                };
                if ready {
                    break;
                }
                notified.await;
            }
        }
    }

    pub(super) async fn mark_completed(&self, spec: &RequestSpec) {
        for (template_id, request_index) in &spec.template_refs {
            let Some(host) = host_key(&spec.url) else {
                continue;
            };
            let scope = SequenceScope {
                host,
                template_id: template_id.clone(),
            };
            let state = self.state(scope).await;
            let completed_index = {
                let mut completed = state.completed.lock().await;
                *completed =
                    Some(completed.map_or(*request_index, |current| current.max(*request_index)));
                *completed
            };
            debug_assert!(completed_index.is_some());
            state.notify.notify_waiters();
        }
    }

    async fn state(&self, scope: SequenceScope) -> Arc<SequenceState> {
        let mut states = self.states.lock().await;
        states
            .entry(scope)
            .or_insert_with(|| Arc::new(SequenceState::default()))
            .clone()
    }
}

use super::target_guard::ensure_external_target;
#[doc(hidden)]
// `is_internal_ip` and `ensure_external_target` moved to `target_guard` to keep
// this module under the modularity limit. Re-exported here so the established
// `http_transport::execute::is_internal_ip` path keeps resolving for callers.
pub(crate) use super::target_guard::is_internal_ip;

impl HttpTransport {
    pub(super) async fn execute_one(
        client: &reqwest::Client,
        spec: &RequestSpec,
        config: ExecuteRequestConfig<'_>,
    ) -> Result<Response> {
        let ExecuteRequestConfig {
            max_response_size,
            allow_internal,
            cookie_jar,
            retries,
            header_config,
            request_index,
        } = config;
        let method: reqwest::Method = spec.method.parse().map_err(|e| Error::Transport {
            url: spec.url.clone(),
            message: format!("invalid method: {}", spec.method),
            source: Some(Box::new(e)),
        })?;

        // Check internal IP ONCE before retry loop  -  no need to re-resolve on every attempt
        if !allow_internal {
            ensure_external_target(&spec.url).await?;
        }

        // O(1) clonable body to avoid career-ending heap allocations on every retry
        let cheap_body = spec.body.as_ref().map(|b| bytes::Bytes::from(b.clone()));

        let mut attempt = 0usize;
        loop {
            let start = std::time::Instant::now();
            if header_config.jitter_max_ms > 0 {
                tokio::time::sleep(random_delay(header_config.jitter_max_ms)).await;
            }
            let mut req = client.request(method.clone(), &spec.url);

            // Build headers without cloning the entire hashmap
            for (key, value) in &spec.headers {
                req = req.header(key.as_str(), value.as_str());
            }
            req =
                apply_browser_headers_to_request(req, &spec.headers, header_config, request_index);

            if spec.cookie_reuse
                && !spec
                    .headers
                    .keys()
                    .any(|key| key.eq_ignore_ascii_case("cookie"))
                && let Some(cookie_header) = cookie_jar.cookie_header(spec)
            {
                req = req.header(COOKIE, cookie_header);
            }

            if let Some(ref body) = cheap_body {
                req = req.body(body.clone());
            }

            let result = async {
                let resp = req.send().await.map_err(|e| Error::Transport {
                    url: spec.url.clone(),
                    message: e.to_string(),
                    source: Some(Box::new(e)),
                })?;

                let status = resp.status().as_u16();
                // Only 429 (rate limit) triggers retry. 500+ responses are
                // returned as normal data  -  templates check for 500 status
                // (error-based SQLi, stack traces, etc.). Throwing away 500s
                // causes false negatives.
                if status == 429 {
                    return Err(Error::Transport {
                        url: spec.url.clone(),
                        message: format!("server returned rate limit status {status}"),
                        source: None,
                    });
                }
                if spec.cookie_reuse {
                    cookie_jar.store_response_cookies(spec, resp.headers());
                }
                let headers: Vec<(String, String)> = {
                    let mut headers = Vec::with_capacity(resp.headers().len());
                    for (key, value) in resp.headers() {
                        let value = match value.to_str() {
                            Ok(value) => Cow::Borrowed(value),
                            Err(_) => {
                                // Lossy decode non-ASCII header values instead of discarding.
                                // `HeaderValue::to_str()` rejects non-ASCII UTF-8; we fall back
                                // to `from_utf8_lossy`.
                                String::from_utf8_lossy(value.as_bytes())
                            }
                        };
                        headers.push((
                            key.as_str().to_string(),
                            value.into_owned(),
                        ));
                    }
                    headers
                };

                let content_length = resp.headers().get(CONTENT_LENGTH).map_or(0, |value| {
                    let value = match value.to_str() {
                        Ok(value) => value,
                        Err(error) => {
                            tracing::warn!(error = %error, "ignoring invalid Content-Length header");
                            return 0;
                        }
                    };
                    match value.parse::<usize>() {
                        Ok(value) => value,
                        Err(error) => {
                            tracing::warn!(
                                value = %value,
                                error = %error,
                                "ignoring unparsable Content-Length header"
                            );
                            0
                        }
                    }
                });

                let initial_capacity = content_length.min(max_response_size);
                let mut body = Vec::with_capacity(initial_capacity);
                let mut resp = resp;
                let mut body_truncated = false;
                loop {
                    let chunk = match resp.chunk().await {
                        Ok(Some(chunk)) => chunk,
                        Ok(None) => break, // end of body
                        Err(e) => {
                            tracing::debug!(
                                url = %spec.url,
                                bytes_read = body.len(),
                                error = %e,
                                "body read ended before Content-Length satisfied"
                            );
                            body_truncated = true;
                            break;
                        }
                    };
                    if body.len().saturating_add(chunk.len()) > max_response_size {
                        return Err(Error::Transport {
                            url: spec.url.clone(),
                            message: format!(
                                "response body too large: exceeds {max_response_size} bytes"
                            ),
                            source: None,
                        });
                    }
                    body.extend_from_slice(&chunk);
                }

                if body_truncated
                    || (content_length > 0 && body.len() < content_length)
                {
                    return Err(Error::Transport {
                        url: spec.url.clone(),
                        message: format!(
                            "incomplete response body: read {} of {} bytes",
                            body.len(),
                            content_length
                        ),
                        source: None,
                    });
                }

                let elapsed = start.elapsed();
                let data = ResponseData::new(status, headers, body)
                    .with_elapsed(elapsed)
                    .with_url(&spec.url);

                Ok(Response {
                    spec: spec.clone(),
                    data,
                    elapsed,
                })
            }
            .await;

            match result {
                Ok(response) => return Ok(response),
                Err(_err) if attempt + 1 < retries => {
                    let backoff_attempt = u32::try_from(attempt).unwrap_or(u32::MAX);
                    tokio::time::sleep(capped_pow2_backoff_ms(100, backoff_attempt, u64::MAX))
                        .await;
                    attempt += 1;
                }
                Err(err) => return Err(err),
            }
        }
    }
}

impl Transport for HttpTransport {
    fn concurrency_limit(&self) -> usize {
        self.concurrency
    }

    fn execute(
        &self,
        plan: &RequestPlan,
    ) -> std::pin::Pin<Box<dyn futures::Stream<Item = Result<Response>> + Send + '_>> {
        let max_response_size = self.max_response_size;
        let allow_internal = self.allow_internal;
        let cookie_jar = Arc::clone(&self.cookie_jar);
        let coordinator = Arc::clone(&self.request_coordinator);
        let retries = self.retries;
        let rotating_clients = self.rotating_clients.clone();
        let default_client = self.client.clone();
        let default_redirect_client = self.redirect_client.clone();
        let rate_limit = self.rate_limit.clone();
        let fixed_delay = self.fixed_delay.clone();
        let global_semaphore = Arc::clone(&self.global_semaphore);
        let host_semaphores = Arc::clone(&self.host_semaphores);
        let adaptive_rate_limiter = Arc::clone(&self.adaptive_rate_limiter);
        let header_config = Arc::clone(&self.header_config);
        let host_error_counts = Arc::clone(&self.host_error_counts);
        let max_host_errors = self.max_host_errors;
        let host_error_tracking_enabled = self.host_error_tracking_enabled;

        let tasks: FuturesUnordered<_> = plan
            .requests
            .iter()
            .enumerate()
            .map(move |(index, spec)| {
                let spec = spec.clone();
                let client = if let Some(pair) =
                    rotating_clients.get(index % rotating_clients.len().max(1))
                {
                    if spec.follow_redirects {
                        pair.redirect_client.clone()
                    } else {
                        pair.client.clone()
                    }
                } else if spec.follow_redirects {
                    default_redirect_client.clone()
                } else {
                    default_client.clone()
                };
                let rate_limit = rate_limit.clone();
                let fixed_delay = fixed_delay.clone();
                let cookie_jar = Arc::clone(&cookie_jar);
                let coordinator = Arc::clone(&coordinator);
                let global_semaphore = Arc::clone(&global_semaphore);
                let host_semaphores = Arc::clone(&host_semaphores);
                let adaptive_rate_limiter = Arc::clone(&adaptive_rate_limiter);
                let header_config = Arc::clone(&header_config);
                let host_error_counts = Arc::clone(&host_error_counts);

                tokio::spawn(async move {
                    super::CURRENT_MAX_REDIRECTS
                        .scope(spec.max_redirects, async move {
                            let host = host_key(&spec.url);
                            coordinator.wait_for_predecessors(&spec).await;

                    let _global_permit =
                        global_semaphore
                            .acquire_owned()
                            .await
                            .map_err(|e| Error::Transport {
                                url: spec.url.clone(),
                                message: "global concurrency semaphore closed".to_string(),
                                source: Some(Box::new(e)),
                            })?;

                    let _host_permit = if let Some(host) = host.as_ref() {
                        let semaphore = host_semaphores
                            .entry(host.clone())
                            .or_insert_with(|| Arc::new(Semaphore::new(PER_HOST_CONCURRENCY_LIMIT)))
                            .clone();
                        Some(semaphore.acquire_owned().await.map_err(|e| Error::Transport {
                            url: spec.url.clone(),
                            message: format!("host concurrency semaphore closed for {host}"),
                            source: Some(Box::new(e)),
                        })?)
                    } else {
                        None
                    };

                    if host_error_tracking_enabled
                        && let Some(host) = host.as_deref()
                    {
                        let is_blocked = host_error_counts
                            .get(host)
                            .is_some_and(|entry| entry.load(Ordering::Relaxed) >= max_host_errors);
                        if is_blocked {
                            coordinator.mark_completed(&spec).await;
                            return Err(Error::Transport {
                                url: spec.url.clone(),
                                message: format!(
                                    "skipping host after reaching --max-host-errors threshold for {host}"
                                ),
                                source: None,
                            });
                        }
                    }

                    if let Some(host) = host.as_deref()
                        && adaptive_rate_limiter.should_throttle(host).await
                    {
                        let wait = adaptive_rate_limiter
                            .throttle_duration(host)
                            .await
                            .unwrap_or_else(|| Duration::from_millis(100));
                        tokio::time::sleep(wait).await;
                    }

                    if let Some(limiter) = rate_limit.as_ref() {
                        limiter.acquire().await;
                    }
                    if let Some(delay) = fixed_delay.as_ref() {
                        delay.acquire().await;
                    }

                    let result = Self::execute_one(
                        &client,
                        &spec,
                        ExecuteRequestConfig {
                            max_response_size,
                            allow_internal,
                            cookie_jar: &cookie_jar,
                            retries,
                            header_config: &header_config,
                            request_index: index,
                        },
                    )
                    .await;

                    if let (Some(host), Ok(response)) = (host.as_deref(), &result) {
                        adaptive_rate_limiter
                            .record_response(host, response.data.status, response.elapsed)
                            .await;
                    }
                    if let (Some(host), Err(_)) = (host.as_deref(), &result)
                        && host_error_tracking_enabled
                    {
                        host_error_counts
                            .entry(host.to_string())
                            .or_insert_with(|| AtomicUsize::new(0))
                            .fetch_add(1, Ordering::Relaxed);
                    }

                            coordinator.mark_completed(&spec).await;
                            result
                        })
                        .await
                })
            })
            .collect();

        Box::pin(tasks.map(|result| {
            result.map_err(|error| Error::Transport {
                url: String::new(),
                message: format!("http execution task failed: {error}"),
                source: Some(Box::new(error)),
            })?
        }))
    }
}
