use std::collections::HashMap;
use std::convert::Infallible;
use std::future::Future;
use std::io;
use std::pin::Pin;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};

use avril_caps::Capabilities;
use avril_core::serde_v8::{JsBuffer, ToJsBuffer};
use avril_core::{Extension, Op};
use bytes::Bytes;
use futures::StreamExt;
use http_body_util::{BodyExt, Full, StreamBody, combinators::UnsyncBoxBody};
use hyper::{
    Method, Request,
    body::Frame,
    header::{HeaderMap, HeaderName, HeaderValue},
};
use hyper_rustls::{HttpsConnector, HttpsConnectorBuilder};
use hyper_util::{
    client::legacy::{Client, connect::HttpConnector},
    rt::TokioExecutor,
};
use tokio::io::{AsyncRead, AsyncReadExt, BufReader};
use tokio::sync::{Mutex as AsyncMutex, mpsc};
use tokio_util::{io::StreamReader, sync::CancellationToken};

type Error = Box<dyn std::error::Error + Send + Sync>;
type Body = UnsyncBoxBody<Bytes, Error>;
type Reader = Pin<Box<dyn AsyncRead + Send>>;
/// Status, status text, headers, final URL, whether redirected, and whether a redirect
/// abandoned the streamed request body.
type Metadata = (u16, String, Vec<(String, String)>, String, bool, bool);
type SendArgs = (
    u64,
    String,
    String,
    Vec<(String, String)>,
    Option<JsBuffer>,
    bool,
    String,
);
type Upload = mpsc::Sender<Result<Frame<Bytes>, Error>>;

const MAX_ID: u64 = 9_007_199_254_740_991;

enum Redirect {
    Follow,
    Error,
    Manual,
}

impl Redirect {
    fn parse(mode: &str) -> Result<Self, Error> {
        match mode {
            "follow" => Ok(Self::Follow),
            "error" => Ok(Self::Error),
            "manual" => Ok(Self::Manual),
            _ => Err("Invalid redirect mode".into()),
        }
    }
}

enum RequestBody {
    Empty,
    Bytes(Bytes),
    Upload(mpsc::Receiver<Result<Frame<Bytes>, Error>>),
}

struct PendingRequest {
    url: String,
    method: String,
    headers: Vec<(String, String)>,
    body: RequestBody,
    redirect: String,
}

struct Resource {
    canceled: CancellationToken,
    body: AsyncMutex<Option<Reader>>,
    upload: Mutex<Option<Upload>>,
    /// Releases writes blocked on a request body that will no longer be sent.
    upload_closed: CancellationToken,
}

impl Resource {
    fn close_upload(&self) {
        self.upload.lock().unwrap().take();
        self.upload_closed.cancel();
    }
}

struct Fetch {
    runtime: tokio::runtime::Handle,
    client: Client<HttpsConnector<HttpConnector>, Body>,
    resources: Arc<Mutex<HashMap<u64, Arc<Resource>>>>,
    next: AtomicU64,
    caps: Arc<Capabilities>,
}

impl Fetch {
    fn resource(&self, id: u64) -> Result<Arc<Resource>, Error> {
        self.resources
            .lock()
            .unwrap()
            .get(&id)
            .cloned()
            .ok_or_else(|| "Fetch resource is closed".into())
    }

    fn run<T: Send + 'static, F: Future<Output = Result<T, Error>> + Send + 'static>(
        &self,
        future: F,
    ) -> impl Future<Output = Result<T, Error>> + use<T, F> {
        let handle = self.runtime.spawn(future);
        async move {
            // Dropping the handle detaches the task rather than aborting it.
            handle
                .await
                .map_err(|_| io::Error::other("Fetch worker stopped"))?
        }
    }

    fn cancel(&self, id: u64) {
        if let Some(resource) = self.resources.lock().unwrap().remove(&id) {
            resource.canceled.cancel();
            resource.close_upload();
        }
    }

    fn send(
        &self,
        (id, url, method, headers, bytes, streaming, redirect): SendArgs,
    ) -> impl Future<Output = Result<Metadata, Error>> + use<> {
        let resource = self.resource(id);
        let client = self.client.clone();
        let caps = self.caps.clone();
        // Install the upload channel before returning the promise, so JS can submit its first chunk.
        let body = if streaming {
            let (send, receive) = mpsc::channel(1);
            if let Ok(resource) = &resource {
                *resource.upload.lock().unwrap() = Some(send);
            }
            RequestBody::Upload(receive)
        } else {
            bytes
                .map(|bytes| Bytes::copy_from_slice(&bytes))
                .map(RequestBody::Bytes)
                .unwrap_or(RequestBody::Empty)
        };
        let request = PendingRequest {
            url,
            method,
            headers,
            body,
            redirect,
        };
        self.run(async move {
            let resource = resource?;
            let canceled = resource.canceled.clone();
            let result = tokio::select! {
                _ = canceled.cancelled() => Err("Fetch aborted".into()),
                result = send_request(client, caps, &resource, request) => result,
            };
            result
        })
    }
}

async fn send_request(
    client: Client<HttpsConnector<HttpConnector>, Body>,
    caps: Arc<Capabilities>,
    resource: &Resource,
    request: PendingRequest,
) -> Result<Metadata, Error> {
    let PendingRequest {
        url,
        method,
        headers,
        body,
        redirect,
    } = request;
    let url = url::Url::parse(&url)?;
    let method = Method::from_bytes(method.as_bytes())?;
    if ["CONNECT", "TRACE", "TRACK"]
        .iter()
        .any(|name| method.as_str().eq_ignore_ascii_case(name))
    {
        return Err("Forbidden HTTP method".into());
    }
    let redirect = Redirect::parse(&redirect)?;
    if !matches!(body, RequestBody::Empty) && (method == Method::GET || method == Method::HEAD) {
        return Err("GET/HEAD requests cannot have a body".into());
    }
    let mut list = HeaderMap::new();
    for (name, value) in headers {
        let name = name.to_ascii_lowercase();
        if [
            "host",
            "content-length",
            "transfer-encoding",
            "connection",
            "upgrade",
            "te",
            "trailer",
            "keep-alive",
            "expect",
        ]
        .contains(&name.as_str())
            || name.starts_with("proxy-")
        {
            return Err(format!("Forbidden request header: {name}").into());
        }
        let value = value
            .chars()
            .map(|c| u8::try_from(c as u32))
            .collect::<Result<Vec<_>, _>>()?;
        list.append(
            HeaderName::from_bytes(name.as_bytes())?,
            HeaderValue::from_bytes(&value)?,
        );
    }
    if !list.contains_key("accept") {
        list.insert("accept", HeaderValue::from_static("*/*"));
    }
    if !list.contains_key("accept-encoding") {
        list.insert(
            "accept-encoding",
            HeaderValue::from_static("gzip, br, deflate"),
        );
    }
    let streaming = matches!(body, RequestBody::Upload(_));
    let (mut url, mut method, mut body) = (url, method, body);
    let mut abandoned = false;
    for hop in 0..=20 {
        if !["http", "https"].contains(&url.scheme()) {
            return Err("not yet implemented: non-HTTP fetch".into());
        }
        if !url.username().is_empty() || url.password().is_some() {
            return Err("Fetch URLs cannot contain credentials".into());
        }
        url.set_fragment(None);
        // No ambient proxies, automatic redirects, or snapshot permissions in the connector.
        caps.check_net(
            url.host_str().ok_or("Missing fetch host")?,
            url.port_or_known_default().ok_or("Missing fetch port")?,
        )?;
        let request_body = match std::mem::replace(&mut body, RequestBody::Empty) {
            RequestBody::Upload(receive) => {
                let stream = futures::stream::unfold(receive, |mut receive| async move {
                    receive.recv().await.map(|frame| (frame, receive))
                });
                StreamBody::new(stream).boxed_unsync()
            }
            RequestBody::Bytes(bytes) => {
                let payload = Full::new(bytes.clone())
                    .map_err(|never| match never {})
                    .boxed_unsync();
                body = RequestBody::Bytes(bytes);
                payload
            }
            RequestBody::Empty => Full::new(Bytes::new())
                .map_err(|never| match never {})
                .boxed_unsync(),
        };
        let mut request = Request::builder()
            .method(method.clone())
            .uri(url.as_str())
            .body(request_body)?;
        *request.headers_mut() = list.clone();
        let response = client.request(request).await?;
        let status = response.status();
        if matches!(status.as_u16(), 301 | 302 | 303 | 307 | 308)
            && !matches!(redirect, Redirect::Manual)
            && let Some(location) = response.headers().get("location")
        {
            if matches!(redirect, Redirect::Error) {
                return Err("Redirect encountered in redirect:error mode".into());
            }
            if hop == 20 {
                return Err("Too many redirects".into());
            }
            let location = String::from_utf8_lossy(location.as_bytes());
            let next = url.join(&location)?;
            if url.origin() != next.origin() {
                for name in ["authorization", "proxy-authorization", "cookie", "host"] {
                    list.remove(name);
                }
            }
            if ((status.as_u16() == 301 || status.as_u16() == 302) && method == Method::POST)
                || (status.as_u16() == 303 && method != Method::GET && method != Method::HEAD)
            {
                method = Method::GET;
                body = RequestBody::Empty;
                for name in [
                    "content-type",
                    "content-length",
                    "content-encoding",
                    "content-language",
                    "content-location",
                ] {
                    list.remove(name);
                }
                resource.close_upload();
                abandoned = streaming;
            } else if streaming && method != Method::GET && method != Method::HEAD {
                return Err(
                    "not yet implemented: replaying streaming request bodies on redirects".into(),
                );
            }
            url = next;
            continue;
        }
        let headers = response
            .headers()
            .iter()
            .map(|(key, value)| {
                (
                    key.to_string(),
                    value.as_bytes().iter().map(|b| char::from(*b)).collect(),
                )
            })
            .collect();
        let status_text = response
            .extensions()
            .get::<hyper::ext::ReasonPhrase>()
            .map(|reason| {
                reason
                    .as_bytes()
                    .iter()
                    .map(|byte| char::from(*byte))
                    .collect::<String>()
            })
            .unwrap_or_else(|| status.canonical_reason().unwrap_or("").into());
        let encoding = response
            .headers()
            .get_all("content-encoding")
            .iter()
            .map(|value| value.to_str())
            .collect::<Result<Vec<_>, _>>()?
            .join(",");
        let stream = response
            .into_body()
            .into_data_stream()
            .map(|result| result.map_err(io::Error::other));
        let mut reader: Reader = Box::pin(StreamReader::new(stream));
        if method != Method::HEAD && !matches!(status.as_u16(), 204 | 205 | 304) {
            for encoding in encoding.split(',').rev().map(str::trim) {
                reader = match encoding.to_ascii_lowercase().as_str() {
                    "" | "identity" => reader,
                    "gzip" => {
                        let mut decoder = async_compression::tokio::bufread::GzipDecoder::new(
                            BufReader::new(reader),
                        );
                        decoder.multiple_members(true);
                        Box::pin(decoder)
                    }
                    "br" => Box::pin(async_compression::tokio::bufread::BrotliDecoder::new(
                        BufReader::new(reader),
                    )),
                    "deflate" => Box::pin(async_compression::tokio::bufread::ZlibDecoder::new(
                        BufReader::new(reader),
                    )),
                    _ => return Err("not yet implemented: response content encoding".into()),
                };
            }
        }
        *resource.body.lock().await = Some(reader);
        return Ok((
            status.as_u16(),
            status_text,
            headers,
            url.into(),
            hop != 0,
            abandoned,
        ));
    }
    unreachable!()
}

pub fn init(runtime: tokio::runtime::Handle, caps: Arc<Capabilities>, base: String) -> Extension {
    let connector = HttpsConnectorBuilder::new()
        .with_webpki_roots()
        .https_or_http()
        .enable_http1()
        .enable_http2()
        .build();
    let client = Client::builder(TokioExecutor::new()).build(connector);
    let state = Arc::new(Fetch {
        runtime,
        client,
        resources: Arc::new(Mutex::new(HashMap::new())),
        next: AtomicU64::new(1),
        caps,
    });
    let create = state.clone();
    let send = state.clone();
    let read = state.clone();
    let collect = state.clone();
    let write = state.clone();
    Extension {
        name: "ext:fetch",
        dependencies: &[
            "ext:encoding",
            "ext:http_types",
            "ext:timers",
            "ext:url",
            "ext:web",
            "ext:webstreams",
        ],
        ops: vec![
            Op::new("op_fetch_base", move |(): ()| {
                Ok::<_, Infallible>(base.clone())
            }),
            Op::new("op_fetch_create", move |_: ()| {
                let id = create
                    .next
                    .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |id| {
                        (id <= MAX_ID).then_some(id + 1)
                    })
                    .map_err(|_| io::Error::other("Fetch IDs exhausted"))?;
                create.resources.lock().unwrap().insert(
                    id,
                    Arc::new(Resource {
                        canceled: CancellationToken::new(),
                        body: AsyncMutex::new(None),
                        upload: Mutex::new(None),
                        upload_closed: CancellationToken::new(),
                    }),
                );
                Ok::<_, io::Error>(id)
            }),
            Op::asynchronous("op_fetch_send", move |args: SendArgs| {
                let pending = send.send(args);
                async move { pending.await.map_err(io::Error::other) }
            }),
            Op::asynchronous("op_fetch_read", move |id: u64| {
                let resource = read.resource(id);
                let resources = read.resources.clone();
                let pending = read.run(async move {
                    let resource = resource?;
                    tokio::select! {
                        _ = resource.canceled.cancelled() => Err("Fetch aborted".into()),
                        result = async {
                            let mut body = resource.body.lock().await;
                            let mut bytes = vec![0; 64 * 1024];
                            let count = body.as_mut().ok_or("Response body is unavailable")?.read(&mut bytes).await?;
                            bytes.truncate(count);
                            if count == 0 { body.take(); resources.lock().unwrap().remove(&id); }
                            Ok::<_, Error>(bytes)
                        } => result,
                    }
                });
                async move {
                    Ok::<_, io::Error>(match pending.await.map_err(io::Error::other)? {
                        bytes if bytes.is_empty() => None,
                        bytes => Some(ToJsBuffer::from(bytes)),
                    })
                }
            }),
            Op::asynchronous("op_fetch_read_all", move |id: u64| {
                let resource = collect.resource(id);
                let resources = collect.resources.clone();
                let pending = collect.run(async move {
                    let resource = resource?;
                    let result = tokio::select! {
                        biased;
                        _ = resource.canceled.cancelled() => Err("Fetch aborted".into()),
                        result = async {
                            let mut reader = resource.body.lock().await.take()
                                .ok_or("Response body is unavailable")?;
                            let mut bytes = Vec::new();
                            reader.read_to_end(&mut bytes).await?;
                            Ok::<_, Error>(bytes)
                        } => result,
                    };
                    resources.lock().unwrap().remove(&id);
                    result
                });
                async move {
                    pending
                        .await
                        .map(ToJsBuffer::from)
                        .map_err(io::Error::other)
                }
            }),
            Op::asynchronous(
                "op_fetch_write",
                move |(id, bytes): (u64, Option<JsBuffer>)| {
                    let resource = write.resource(id);
                    let bytes = bytes.map(|bytes| Bytes::copy_from_slice(&bytes));
                    let pending = write.run(async move {
                        let Ok(resource) = resource else {
                            return Ok(false);
                        };
                        if let Some(bytes) = bytes {
                            let Some(upload) = resource.upload.lock().unwrap().clone() else {
                                return Ok(false);
                            };
                            tokio::select! {
                                _ = resource.canceled.cancelled() => Err("Fetch aborted".into()),
                                _ = resource.upload_closed.cancelled() => Ok(false),
                                result = upload.send(Ok(Frame::data(bytes))) => Ok(result.is_ok()),
                            }
                        } else {
                            Ok(resource.upload.lock().unwrap().take().is_some())
                        }
                    });
                    async move { pending.await.map_err(io::Error::other) }
                },
            ),
            Op::new("op_fetch_cancel", move |id: u64| {
                state.cancel(id);
                Ok::<_, io::Error>(())
            }),
        ],
        modules: &[
            (
                "ext:fetch/main.js",
                include_str!(concat!(env!("OUT_DIR"), "/main.js")),
            ),
            (
                "ext:fetch/fetch.js",
                include_str!(concat!(env!("OUT_DIR"), "/fetch.js")),
            ),
            (
                "ext:fetch/eventsource.js",
                include_str!(concat!(env!("OUT_DIR"), "/eventsource.js")),
            ),
        ],
        entry_point: Some("ext:fetch/main.js"),
    }
}