use std::future::Future;
use std::net::{IpAddr, SocketAddr};
use std::process::Stdio;
use std::time::Duration;

use anyhow::{Context, Result, anyhow, bail};
use granc_core::client::{DynamicRequest, DynamicResponse, GrancClient};
use granc_core::tonic;
use reqwest::header::{CONTENT_TYPE, COOKIE, HeaderMap, HeaderName, HeaderValue, LOCATION};
use serde_json::{Value, json};
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWriteExt};
use tokio::process::Command as TokioCommand;
use url::Url;

use crate::security::allowlist::ensure_url_allowed;
use crate::security::dns::resolve_and_validate_host;

use super::builder::{PreparedBody, PreparedMultipartPart, PreparedProtocol, PreparedRequest};
use super::decode::decode_response;
use super::extract::extract_result;

#[derive(Debug, Clone)]
pub struct ExecutionResult {
    pub status: u16,
    pub url: String,
    pub result: Value,
    pub decoded: Value,
}

pub async fn execute_prepared_request(prepared: &PreparedRequest) -> Result<ExecutionResult> {
    execute_prepared_request_with_host_validator(prepared, |url: Url| async move {
        let host = url
            .host_str()
            .ok_or_else(|| anyhow::anyhow!("request URL missing host"))?;
        resolve_and_validate_host(host).await
    })
    .await
}

pub async fn execute_prepared_request_with_host_validator<F, Fut>(
    prepared: &PreparedRequest,
    mut host_validator: F,
) -> Result<ExecutionResult>
where
    F: FnMut(Url) -> Fut,
    Fut: Future<Output = Result<Vec<IpAddr>>>,
{
    let attempts = match prepared.protocol {
        PreparedProtocol::Cli => 1,
        PreparedProtocol::Http | PreparedProtocol::Graphql | PreparedProtocol::Grpc => {
            prepared.transport.retry_max_attempts.max(1)
        }
    };
    let mut last_err: Option<anyhow::Error> = None;

    for attempt in 1..=attempts {
        let outcome = match prepared.protocol {
            PreparedProtocol::Http | PreparedProtocol::Graphql => {
                execute_http_once_with_host_validator(prepared, &mut host_validator).await
            }
            PreparedProtocol::Grpc => {
                execute_grpc_once_with_host_validator(prepared, &mut host_validator).await
            }
            PreparedProtocol::Cli => execute_cli_once(prepared).await,
        };

        match outcome {
            Ok(outcome) => {
                if prepared.protocol != PreparedProtocol::Cli
                    && prepared.transport.retry_on_status.contains(&outcome.status)
                    && attempt < attempts
                {
                    tokio::time::sleep(backoff(prepared.transport.retry_backoff, attempt)).await;
                    continue;
                }
                return Ok(outcome);
            }
            Err(err) => {
                if attempt >= attempts {
                    return Err(err);
                }
                last_err = Some(err);
                tokio::time::sleep(backoff(prepared.transport.retry_backoff, attempt)).await;
            }
        }
    }

    Err(last_err.unwrap_or_else(|| anyhow::anyhow!("request failed without details")))
}

async fn execute_cli_once(prepared: &PreparedRequest) -> Result<ExecutionResult> {
    let cli = prepared
        .cli
        .as_ref()
        .ok_or_else(|| anyhow::anyhow!("prepared CLI request missing command data"))?;

    let mut command = TokioCommand::new(&cli.command);
    command.args(&cli.args);
    if let Some(cwd) = &cli.cwd {
        command.current_dir(cwd);
    }
    for (key, value) in &cli.env {
        command.env(key, value);
    }
    command.stdout(Stdio::piped());
    command.stderr(Stdio::piped());
    if cli.stdin.is_some() {
        command.stdin(Stdio::piped());
    } else {
        command.stdin(Stdio::null());
    }

    let mut child = command
        .spawn()
        .with_context(|| format!("failed spawning CLI command `{}`", cli.command))?;

    let stdout = child
        .stdout
        .take()
        .ok_or_else(|| anyhow::anyhow!("failed capturing CLI stdout for `{}`", cli.command))?;
    let stderr = child
        .stderr
        .take()
        .ok_or_else(|| anyhow::anyhow!("failed capturing CLI stderr for `{}`", cli.command))?;
    let max_bytes = prepared.transport.max_response_bytes;

    let stdout_command = cli.command.clone();
    let stdout_reader = tokio::spawn(async move {
        read_stream_limited(stdout, max_bytes, "stdout", &stdout_command).await
    });
    let stderr_command = cli.command.clone();
    let stderr_reader = tokio::spawn(async move {
        read_stream_limited(stderr, max_bytes, "stderr", &stderr_command).await
    });

    if let Some(input) = &cli.stdin
        && let Some(mut stdin) = child.stdin.take()
    {
        stdin
            .write_all(input.as_bytes())
            .await
            .with_context(|| format!("failed writing stdin for CLI command `{}`", cli.command))?;
    }

    let status = match tokio::time::timeout(prepared.transport.timeout, child.wait()).await {
        Ok(wait_result) => wait_result
            .with_context(|| format!("failed waiting for CLI command `{}`", cli.command))?,
        Err(_) => {
            let _ = child.kill().await;
            let _ = child.wait().await;
            bail!(
                "CLI command `{}` timed out after {:?}",
                cli.command,
                prepared.transport.timeout
            );
        }
    };

    let stdout = stdout_reader
        .await
        .context("failed joining stdout reader task")??;
    let stderr = stderr_reader
        .await
        .context("failed joining stderr reader task")??;

    let status = status
        .code()
        .map(|code| code.clamp(0, u16::MAX as i32) as u16)
        .unwrap_or(1);
    let bytes = if stdout.is_empty() { &stderr } else { &stdout };

    let decoded_body = decode_response(prepared.result_template.decode, None, bytes)
        .context("failed decoding CLI command output")?;
    let extracted = extract_result(prepared.result_template.extract.as_ref(), &decoded_body)?;

    Ok(ExecutionResult {
        status,
        url: format!("cli://{}", cli.command),
        result: extracted,
        decoded: decoded_body.to_json_value(),
    })
}

#[derive(Debug)]
struct RawHttpExecutionResult {
    status: u16,
    url: String,
    body: Vec<u8>,
    content_type: Option<String>,
}

async fn execute_http_once_with_host_validator<F, Fut>(
    prepared: &PreparedRequest,
    host_validator: &mut F,
) -> Result<ExecutionResult>
where
    F: FnMut(Url) -> Fut,
    Fut: Future<Output = Result<Vec<IpAddr>>>,
{
    let mut method = prepared.method.clone();
    let mut body = prepared.body.clone();
    let mut url = prepared.url.clone();

    for hop in 0..=prepared.transport.max_redirect_hops {
        ensure_url_allowed(&url, &prepared.allow_rules)?;
        let resolved_ips = host_validator(url.clone()).await?;
        let client = build_http_client(prepared, &url, &resolved_ips)?;

        let request = build_request(
            &client,
            &method,
            &url,
            &prepared.headers,
            &prepared.cookies,
            &prepared.query,
            &body,
        )?;
        let response = request
            .send()
            .await
            .with_context(|| format!("request execution failed for `{}`", url.as_str()))?;

        if response.status().is_redirection() && prepared.transport.follow_redirects {
            if hop >= prepared.transport.max_redirect_hops {
                bail!(
                    "maximum redirect hops reached ({})",
                    prepared.transport.max_redirect_hops
                );
            }

            let location = response
                .headers()
                .get(LOCATION)
                .ok_or_else(|| anyhow::anyhow!("redirect response missing Location header"))?
                .to_str()
                .context("redirect Location header is not valid UTF-8")?
                .to_string();

            let new_url = url
                .join(&location)
                .with_context(|| format!("invalid redirect Location `{location}`"))?;

            let status = response.status().as_u16();
            if status == 303
                || ((status == 301 || status == 302) && method == reqwest::Method::POST)
            {
                method = reqwest::Method::GET;
                body = PreparedBody::Empty;
            }
            url = new_url;
            continue;
        }

        let status = response.status().as_u16();
        let content_type = response
            .headers()
            .get(CONTENT_TYPE)
            .and_then(|v| v.to_str().ok())
            .map(|v| v.to_string());

        let body_bytes =
            read_response_body_limited(response, prepared.transport.max_response_bytes).await?;
        let raw = RawHttpExecutionResult {
            status,
            url: url.to_string(),
            body: body_bytes,
            content_type,
        };

        let decoded_body = decode_response(
            prepared.result_template.decode,
            raw.content_type.as_deref(),
            &raw.body,
        )
        .context("failed decoding HTTP response")?;
        let extracted = extract_result(prepared.result_template.extract.as_ref(), &decoded_body)?;

        return Ok(ExecutionResult {
            status: raw.status,
            url: raw.url,
            result: extracted,
            decoded: decoded_body.to_json_value(),
        });
    }

    bail!("redirect handling failed unexpectedly")
}

async fn execute_grpc_once_with_host_validator<F, Fut>(
    prepared: &PreparedRequest,
    host_validator: &mut F,
) -> Result<ExecutionResult>
where
    F: FnMut(Url) -> Fut,
    Fut: Future<Output = Result<Vec<IpAddr>>>,
{
    let grpc = prepared
        .grpc
        .as_ref()
        .ok_or_else(|| anyhow!("prepared gRPC request missing grpc data"))?;
    let rpc_url = grpc_method_url(&prepared.url, &grpc.service, &grpc.method)?;

    ensure_url_allowed(&rpc_url, &prepared.allow_rules)?;
    let resolved_ips = host_validator(rpc_url.clone()).await?;
    if resolved_ips.is_empty() {
        bail!("host validation returned no resolved IP addresses");
    }
    if prepared.transport.proxy_url.is_some() {
        bail!("gRPC transport proxy_profile is not supported");
    }
    if prepared.transport.tls_min_version.is_some() {
        bail!("gRPC transport.tls.min_version is not supported");
    }

    let endpoint = tonic::transport::Endpoint::new(prepared.url.to_string())
        .context("invalid gRPC endpoint URL")?
        .timeout(prepared.transport.timeout)
        .connect_timeout(prepared.transport.timeout);

    let channel: tonic::transport::Channel = endpoint
        .connect()
        .await
        .with_context(|| format!("failed connecting to gRPC endpoint `{}`", prepared.url))?;

    let dynamic_request = DynamicRequest {
        body: grpc_request_body(prepared)?,
        headers: prepared.headers.clone(),
        service: grpc.service.clone(),
        method: grpc.method.clone(),
    };

    let dynamic_response = if let Some(descriptor_set) = &grpc.descriptor_set {
        let mut client = GrancClient::from(channel)
            .with_file_descriptor(descriptor_set.clone())
            .context("invalid gRPC descriptor set bytes")?;
        client
            .dynamic(dynamic_request)
            .await
            .map_err(|err| anyhow!("gRPC request failed: {err}"))?
    } else {
        let mut client = GrancClient::from(channel);
        client
            .dynamic(dynamic_request)
            .await
            .map_err(|err| anyhow!("gRPC reflection request failed: {err}"))?
    };

    let (status, payload) = normalize_dynamic_response(dynamic_response);
    let payload_bytes = serde_json::to_vec(&payload).context("failed serializing gRPC payload")?;
    if payload_bytes.len() > prepared.transport.max_response_bytes {
        bail!(
            "gRPC response exceeded configured max_response_bytes ({} bytes)",
            prepared.transport.max_response_bytes
        );
    }

    let decoded_body = decode_response(
        prepared.result_template.decode,
        Some("application/json"),
        &payload_bytes,
    )
    .context("failed decoding gRPC response")?;
    let extracted = extract_result(prepared.result_template.extract.as_ref(), &decoded_body)?;

    Ok(ExecutionResult {
        status,
        url: rpc_url.to_string(),
        result: extracted,
        decoded: decoded_body.to_json_value(),
    })
}

fn build_request(
    client: &reqwest::Client,
    method: &reqwest::Method,
    url: &Url,
    headers: &[(String, String)],
    cookies: &[(String, String)],
    query: &[(String, String)],
    body: &PreparedBody,
) -> Result<reqwest::RequestBuilder> {
    let mut builder = client.request(method.clone(), url.clone());

    if !query.is_empty() {
        builder = builder.query(query);
    }

    let mut header_map = HeaderMap::new();
    for (name, value) in headers {
        let header_name = HeaderName::from_bytes(name.as_bytes())
            .with_context(|| format!("invalid header name `{name}`"))?;
        let header_value = HeaderValue::from_str(value)
            .with_context(|| format!("invalid header value for `{name}`"))?;
        header_map.append(header_name, header_value);
    }

    if !cookies.is_empty() {
        let cookie_value = cookies
            .iter()
            .map(|(k, v)| format!("{k}={v}"))
            .collect::<Vec<_>>()
            .join("; ");
        header_map.insert(
            COOKIE,
            HeaderValue::from_str(&cookie_value).context("invalid cookie header value")?,
        );
    }

    builder = builder.headers(header_map);

    match body {
        PreparedBody::Empty => {}
        PreparedBody::Json(value) => {
            builder = builder.json(value);
        }
        PreparedBody::Form(fields) => {
            builder = builder.form(fields);
        }
        PreparedBody::Multipart(parts) => {
            builder = builder.multipart(build_multipart(parts)?);
        }
        PreparedBody::RawBytes {
            bytes,
            content_type,
        } => {
            if let Some(content_type) = content_type {
                builder = builder.header(CONTENT_TYPE, content_type);
            }
            builder = builder.body(bytes.clone());
        }
    }

    Ok(builder)
}

fn build_http_client(
    prepared: &PreparedRequest,
    url: &Url,
    resolved_ips: &[IpAddr],
) -> Result<reqwest::Client> {
    if resolved_ips.is_empty() {
        bail!("host validation returned no resolved IP addresses");
    }

    let mut builder = reqwest::Client::builder()
        .timeout(prepared.transport.timeout)
        .redirect(reqwest::redirect::Policy::none())
        .gzip(prepared.transport.compression)
        .brotli(prepared.transport.compression)
        .zstd(prepared.transport.compression)
        .deflate(prepared.transport.compression);

    if let Some(version) = prepared.transport.tls_min_version {
        builder = builder.min_tls_version(version);
    }

    if let Some(proxy_url) = &prepared.transport.proxy_url {
        let proxy = reqwest::Proxy::all(proxy_url)
            .with_context(|| format!("invalid proxy URL `{proxy_url}`"))?;
        builder = builder.proxy(proxy);
    }

    let host = url
        .host_str()
        .ok_or_else(|| anyhow::anyhow!("request URL missing host"))?;
    let port = url
        .port_or_known_default()
        .ok_or_else(|| anyhow::anyhow!("request URL missing port"))?;

    if !resolved_ips.is_empty() {
        let addrs: Vec<SocketAddr> = resolved_ips
            .iter()
            .map(|ip| SocketAddr::new(*ip, port))
            .collect();
        builder = builder.resolve_to_addrs(host, &addrs);
    }

    builder
        .build()
        .context("failed constructing reqwest client")
}

async fn read_response_body_limited(
    mut response: reqwest::Response,
    limit: usize,
) -> Result<Vec<u8>> {
    let mut out = Vec::new();
    while let Some(chunk) = response.chunk().await? {
        if out.len().saturating_add(chunk.len()) > limit {
            bail!("response body exceeded configured max_response_bytes ({limit} bytes)");
        }
        out.extend_from_slice(&chunk);
    }
    Ok(out)
}

async fn read_stream_limited<R>(
    mut reader: R,
    limit: usize,
    label: &str,
    command: &str,
) -> Result<Vec<u8>>
where
    R: AsyncRead + Unpin,
{
    let mut out = Vec::new();
    let mut buf = [0_u8; 8192];

    loop {
        let bytes_read = reader
            .read(&mut buf)
            .await
            .with_context(|| format!("failed reading CLI {label} for `{command}`"))?;
        if bytes_read == 0 {
            break;
        }

        if out.len().saturating_add(bytes_read) > limit {
            bail!(
                "CLI {label} for `{command}` exceeded configured max_response_bytes ({limit} bytes)"
            );
        }
        out.extend_from_slice(&buf[..bytes_read]);
    }

    Ok(out)
}

fn build_multipart(parts: &[PreparedMultipartPart]) -> Result<reqwest::multipart::Form> {
    let mut form = reqwest::multipart::Form::new();
    for part in parts {
        let mut req_part = reqwest::multipart::Part::bytes(part.bytes.clone());
        if let Some(content_type) = &part.content_type {
            req_part = req_part.mime_str(content_type)?;
        }
        if let Some(filename) = &part.filename {
            req_part = req_part.file_name(filename.clone());
        }
        form = form.part(part.name.clone(), req_part);
    }
    Ok(form)
}

fn backoff(base: Duration, attempt: usize) -> Duration {
    let factor = attempt.max(1) as u32;
    base.saturating_mul(factor)
}

fn grpc_method_url(base: &Url, service: &str, method: &str) -> Result<Url> {
    let service = service.trim();
    let method = method.trim();
    if service.is_empty() || method.is_empty() {
        bail!("gRPC service and method must not be empty");
    }

    let mut rpc_url = base.clone();
    let prefix = rpc_url.path().trim_end_matches('/');
    let path = if prefix.is_empty() || prefix == "/" {
        format!("/{service}/{method}")
    } else {
        format!("{prefix}/{service}/{method}")
    };
    rpc_url.set_path(&path);
    Ok(rpc_url)
}

fn grpc_request_body(prepared: &PreparedRequest) -> Result<Value> {
    match &prepared.body {
        PreparedBody::Json(value) => Ok(value.clone()),
        PreparedBody::Empty => Ok(Value::Object(Default::default())),
        _ => bail!("gRPC request body must be JSON"),
    }
}

fn normalize_dynamic_response(response: DynamicResponse) -> (u16, Value) {
    match response {
        DynamicResponse::Unary(Ok(value)) => (0, value),
        DynamicResponse::Unary(Err(status)) => {
            (grpc_code(status.code()), grpc_status_payload(&status))
        }
        DynamicResponse::Streaming(Ok(values)) => {
            let mut status = 0_u16;
            let mut payload = Vec::with_capacity(values.len());
            for value in values {
                match value {
                    Ok(item) => payload.push(item),
                    Err(err) => {
                        if status == 0 {
                            status = grpc_code(err.code());
                        }
                        payload.push(grpc_status_payload(&err));
                    }
                }
            }
            (status, Value::Array(payload))
        }
        DynamicResponse::Streaming(Err(status)) => {
            (grpc_code(status.code()), grpc_status_payload(&status))
        }
    }
}

fn grpc_status_payload(status: &tonic::Status) -> Value {
    json!({
        "grpc_code": status.code().to_string(),
        "grpc_code_number": grpc_code(status.code()),
        "message": status.message(),
    })
}

fn grpc_code(code: tonic::Code) -> u16 {
    code as i32 as u16
}
