//! The HTTP client the cloud stores send their requests through: `ureq`

//! on rustls with the web roots, the client oxy-core itself uses, so the

//! stores cross-compile for every target oxy is released for. Each

//! request runs on the runtime's blocking pool; bodies are held whole,

//! which is how the stores use them (a bundle, a record, a listing).

use async_trait::async_trait;

use bytes::Bytes;

use http_body_util::{BodyExt, Full};

use object_store::client::{

    ClientOptions, HttpClient, HttpConnector, HttpError, HttpErrorKind, HttpRequest, HttpResponse,

    HttpResponseBody, HttpService,

};

use std::io::Read;

use std::time::Duration;

/// How long a request may take in all, and to connect.

const TIMEOUT: Duration = Duration::from_secs(600);

const CONNECT_TIMEOUT: Duration = Duration::from_secs(30);

/// A rustls configuration trusting the web roots.

fn tls() -> Result<std::sync::Arc<rustls::ClientConfig>, String> {

    oxy_core::protocol::tls_client_config(None).map_err(|e| e.to_string())

}

/// Make the agent every request of a store goes through.

pub fn agent() -> Result<ureq::Agent, String> {

    Ok(ureq::AgentBuilder::new()

        .tls_config(tls()?)

        .timeout_connect(CONNECT_TIMEOUT)

        .timeout(TIMEOUT)

        .build())

}

#[derive(Debug, Clone, Copy, Default)]

pub struct UreqConnector;

impl HttpConnector for UreqConnector {

    fn connect(&self, _options: &ClientOptions) -> object_store::Result<HttpClient> {

        let agent = agent().map_err(|e| object_store::Error::Generic {

            store: "ureq",

            source: e.into(),

        })?;

        Ok(HttpClient::new(UreqService { agent }))

    }

}

#[derive(Debug, Clone)]

struct UreqService {

    agent: ureq::Agent,

}

#[async_trait]

impl HttpService for UreqService {

    async fn call(&self, req: HttpRequest) -> Result<HttpResponse, HttpError> {

        let (parts, body) = req.into_parts();

        let bytes = body.collect().await?.to_bytes();

        let agent = self.agent.clone();

        tokio::task::spawn_blocking(move || send(&agent, parts, &bytes))

            .await

            .map_err(|e| HttpError::new(HttpErrorKind::Interrupted, e))?

    }

}

fn send(

    agent: &ureq::Agent,

    parts: http::request::Parts,

    bytes: &[u8],

) -> Result<HttpResponse, HttpError> {

    let url = parts.uri.to_string();

    let mut r = agent.request(parts.method.as_str(), &url);

    for (name, value) in &parts.headers {

        if let Ok(v) = value.to_str() {

            r = r.set(name.as_str(), v);

        }

    }

    let sent = if bytes.is_empty()

        && matches!(

            parts.method,

            http::Method::GET | http::Method::HEAD | http::Method::DELETE

        ) {

        r.call()

    } else {

        r.send_bytes(bytes)

    };

    let resp = match sent {

        Ok(resp) => resp,

        // A refusal is a response the store interprets (a 404, a 412).

        Err(ureq::Error::Status(_, resp)) => resp,

        Err(ureq::Error::Transport(t)) => {

            let kind = match t.kind() {

                ureq::ErrorKind::Dns

                | ureq::ErrorKind::ConnectionFailed

                | ureq::ErrorKind::ProxyConnect => HttpErrorKind::Connect,

                ureq::ErrorKind::Io => HttpErrorKind::Timeout,

                _ => HttpErrorKind::Request,

            };

            return Err(HttpError::new(kind, t));

        }

    };

    let status = http::StatusCode::from_u16(resp.status())

        .map_err(|e| HttpError::new(HttpErrorKind::Decode, e))?;

    let mut builder = http::Response::builder().status(status);

    for name in resp.headers_names() {

        for value in resp.all(&name) {

            builder = builder.header(name.as_str(), value);

        }

    }

    let mut body = Vec::new();

    resp.into_reader()

        .read_to_end(&mut body)

        .map_err(|e| HttpError::new(HttpErrorKind::Decode, e))?;

    let body = HttpResponseBody::new(

        Full::new(Bytes::from(body)).map_err(|never: std::convert::Infallible| match never {}),

    );

    builder

        .body(body)

        .map_err(|e| HttpError::new(HttpErrorKind::Decode, e))

}

/// The region that holds an S3 bucket, as S3 itself reports it in

/// `x-amz-bucket-region` on any request, even a refused one.

pub fn bucket_region(bucket: &str) -> Option<String> {

    let agent = ureq::AgentBuilder::new()

        .tls_config(tls().ok()?)

        .timeout(Duration::from_secs(30))

        .redirects(0)

        .build();

    // S3's wildcard certificate covers one label: dotted names go by path.

    let url = if bucket.contains('.') {

        format!("https://s3.amazonaws.com/{bucket}")

    } else {

        format!("https://{bucket}.s3.amazonaws.com")

    };

    let resp = match agent.head(&url).call() {

        Ok(r) | Err(ureq::Error::Status(_, r)) => r,

        Err(_) => return None,

    };

    resp.header("x-amz-bucket-region").map(str::to_string)

}