/*
 *     Copyright 2024 The Dragonfly Authors
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

use dragonfly_client_core::{
    error::{ErrorType, OrErr},
    Error, Result,
};
use futures::TryStreamExt;
use rustls_pki_types::CertificateDer;
use std::io::{Error as IOError, ErrorKind};
use tokio_util::io::StreamReader;
use tracing::error;

// HTTP is the HTTP backend.
pub struct HTTP;

// HTTP implements the http interface.
impl HTTP {
    // new returns a new HTTP.
    pub fn new() -> HTTP {
        Self {}
    }

    // client returns a new reqwest client.
    fn client(
        &self,
        client_certs: Option<Vec<CertificateDer<'static>>>,
    ) -> Result<reqwest::Client> {
        match client_certs {
            Some(client_certs) => {
                // TLS client config using the custom CA store for lookups.
                let mut root_cert_store = rustls::RootCertStore::empty();
                root_cert_store.add_parsable_certificates(&client_certs);
                let client_config = rustls::ClientConfig::builder()
                    .with_safe_defaults()
                    .with_root_certificates(root_cert_store)
                    .with_no_client_auth();

                let client = reqwest::Client::builder()
                    .use_preconfigured_tls(client_config)
                    .build()
                    .or_err(ErrorType::HTTPError)?;
                Ok(client)
            }
            None => {
                // Default TLS client config with native roots.
                let client = reqwest::Client::builder()
                    .use_native_tls()
                    .build()
                    .or_err(ErrorType::HTTPError)?;
                Ok(client)
            }
        }
    }
}

// Backend implements the Backend trait.
#[tonic::async_trait]
impl crate::Backend for HTTP {
    // head gets the header of the request.
    async fn head(&self, request: crate::HeadRequest) -> Result<crate::HeadResponse> {
        // The header of the request is required.
        let header = request.http_header.ok_or(Error::InvalidParameter)?;

        // The signature in the signed URL generated by the object storage client will include
        // the request method. Therefore, the signed URL of the GET method cannot be requested
        // through the HEAD method. Use GET request to replace of HEAD request
        // to get header and status code.
        let response = self
            .client(request.client_certs)?
            .get(&request.url)
            .headers(header)
            .timeout(request.timeout)
            .send()
            .await
            .or_err(ErrorType::HTTPError)
            .map_err(|err| {
                error!("head request failed: {}", err);
                err
            })?;

        let header = response.headers().clone();
        let status_code = response.status();

        Ok(crate::HeadResponse {
            http_header: Some(header),
            http_status_code: Some(status_code),
        })
    }

    // get gets the content of the request.
    async fn get(&self, request: crate::GetRequest) -> Result<crate::GetResponse<crate::Body>> {
        // The header of the request is required.
        let header = request.http_header.ok_or(Error::InvalidParameter)?;
        let response = self
            .client(request.client_certs)?
            .get(&request.url)
            .headers(header)
            .timeout(request.timeout)
            .send()
            .await
            .or_err(ErrorType::HTTPError)
            .map_err(|err| {
                error!("get request failed: {}", err);
                err
            })?;

        let header = response.headers().clone();
        let status_code = response.status();
        let reader = Box::new(StreamReader::new(
            response
                .bytes_stream()
                .map_err(|err| IOError::new(ErrorKind::Other, err)),
        ));
        Ok(crate::GetResponse {
            http_header: Some(header),
            http_status_code: Some(status_code),
            reader,
        })
    }
}

// Default implements the Default trait.
impl Default for HTTP {
    // default returns a new default HTTP.
    fn default() -> Self {
        Self::new()
    }
}

#[cfg(test)]
mod tests {
    use crate::{Backend, GetRequest, HeadRequest};
    use httpmock::{Method, MockServer};
    use reqwest::{header::HeaderMap, StatusCode};

    use super::*;

    #[tokio::test]
    async fn should_get_head_response() {
        let server = MockServer::start();
        server.mock(|when, then| {
            when.method(Method::GET).path("/head");
            then.status(200)
                .header("content-type", "text/html; charset=UTF-8")
                .body("");
        });

        let http_backend = HTTP::new();
        let resp = http_backend
            .head(HeadRequest {
                url: server.url("/head"),
                http_header: Some(HeaderMap::new()),
                timeout: std::time::Duration::from_secs(5),
                client_certs: None,
            })
            .await
            .unwrap();

        assert_eq!(resp.http_status_code, Some(StatusCode::OK))
    }

    #[tokio::test]
    async fn should_return_error_response_when_head_notexists() {
        let server = MockServer::start();
        server.mock(|when, then| {
            when.method(Method::GET).path("/head");
            then.status(200)
                .header("content-type", "text/html; charset=UTF-8")
                .body("");
        });

        let http_backend = HTTP::new();
        let resp = http_backend
            .head(HeadRequest {
                url: server.url("/head"),
                http_header: None,
                timeout: std::time::Duration::from_secs(5),
                client_certs: None,
            })
            .await;

        assert!(resp.is_err());
    }

    #[tokio::test]
    async fn should_get_response() {
        let server = MockServer::start();
        server.mock(|when, then| {
            when.method(Method::GET).path("/get");
            then.status(200)
                .header("content-type", "text/html; charset=UTF-8")
                .body("OK");
        });

        let http_backend = HTTP::new();
        let mut resp = http_backend
            .get(GetRequest {
                url: server.url("/get"),
                http_header: Some(HeaderMap::new()),
                timeout: std::time::Duration::from_secs(5),
                client_certs: None,
            })
            .await
            .unwrap();

        assert_eq!(resp.http_status_code, Some(StatusCode::OK));
        assert_eq!(resp.text().await.unwrap(), "OK");
    }
}