//! Oxygen Strata: repositories kept on cloud object storage (docs/stores.md).

//!

//! This crate gives `oxy` the stores that need an HTTP client and the

//! providers' signing: Amazon S3 (`s3://BUCKET/PREFIX`), Google Cloud

//! Storage (`gs://BUCKET/PREFIX`) and Azure Blob Storage

//! (`az://ACCOUNT/CONTAINER/PREFIX`), each as a [`Store`] over the

//! `object_store` crate. Credentials come from each provider's standard

//! places, with nothing oxy-specific to set up; a URL may carry settings

//! as query parameters (`?profile=work&region=eu-west-2`), which is where

//! `oxy remote add --profile` keeps them.

//!

//! [`register`] hands these stores to oxy-core's remote opener; a program

//! that links this crate calls it once at start.

mod aws;

mod azure;

mod gcp;

mod http;

mod ini;

use futures::TryStreamExt;

use object_store::path::Path as ObjectPath;

use object_store::{

    GetOptions, ObjectStore, ObjectStoreExt, PutMode, PutOptions, PutPayload, UpdateVersion,

};

use oxy_core::strata::{check_key, Entry, Error, Result, Store, Version};

use std::collections::BTreeMap;

use std::future::Future;

use std::ops::Range;

use std::sync::{Arc, Mutex};

/// Make the cloud stores available to `oxy_core::protocol::open_remote`.

/// Safe to call more than once.

pub fn register() {

    // object_store's reqwest is built with rustls and no bundled crypto

    // provider; oxy links ring already, so that is the one to install.

    let _ = rustls::crypto::ring::default_provider().install_default();

    oxy_core::protocol::register_store_opener(open_url);

}

fn open_url(

    url: &str,

    _options: &oxy_core::protocol::RemoteOptions,

) -> Option<Result<Box<dyn Store>>> {

    let scheme = url.split_once("://")?.0;

    if !matches!(scheme, "s3" | "gs" | "az") {

        return None;

    }

    Some(CloudStore::open(url).map(|s| Box::new(s) as Box<dyn Store>))

}

/// A parsed store URL: `SCHEME://HOST/PATH?PARAMS`.

#[derive(Clone, Debug, PartialEq, Eq)]

pub struct StoreUrl {

    pub scheme: String,

    /// The bucket (S3, GCS) or the storage account (Azure).

    pub host: String,

    /// The path under the host, without leading or trailing slashes; for

    /// Azure the container is its first segment.

    pub path: String,

    pub params: BTreeMap<String, String>,

}

impl StoreUrl {

    pub fn parse(url: &str) -> Result<StoreUrl> {

        let bad = |why: &str| Error::Backend(format!("{url}: {why}"));

        let (scheme, rest) = url.split_once("://").ok_or_else(|| bad("not a URL"))?;

        let (rest, query) = match rest.split_once('?') {

            Some((r, q)) => (r, Some(q)),

            None => (rest, None),

        };

        let (host, path) = match rest.split_once('/') {

            Some((h, p)) => (h, p.trim_matches('/')),

            None => (rest, ""),

        };

        if host.is_empty() {

            return Err(bad("names no bucket"));

        }

        let mut params = BTreeMap::new();

        for pair in query.unwrap_or("").split('&').filter(|p| !p.is_empty()) {

            let (k, v) = pair.split_once('=').unwrap_or((pair, ""));

            params.insert(k.to_string(), percent_decode(v));

        }

        Ok(StoreUrl {

            scheme: scheme.to_string(),

            host: host.to_string(),

            path: path.to_string(),

            params,

        })

    }

    /// The URL without its parameters, for messages.

    pub fn plain(&self) -> String {

        if self.path.is_empty() {

            format!("{}://{}", self.scheme, self.host)

        } else {

            format!("{}://{}/{}", self.scheme, self.host, self.path)

        }

    }

    pub fn param(&self, name: &str) -> Option<&str> {

        self.params.get(name).map(String::as_str)

    }

}

fn percent_decode(s: &str) -> String {

    let bytes = s.as_bytes();

    let mut out = Vec::with_capacity(bytes.len());

    let mut i = 0;

    while i < bytes.len() {

        if bytes[i] == b'%' && i + 2 < bytes.len() {

            if let Ok(v) = u8::from_str_radix(&s[i + 1..i + 3], 16) {

                out.push(v);

                i += 3;

                continue;

            }

        }

        out.push(bytes[i]);

        i += 1;

    }

    String::from_utf8_lossy(&out).into_owned()

}

/// A store on a cloud provider: the provider's client, the prefix every

/// key goes under, and a runtime to run the client on.

pub struct CloudStore {

    inner: Arc<dyn ObjectStore>,

    prefix: String,

    name: String,

    hint: &'static str,

    runtime: tokio::runtime::Runtime,

    /// One call at a time: the runtime runs on the calling thread.

    turn: Mutex<()>,

}

impl CloudStore {

    /// Open the store a URL names. Nothing is sent yet: credentials are

    /// resolved here, but the bucket is first spoken to on the first call.

    pub fn open(url: &str) -> Result<CloudStore> {

        let parsed = StoreUrl::parse(url)?;

        let runtime = tokio::runtime::Builder::new_current_thread()

            .enable_all()

            .build()

            .map_err(|e| Error::Backend(format!("starting the HTTP runtime: {e}")))?;

        let (inner, prefix, hint): (Arc<dyn ObjectStore>, String, &'static str) =

            match parsed.scheme.as_str() {

                "s3" => (

                    Arc::new(aws::open(&parsed)?),

                    parsed.path.clone(),

                    aws::HINT,

                ),

                "gs" => (

                    Arc::new(gcp::open(&parsed)?),

                    parsed.path.clone(),

                    gcp::HINT,

                ),

                "az" => {

                    let (store, prefix) = azure::open(&parsed)?;

                    (Arc::new(store), prefix, azure::HINT)

                }

                other => {

                    return Err(Error::Backend(format!(

                        "{url}: {other}:// is not a store oxy knows (s3://, gs://, az://, dir://)"

                    )))

                }

            };

        Ok(CloudStore {

            inner,

            prefix,

            name: parsed.plain(),

            hint,

            runtime,

            turn: Mutex::new(()),

        })

    }

    fn block<F: Future>(&self, f: F) -> F::Output {

        let _turn = self.turn.lock().unwrap_or_else(|p| p.into_inner());

        self.runtime.block_on(f)

    }

    fn path(&self, key: &str) -> Result<ObjectPath> {

        check_key(key)?;

        let full = if self.prefix.is_empty() {

            key.to_string()

        } else {

            format!("{}/{key}", self.prefix)

        };

        ObjectPath::parse(&full).map_err(|e| Error::Backend(format!("{full}: {e}")))

    }

    /// Our error for the provider's, with what to do next where it helps.

    fn map(&self, e: object_store::Error, key: &str) -> Error {

        use object_store::Error as E;

        match e {

            E::NotFound { .. } => Error::NotFound(key.to_string()),

            E::AlreadyExists { .. } => Error::Exists(key.to_string()),

            E::Precondition { .. } | E::NotModified { .. } => Error::Conflict(key.to_string()),

            other => {

                let text = other.to_string();

                let lower = text.to_ascii_lowercase();

                let advice = if lower.contains("403")

                    || lower.contains("accessdenied")

                    || lower.contains("forbidden")

                    || lower.contains("authorizationpermissionmismatch")

                {

                    format!(

                        ": access denied; the credentials in use may not do this on {}: {}",

                        self.name, self.hint

                    )

                } else if lower.contains("credential") || lower.contains("no valid credentials") {

                    format!(": no usable credentials: {}", self.hint)

                } else if lower.contains("nosuchbucket")

                    || lower.contains("bucketnotfound")

                    || lower.contains("containernotfound")

                {

                    format!(

                        ": {} does not exist, or the credentials cannot see it",

                        self.name

                    )

                } else if lower.contains("dns")

                    || lower.contains("connect")

                    || lower.contains("timed out")

                {

                    ": cannot reach the service; check the network, the endpoint and the region"

                        .to_string()

                } else {

                    String::new()

                };

                Error::Backend(format!("{}: {text}{advice}", self.name))

            }

        }

    }

    fn version_of(e_tag: Option<String>, version: Option<String>) -> Version {

        Version(format!(

            "{}\t{}",

            e_tag.unwrap_or_default(),

            version.unwrap_or_default()

        ))

    }

    fn update_version(v: &Version) -> UpdateVersion {

        let (e_tag, version) = v.0.split_once('\t').unwrap_or((v.0.as_str(), ""));

        UpdateVersion {

            e_tag: (!e_tag.is_empty()).then(|| e_tag.to_string()),

            version: (!version.is_empty()).then(|| version.to_string()),

        }

    }

}

impl Store for CloudStore {

    fn describe(&self) -> String {

        self.name.clone()

    }

    fn has(&self, key: &str) -> Result<bool> {

        let path = self.path(key)?;

        match self.block(self.inner.head(&path)) {

            Ok(_) => Ok(true),

            Err(object_store::Error::NotFound { .. }) => Ok(false),

            Err(e) => Err(self.map(e, key)),

        }

    }

    fn get(&self, key: &str) -> Result<Vec<u8>> {

        let path = self.path(key)?;

        self.block(async {

            let r = self.inner.get(&path).await?;

            r.bytes().await

        })

        .map(|b| b.to_vec())

        .map_err(|e| self.map(e, key))

    }

    fn get_range(&self, key: &str, range: Range<u64>) -> Result<Vec<u8>> {

        let path = self.path(key)?;

        if range.start > range.end {

            return Err(Error::Backend(format!(

                "{key}: bytes {}..{} is not a range",

                range.start, range.end

            )));

        }

        if range.is_empty() {

            return Ok(Vec::new());

        }

        let want = (range.end - range.start) as usize;

        let bytes = self

            .block(self.inner.get_range(&path, range.clone()))

            .map_err(|e| self.map(e, key))?;

        if bytes.len() != want {

            return Err(Error::Backend(format!(

                "{key}: bytes {}..{} asked, {} returned: the object is shorter",

                range.start,

                range.end,

                bytes.len()

            )));

        }

        Ok(bytes.to_vec())

    }

    fn put_new(&self, key: &str, bytes: &[u8]) -> Result<()> {

        let path = self.path(key)?;

        let opts = PutOptions {

            mode: PutMode::Create,

            ..Default::default()

        };

        self.block(

            self.inner

                .put_opts(&path, PutPayload::from(bytes.to_vec()), opts),

        )

        .map(|_| ())

        .map_err(|e| self.map(e, key))

    }

    fn list(&self, prefix: &str) -> Result<Vec<Entry>> {

        // Providers list by whole path segments; the asked prefix may end

        // mid-segment, so list its directory and filter here.

        let dir = prefix.rsplit_once('/').map(|(d, _)| d).unwrap_or("");

        let full_dir = match (self.prefix.as_str(), dir) {

            ("", "") => None,

            ("", d) => Some(d.to_string()),

            (p, "") => Some(p.to_string()),

            (p, d) => Some(format!("{p}/{d}")),

        };

        let full_dir = full_dir

            .map(|d| ObjectPath::parse(&d).map_err(|e| Error::Backend(format!("{d}: {e}"))))

            .transpose()?;

        let metas: Vec<object_store::ObjectMeta> = self

            .block(self.inner.list(full_dir.as_ref()).try_collect())

            .map_err(|e| self.map(e, prefix))?;

        let strip = if self.prefix.is_empty() {

            String::new()

        } else {

            format!("{}/", self.prefix)

        };

        let mut out: Vec<Entry> = metas

            .into_iter()

            .filter_map(|m| {

                let loc = m.location.to_string();

                let key = loc.strip_prefix(&strip)?;

                key.starts_with(prefix).then(|| Entry {

                    key: key.to_string(),

                    size: m.size,

                    modified_ms: u64::try_from(m.last_modified.timestamp_millis()).ok(),

                })

            })

            .collect();

        out.sort_by(|a, b| a.key.cmp(&b.key));

        Ok(out)

    }

    fn delete(&self, key: &str) -> Result<()> {

        let path = self.path(key)?;

        match self.block(self.inner.delete(&path)) {

            Ok(()) | Err(object_store::Error::NotFound { .. }) => Ok(()),

            Err(e) => Err(self.map(e, key)),

        }

    }

    fn read_ref(&self, key: &str) -> Result<Option<(Vec<u8>, Version)>> {

        let path = self.path(key)?;

        let got = self.block(async {

            let r = self.inner.get_opts(&path, GetOptions::default()).await?;

            let meta = r.meta.clone();

            let bytes = r.bytes().await?;

            Ok::<_, object_store::Error>((bytes, meta))

        });

        match got {

            Ok((bytes, meta)) => Ok(Some((

                bytes.to_vec(),

                Self::version_of(meta.e_tag, meta.version),

            ))),

            Err(object_store::Error::NotFound { .. }) => Ok(None),

            Err(e) => Err(self.map(e, key)),

        }

    }

    fn write_ref(&self, key: &str, bytes: &[u8], expected: Option<&Version>) -> Result<Version> {

        let path = self.path(key)?;

        let mode = match expected {

            None => PutMode::Create,

            Some(v) => PutMode::Update(Self::update_version(v)),

        };

        let opts = PutOptions {

            mode,

            ..Default::default()

        };

        let put = self

            .block(

                self.inner

                    .put_opts(&path, PutPayload::from(bytes.to_vec()), opts),

            )

            .map_err(|e| match self.map(e, key) {

                // A record that exists when none was expected is a lost

                // race too, not a misuse of put_new.

                Error::Exists(k) => Error::Conflict(k),

                other => other,

            })?;

        Ok(Self::version_of(put.e_tag, put.version))

    }

}

#[cfg(test)]

mod tests {

    use super::*;

    #[test]

    fn urls_parse_with_their_parameters() {

        let u =

            StoreUrl::parse("s3://my-bucket/repos/oxygen?profile=work&region=eu-west-2").unwrap();

        assert_eq!(u.scheme, "s3");

        assert_eq!(u.host, "my-bucket");

        assert_eq!(u.path, "repos/oxygen");

        assert_eq!(u.param("profile"), Some("work"));

        assert_eq!(u.param("region"), Some("eu-west-2"));

        assert_eq!(u.plain(), "s3://my-bucket/repos/oxygen");

        let u = StoreUrl::parse("gs://bucket").unwrap();

        assert_eq!((u.host.as_str(), u.path.as_str()), ("bucket", ""));

        assert_eq!(u.plain(), "gs://bucket");

        let u = StoreUrl::parse("az://acct/container/a/b/?endpoint=http%3A%2F%2F127.0.0.1%3A10000")

            .unwrap();

        assert_eq!(u.path, "container/a/b");

        assert_eq!(u.param("endpoint"), Some("http://127.0.0.1:10000"));

        assert!(StoreUrl::parse("s3://").is_err());

        assert!(StoreUrl::parse("nonsense").is_err());

    }

    #[test]

    fn versions_carry_both_an_etag_and_a_generation() {

        let v = CloudStore::version_of(Some("\"abc\"".into()), Some("17".into()));

        let u = CloudStore::update_version(&v);

        assert_eq!(u.e_tag.as_deref(), Some("\"abc\""));

        assert_eq!(u.version.as_deref(), Some("17"));

        let u = CloudStore::update_version(&CloudStore::version_of(None, Some("3".into())));

        assert_eq!((u.e_tag, u.version.as_deref()), (None, Some("3")));

    }

}