use std::{sync::Arc, time::Duration};

use awc::{
    error::{ConnectError, PayloadError, SendRequestError},
    http::StatusCode,
    http::{header::HeaderValue, Method},
    Client, ClientRequest, ClientResponse, Connector,
};
use bytes::Bytes;
use futures::prelude::*;
use libsignal_service::{
    configuration::*, prelude::ProtobufMessage, push_service::*,
    websocket::SignalWebSocket,
};
use serde::{Deserialize, Serialize};
use tracing_futures::Instrument;

use crate::websocket::AwcWebSocket;

#[derive(Clone)]
pub struct AwcPushService {
    cfg: ServiceConfiguration,
    credentials: Option<HttpAuth>,
    client: awc::Client,
}

impl AwcPushService {
    pub fn new(
        cfg: impl Into<ServiceConfiguration>,
        credentials: Option<ServiceCredentials>,
        user_agent: String,
    ) -> Self {
        let cfg = cfg.into();
        let client = get_client(&cfg, user_agent);
        Self {
            cfg,
            credentials: credentials.and_then(|c| c.authorization()),
            client,
        }
    }

    fn request(
        &self,
        method: Method,
        endpoint: Endpoint,
        path: impl AsRef<str>,
        additional_headers: &[(&str, &str)],
        credentials_override: HttpAuthOverride,
    ) -> Result<ClientRequest, ServiceError> {
        let url = self.cfg.base_url(endpoint).join(path.as_ref())?;
        tracing::debug!(%url, %method, "HTTP request");
        let mut builder = self.client.request(method, url.as_str());
        for &header in additional_headers {
            builder = builder.insert_header(header);
        }
        builder = match credentials_override {
            HttpAuthOverride::NoOverride => {
                if let Some(credentials) = self.credentials.as_ref() {
                    builder.basic_auth(
                        &credentials.username,
                        &credentials.password,
                    )
                } else {
                    builder
                }
            },
            HttpAuthOverride::Identified(HttpAuth { username, password }) => {
                builder.basic_auth(username, password)
            },
            HttpAuthOverride::Unidentified => builder,
        };
        Ok(builder)
    }

    fn json<T>(text: &[u8]) -> Result<T, ServiceError>
    where
        for<'de> T: Deserialize<'de>,
    {
        let value = if text.is_empty() {
            serde_json::from_value(serde_json::Value::Null)
        } else {
            serde_json::from_slice(text)
        };
        value.map_err(|e| ServiceError::JsonDecodeError {
            reason: e.to_string(),
        })
    }

    #[tracing::instrument(name = "extracting error", skip(response))]
    async fn from_response<S>(
        response: &mut ClientResponse<S>,
    ) -> Result<(), ServiceError>
    where
        S: Stream<Item = Result<Bytes, PayloadError>> + Unpin,
    {
        match response.status() {
            StatusCode::OK => Ok(()),
            StatusCode::NO_CONTENT => Ok(()),
            StatusCode::UNAUTHORIZED | StatusCode::FORBIDDEN => {
                Err(ServiceError::Unauthorized)
            },
            StatusCode::NOT_FOUND => {
                // This is 404 and means that e.g. recipient is not registered
                Err(ServiceError::NotFoundError)
            },
            StatusCode::PAYLOAD_TOO_LARGE => {
                // This is 413 and means rate limit exceeded for Signal.
                Err(ServiceError::RateLimitExceeded)
            },
            StatusCode::CONFLICT => {
                let mismatched_devices =
                    response.json().await.map_err(|e| {
                        tracing::error!(
                            ?response,
                            "Failed to decode HTTP 409 response: {}",
                            e
                        );
                        ServiceError::UnhandledResponseCode {
                            http_code: StatusCode::CONFLICT.as_u16(),
                        }
                    })?;
                Err(ServiceError::MismatchedDevicesException(
                    mismatched_devices,
                ))
            },
            StatusCode::GONE => {
                let stale_devices = response.json().await.map_err(|e| {
                    tracing::error!(
                        ?response,
                        "Failed to decode HTTP 410 response: {}",
                        e
                    );
                    ServiceError::UnhandledResponseCode {
                        http_code: StatusCode::GONE.as_u16(),
                    }
                })?;
                Err(ServiceError::StaleDevices(stale_devices))
            },
            StatusCode::LOCKED => {
                let locked = response.json().await.map_err(|e| {
                    tracing::error!(
                        ?response,
                        "Failed to decode HTTP 423 response: {}",
                        e
                    );
                    ServiceError::UnhandledResponseCode {
                        http_code: StatusCode::LOCKED.as_u16(),
                    }
                })?;
                Err(ServiceError::Locked(locked))
            },
            StatusCode::PRECONDITION_REQUIRED => {
                let proof_required = response.json().await.map_err(|e| {
                    tracing::error!(
                        ?response,
                        "Failed to decode HTTP 428 response: {}",
                        e
                    );
                    ServiceError::UnhandledResponseCode {
                        http_code: StatusCode::PRECONDITION_REQUIRED.as_u16(),
                    }
                })?;
                Err(ServiceError::ProofRequiredError(proof_required))
            },
            // XXX: fill in rest from PushServiceSocket
            code => {
                let contents = response.body().await;
                tracing::trace!(
                    ?response,
                    "Unhandled response {} with body: {:?}",
                    code.as_u16(),
                    contents,
                );
                Err(ServiceError::UnhandledResponseCode {
                    http_code: code.as_u16(),
                })
            },
        }
    }
}

#[async_trait::async_trait(?Send)]
impl PushService for AwcPushService {
    // This is in principle known at compile time, but long to write out.
    type ByteStream = Box<dyn futures::io::AsyncRead + Unpin>;

    async fn get_json<T>(
        &mut self,
        endpoint: Endpoint,
        path: &str,
        additional_headers: &[(&str, &str)],
        credentials_override: HttpAuthOverride,
    ) -> Result<T, ServiceError>
    where
        for<'de> T: Deserialize<'de>,
    {
        use awc::error::{ConnectError, SendRequestError};
        let mut response = self
            .request(
                Method::GET,
                endpoint,
                path,
                additional_headers,
                credentials_override,
            )?
            .send()
            .await
            .map_err(|e| match e {
                SendRequestError::Connect(ConnectError::Timeout) => {
                    ServiceError::Timeout {
                        reason: e.to_string(),
                    }
                },
                _ => ServiceError::SendError {
                    reason: e.to_string(),
                },
            })?;

        let _span = tracing::debug_span!("processing response", ?response);

        Self::from_response(&mut response).await?;

        // In order to catch the zero-length output, we have to collect
        // the whole response. The actix-web api is meant to used as a
        // streaming deserializer, so we have this little awkward match.
        //
        // This is also the reason we depend directly on serde_json, however
        // actix already imports that anyway.
        let text = match response.body().await {
            Ok(text) => {
                tracing::debug!(
                    "GET response: {:?}",
                    String::from_utf8_lossy(&text)
                );
                text
            },
            Err(e) => {
                return Err(ServiceError::ResponseError {
                    reason: e.to_string(),
                })
            },
        };
        Self::json(&text)
    }

    /// Deletes a resource through the HTTP DELETE verb.
    async fn delete_json<T>(
        &mut self,
        endpoint: Endpoint,
        path: &str,
        additional_headers: &[(&str, &str)],
    ) -> Result<T, ServiceError>
    where
        for<'de> T: Deserialize<'de>,
    {
        let mut response = self
            .request(
                Method::DELETE,
                endpoint,
                path,
                additional_headers,
                HttpAuthOverride::NoOverride,
            )?
            .send()
            .await
            .map_err(|e| match e {
                SendRequestError::Connect(ConnectError::Timeout) => {
                    ServiceError::Timeout {
                        reason: e.to_string(),
                    }
                },
                _ => ServiceError::SendError {
                    reason: e.to_string(),
                },
            })?;

        let _span =
            tracing::debug_span!("processing response", ?response).entered();

        Self::from_response(&mut response).await?;

        // In order to catch the zero-length output, we have to collect
        // the whole response. The actix-web api is meant to used as a
        // streaming deserializer, so we have this little awkward match.
        //
        // This is also the reason we depend directly on serde_json, however
        // actix already imports that anyway.
        let text = match response.body().await {
            Ok(text) => {
                tracing::debug!(
                    "GET response: {:?}",
                    String::from_utf8_lossy(&text)
                );
                text
            },
            Err(e) => {
                return Err(ServiceError::ResponseError {
                    reason: e.to_string(),
                })
            },
        };

        Self::json(&text)
    }

    async fn put_json<D, S>(
        &mut self,
        endpoint: Endpoint,
        path: &str,
        additional_headers: &[(&str, &str)],
        credentials_override: HttpAuthOverride,
        value: S,
    ) -> Result<D, ServiceError>
    where
        for<'de> D: Deserialize<'de>,
        S: Serialize,
    {
        let mut response = self
            .request(
                Method::PUT,
                endpoint,
                path,
                additional_headers,
                credentials_override,
            )?
            .send_json(&value)
            .await
            .map_err(|e| ServiceError::SendError {
                reason: e.to_string(),
            })?;

        let _span =
            tracing::debug_span!("processing response", ?response).entered();

        Self::from_response(&mut response).await?;

        // In order to catch the zero-length output, we have to collect
        // the whole response. The actix-web api is meant to used as a
        // streaming deserializer, so we have this little awkward match.
        //
        // This is also the reason we depend directly on serde_json, however
        // actix already imports that anyway.
        let text = match response.body().await {
            Ok(text) => {
                tracing::debug!(
                    "GET response: {:?}",
                    String::from_utf8_lossy(&text)
                );
                text
            },
            Err(e) => {
                return Err(ServiceError::ResponseError {
                    reason: e.to_string(),
                })
            },
        };
        Self::json(&text)
    }

    async fn patch_json<D, S>(
        &mut self,
        endpoint: Endpoint,
        path: &str,
        additional_headers: &[(&str, &str)],
        credentials_override: HttpAuthOverride,
        value: S,
    ) -> Result<D, ServiceError>
    where
        for<'de> D: Deserialize<'de>,
        S: Serialize,
    {
        let mut response = self
            .request(
                Method::PATCH,
                endpoint,
                path,
                additional_headers,
                credentials_override,
            )?
            .send_json(&value)
            .await
            .map_err(|e| ServiceError::SendError {
                reason: e.to_string(),
            })?;

        let _span =
            tracing::debug_span!("processing response", ?response).entered();

        Self::from_response(&mut response).await?;

        // In order to catch the zero-length output, we have to collect
        // the whole response. The actix-web api is meant to used as a
        // streaming deserializer, so we have this little awkward match.
        //
        // This is also the reason we depend directly on serde_json, however
        // actix already imports that anyway.
        let text = match response.body().await {
            Ok(text) => {
                tracing::debug!(
                    "PATCH response: {:?}",
                    String::from_utf8_lossy(&text)
                );
                text
            },
            Err(e) => {
                return Err(ServiceError::ResponseError {
                    reason: e.to_string(),
                })
            },
        };
        Self::json(&text)
    }

    async fn post_json<D, S>(
        &mut self,
        endpoint: Endpoint,
        path: &str,
        additional_headers: &[(&str, &str)],
        credentials_override: HttpAuthOverride,
        value: S,
    ) -> Result<D, ServiceError>
    where
        for<'de> D: Deserialize<'de>,
        S: Serialize,
    {
        let mut response = self
            .request(
                Method::POST,
                endpoint,
                path,
                additional_headers,
                credentials_override,
            )?
            .send_json(&value)
            .await
            .map_err(|e| ServiceError::SendError {
                reason: e.to_string(),
            })?;

        let _span =
            tracing::debug_span!("processing response", ?response).entered();

        Self::from_response(&mut response).await?;

        // In order to catch the zero-length output, we have to collect
        // the whole response. The actix-web api is meant to used as a
        // streaming deserializer, so we have this little awkward match.
        //
        // This is also the reason we depend directly on serde_json, however
        // actix already imports that anyway.
        let text = match response.body().await {
            Ok(text) => {
                tracing::debug!(
                    "GET response: {:?}",
                    String::from_utf8_lossy(&text)
                );
                text
            },
            Err(e) => {
                return Err(ServiceError::ResponseError {
                    reason: e.to_string(),
                })
            },
        };
        Self::json(&text)
    }

    async fn get_protobuf<T>(
        &mut self,
        endpoint: Endpoint,
        path: &str,
        additional_headers: &[(&str, &str)],
        credentials_override: HttpAuthOverride,
    ) -> Result<T, ServiceError>
    where
        T: Default + ProtobufMessage,
    {
        let mut response = self
            .request(
                Method::GET,
                endpoint,
                path,
                additional_headers,
                credentials_override,
            )?
            .send()
            .await
            .map_err(|e| ServiceError::SendError {
                reason: e.to_string(),
            })?;

        let text =
            response
                .body()
                .await
                .map_err(|e| ServiceError::ResponseError {
                    reason: e.to_string(),
                })?;
        Ok(T::decode(text)?)
    }

    async fn put_protobuf<D, S>(
        &mut self,
        endpoint: Endpoint,
        path: &str,
        additional_headers: &[(&str, &str)],
        value: S,
    ) -> Result<D, ServiceError>
    where
        D: Default + ProtobufMessage,
        S: Sized + ProtobufMessage,
    {
        let buf = value.encode_to_vec();

        let mut response = self
            .request(
                Method::PUT,
                endpoint,
                path,
                additional_headers,
                HttpAuthOverride::NoOverride,
            )?
            .content_type(HeaderValue::from_static("application/x-protobuf"))
            .send_body(buf)
            .await
            .map_err(|e| ServiceError::SendError {
                reason: e.to_string(),
            })?;

        let text =
            response
                .body()
                .await
                .map_err(|e| ServiceError::ResponseError {
                    reason: e.to_string(),
                })?;
        Ok(D::decode(text)?)
    }

    async fn get_from_cdn(
        &mut self,
        cdn_id: u32,
        path: &str,
    ) -> Result<Self::ByteStream, ServiceError> {
        let mut response = self
            .request(
                Method::GET,
                Endpoint::Cdn(cdn_id),
                path,
                &[],
                HttpAuthOverride::Unidentified,
            )?
            .send()
            .await
            .map_err(|e| ServiceError::SendError {
                reason: e.to_string(),
            })?;

        let _span =
            tracing::debug_span!("processing response", ?response).entered();

        Self::from_response(&mut response).await?;

        Ok(Box::new(
            response
                .map_err(|e| {
                    use awc::error::PayloadError;
                    match e {
                        PayloadError::Io(e) => e,
                        other => std::io::Error::new(
                            std::io::ErrorKind::Other,
                            other,
                        ),
                    }
                })
                .into_async_read(),
        ))
    }

    async fn post_to_cdn0<'s, C: std::io::Read + Send + 's>(
        &mut self,
        path: &str,
        value: &[(&str, &str)],
        file: Option<(&str, &'s mut C)>,
    ) -> Result<(), ServiceError> {
        let request = self.request(
            Method::POST,
            Endpoint::Cdn(0),
            path,
            &[],
            HttpAuthOverride::NoOverride,
        )?;

        let mut form = mpart_async::client::MultipartRequest::default();

        // mpart-async has a peculiar ordering of the form items,
        // and Amazon S3 expects them in a very specific order (i.e., the file contents should
        // go last.
        //
        // mpart-async uses a VecDeque internally for ordering the fields in the order given.
        //
        // https://github.com/cetra3/mpart-async/issues/16

        for &(k, v) in value {
            form.add_field(k, v);
        }

        if let Some((filename, file)) = file {
            // XXX Actix doesn't cope with none-'static lifetimes
            // https://docs.rs/actix-web/3.2.0/actix_web/body/enum.Body.html
            let mut buf = Vec::new();
            file.read_to_end(&mut buf)
                .expect("infallible Read instance");
            form.add_stream(
                "file",
                filename,
                "application/octet-stream",
                futures::future::ok::<_, ()>(Bytes::from(buf)).into_stream(),
            );
        }

        let content_type =
            format!("multipart/form-data; boundary={}", form.get_boundary());

        // XXX Amazon S3 needs the Content-Length, but we don't know it without depleting the whole
        // stream.  Sadly, Content-Length != contents.len(), but should include the whole form.
        let mut body_contents = vec![];
        use futures::stream::StreamExt;
        while let Some(b) = form.next().await {
            // Unwrap, because no error type was used above
            body_contents.extend(b.unwrap());
        }
        tracing::trace!(
            "Sending PUT with Content-Type={} and length {}",
            content_type,
            body_contents.len()
        );

        let mut response = request
            .content_type(&content_type)
            .content_length(body_contents.len() as u64)
            .send_body(body_contents)
            .await
            .map_err(|e| ServiceError::SendError {
                reason: e.to_string(),
            })?;

        let _span =
            tracing::debug_span!("processing response", ?response).entered();

        Self::from_response(&mut response).await?;

        Ok(())
    }

    async fn ws(
        &mut self,
        path: &str,
        keep_alive_path: &str,
        additional_headers: &[(&str, &str)],
        credentials: Option<ServiceCredentials>,
    ) -> Result<SignalWebSocket, ServiceError> {
        let span = tracing::debug_span!("websocket");
        let (ws, stream) = AwcWebSocket::with_client(
            &mut self.client,
            self.cfg.base_url(Endpoint::Service),
            path,
            additional_headers,
            credentials.as_ref(),
        )
        .instrument(span.clone())
        .await?;
        let (ws, task) = SignalWebSocket::from_socket(
            ws,
            stream,
            keep_alive_path.to_owned(),
        );
        actix_rt::spawn(task.instrument(span));
        Ok(ws)
    }
}

/// Creates a `awc::Client` with usable default settings:
/// Creates a default `awc::Client`.
///
/// Creates a `awc::Client` with usable default settings:
/// * certificate authority from the `ServiceConfiguration`
/// * 10s timeout on TCP connection
/// * 65s timeout on HTTP request
/// * provided user-agent
fn get_client(cfg: &ServiceConfiguration, user_agent: String) -> Client {
    let mut cert_bytes = std::io::Cursor::new(&cfg.certificate_authority);
    let roots =
        rustls_pemfile::certs(&mut cert_bytes).expect("parseable PEM files");
    let roots = roots.iter().map(|v| rustls::Certificate(v.clone()));

    let mut root_certs = rustls::RootCertStore::empty();
    for root in roots {
        root_certs.add(&root).unwrap();
    }

    let mut ssl_config = rustls::ClientConfig::builder()
        .with_safe_defaults()
        .with_root_certificates(root_certs)
        .with_no_client_auth();
    ssl_config.alpn_protocols = vec![b"http/1.1".to_vec()];

    let connector = Connector::new()
        .rustls_021(Arc::new(ssl_config))
        .timeout(Duration::from_secs(10)); // https://github.com/actix/actix-web/issues/1047
    let client = awc::ClientBuilder::new()
        .connector(connector)
        .add_default_header(("X-Signal-Agent", user_agent.clone()))
        .add_default_header(("User-Agent", user_agent))
        .timeout(Duration::from_secs(65)); // as in Signal-Android

    client.finish()
}

#[cfg(test)]
mod tests {
    use libsignal_service::configuration::SignalServers;

    #[test]
    fn create_clients() {
        let configs = &[SignalServers::Staging, SignalServers::Production];

        for cfg in configs {
            let _ = super::get_client(
                &cfg.into(),
                "libsignal-service test".to_string(),
            );
        }
    }
}