pub use crate::client::http::reqwest::remote::Options;

pub struct Remote {
    /// A worker thread which performs the actual request.
    handle: Option<std::thread::JoinHandle<Result<(), remote::Error>>>,
    /// A channel to send requests (work) to the worker thread.
    request: std::sync::mpsc::SyncSender<remote::Request>,
    /// A channel to receive the result of the prior request.
    response: std::sync::mpsc::Receiver<remote::Response>,
    /// A mechanism for configuring the remote.
    config: Options,
}

mod remote {
    use std::{
        any::Any,
        convert::TryFrom,
        io::Write,
        str::FromStr,
        sync::{Arc, Mutex},
    };

    use git_features::io::pipe;

    use crate::client::{http, http::reqwest::Remote};

    #[derive(Debug, thiserror::Error)]
    pub enum Error {
        #[error(transparent)]
        Reqwest(#[from] reqwest::Error),
        #[error("Request configuration failed")]
        ConfigureRequest(#[from] Box<dyn std::error::Error + Send + Sync + 'static>),
    }

    impl Default for Remote {
        fn default() -> Self {
            let (req_send, req_recv) = std::sync::mpsc::sync_channel(0);
            let (res_send, res_recv) = std::sync::mpsc::sync_channel(0);
            let handle = std::thread::spawn(move || -> Result<(), Error> {
                for Request {
                    url,
                    headers,
                    upload,
                    config,
                } in req_recv
                {
                    // We may error while configuring, which is expected as part of the internal protocol. The error will be
                    // received and the sender of the request might restart us.
                    let client = reqwest::blocking::ClientBuilder::new()
                        .connect_timeout(std::time::Duration::from_secs(20))
                        .build()?;
                    let mut req_builder = if upload { client.post(url) } else { client.get(url) }.headers(headers);
                    let (post_body_tx, post_body_rx) = pipe::unidirectional(0);
                    if upload {
                        req_builder = req_builder.body(reqwest::blocking::Body::new(post_body_rx));
                    }
                    let mut req = req_builder.build()?;
                    let (mut response_body_tx, response_body_rx) = pipe::unidirectional(0);
                    let (mut headers_tx, headers_rx) = pipe::unidirectional(0);
                    if res_send
                        .send(Response {
                            headers: headers_rx,
                            body: response_body_rx,
                            upload_body: post_body_tx,
                        })
                        .is_err()
                    {
                        // This means our internal protocol is violated as the one who sent the request isn't listening anymore.
                        // Shut down as something is off.
                        break;
                    }
                    if let Some(mutex) = config.configure_request {
                        let mut configure_request = mutex.lock().expect("our thread cannot ordinarily panic");
                        configure_request(&mut req)?;
                    }
                    let mut res = match client.execute(req).and_then(|res| res.error_for_status()) {
                        Ok(res) => res,
                        Err(err) => {
                            let (kind, err) = match err.status() {
                                Some(status) => {
                                    let kind = if status == reqwest::StatusCode::UNAUTHORIZED {
                                        std::io::ErrorKind::PermissionDenied
                                    } else {
                                        std::io::ErrorKind::Other
                                    };
                                    (kind, format!("Received HTTP status {}", status.as_str()))
                                }
                                None => (std::io::ErrorKind::Other, err.to_string()),
                            };
                            let err = Err(std::io::Error::new(kind, err));
                            headers_tx.channel.send(err).ok();
                            continue;
                        }
                    };

                    let send_headers = {
                        let headers = res.headers();
                        move || -> std::io::Result<()> {
                            for (name, value) in headers {
                                headers_tx.write_all(name.as_str().as_bytes())?;
                                headers_tx.write_all(b":")?;
                                headers_tx.write_all(value.as_bytes())?;
                                headers_tx.write_all(b"\n")?;
                            }
                            // Make sure this is an FnOnce closure to signal the remote reader we are done.
                            drop(headers_tx);
                            Ok(())
                        }
                    };

                    // We don't have to care if anybody is receiving the header, as a matter of fact we cannot fail sending them.
                    // Thus an error means the receiver failed somehow, but might also have decided not to read headers at all. Fine with us.
                    send_headers().ok();

                    // reading the response body is streaming and may fail for many reasons. If so, we send the error over the response
                    // body channel and that's all we can do.
                    if let Err(err) = std::io::copy(&mut res, &mut response_body_tx) {
                        response_body_tx.channel.send(Err(err)).ok();
                    }
                }
                Ok(())
            });

            Remote {
                handle: Some(handle),
                request: req_send,
                response: res_recv,
                config: Options::default(),
            }
        }
    }

    /// utilities
    impl Remote {
        fn make_request(
            &mut self,
            url: &str,
            headers: impl IntoIterator<Item = impl AsRef<str>>,
            upload: bool,
        ) -> Result<http::PostResponse<pipe::Reader, pipe::Reader, pipe::Writer>, http::Error> {
            let mut header_map = reqwest::header::HeaderMap::new();
            for header_line in headers {
                let header_line = header_line.as_ref();
                let colon_pos = header_line
                    .find(':')
                    .expect("header line must contain a colon to separate key and value");
                let header_name = &header_line[..colon_pos];
                let value = &header_line[colon_pos + 1..];

                match reqwest::header::HeaderName::from_str(header_name)
                    .ok()
                    .zip(reqwest::header::HeaderValue::try_from(value.trim()).ok())
                {
                    Some((key, val)) => header_map.insert(key, val),
                    None => continue,
                };
            }
            self.request
                .send(Request {
                    url: url.to_owned(),
                    headers: header_map,
                    upload,
                    config: self.config.clone(),
                })
                .expect("the remote cannot be down at this point");

            let Response {
                headers,
                body,
                upload_body,
            } = match self.response.recv() {
                Ok(res) => res,
                Err(_) => {
                    let err = self
                        .handle
                        .take()
                        .expect("always present")
                        .join()
                        .expect("no panic")
                        .expect_err("no receiver means thread is down with init error");
                    *self = Self::default();
                    return Err(http::Error::InitHttpClient { source: Box::new(err) });
                }
            };

            Ok(http::PostResponse {
                post_body: upload_body,
                headers,
                body,
            })
        }
    }

    impl http::Http for Remote {
        type Headers = pipe::Reader;
        type ResponseBody = pipe::Reader;
        type PostBody = pipe::Writer;

        fn get(
            &mut self,
            url: &str,
            headers: impl IntoIterator<Item = impl AsRef<str>>,
        ) -> Result<http::GetResponse<Self::Headers, Self::ResponseBody>, http::Error> {
            self.make_request(url, headers, false).map(Into::into)
        }

        fn post(
            &mut self,
            url: &str,
            headers: impl IntoIterator<Item = impl AsRef<str>>,
        ) -> Result<http::PostResponse<Self::Headers, Self::ResponseBody, Self::PostBody>, http::Error> {
            self.make_request(url, headers, true)
        }

        fn configure(&mut self, config: &dyn Any) -> Result<(), Box<dyn std::error::Error + Send + Sync + 'static>> {
            if let Some(config) = config.downcast_ref::<Options>() {
                self.config = config.clone();
            }
            Ok(())
        }
    }

    /// Options to configure the reqwest HTTP handler.
    #[derive(Default, Clone)]
    pub struct Options {
        /// A function to configure the request that is about to be made.
        pub configure_request: Option<
            Arc<
                Mutex<
                    dyn FnMut(
                            &mut reqwest::blocking::Request,
                        ) -> Result<(), Box<dyn std::error::Error + Send + Sync + 'static>>
                        + Send
                        + Sync
                        + 'static,
                >,
            >,
        >,
    }

    pub struct Request {
        pub url: String,
        pub headers: reqwest::header::HeaderMap,
        pub upload: bool,
        pub config: Options,
    }

    /// A link to a thread who provides data for the contained readers.
    /// The expected order is:
    /// - write `upload_body`
    /// - read `headers` to end
    /// - read `body` to hend
    pub struct Response {
        pub headers: pipe::Reader,
        pub body: pipe::Reader,
        pub upload_body: pipe::Writer,
    }
}