//! 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)
}