//! 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;
    }
    let lookup = *CREDENTIALS.lock().unwrap_or_else(|p| p.into_inner());
    let given = lookup.and_then(|f| f(url));
    Some(CloudStore::open_with(url, given.as_ref()).map(|s| Box::new(s) as Box<dyn Store>))
}

/// Credentials a program keeps itself and hands over, instead of the
/// provider's usual places: a hub keeps each repository's sealed. When a
/// store is opened with them, nothing from the environment or the
/// provider's files is used.
#[derive(Clone, PartialEq, Eq)]
pub enum Credentials {
    /// An S3 access key.
    Aws {
        access_key_id: String,
        secret_access_key: String,
        session_token: Option<String>,
    },
    /// A Google Cloud service account key, as the JSON Google issues.
    Gcp { service_account_key: String },
    /// An Azure service principal.
    Azure {
        tenant_id: String,
        client_id: String,
        client_secret: String,
    },
}

impl std::fmt::Debug for Credentials {
    /// Names the kind and the public part only: never the secret.
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Credentials::Aws { access_key_id, .. } => {
                write!(f, "Aws({access_key_id}, secret hidden)")
            }
            Credentials::Gcp { .. } => write!(f, "Gcp(service account key hidden)"),
            Credentials::Azure { client_id, .. } => write!(f, "Azure({client_id}, secret hidden)"),
        }
    }
}

impl Credentials {
    /// The scheme of the stores these are for.
    pub fn scheme(&self) -> &'static str {
        match self {
            Credentials::Aws { .. } => "s3",
            Credentials::Gcp { .. } => "gs",
            Credentials::Azure { .. } => "az",
        }
    }

    /// As JSON, for keeping (sealed) and reading back with [`Credentials::from_json`].
    pub fn to_json(&self) -> String {
        let v = match self {
            Credentials::Aws {
                access_key_id,
                secret_access_key,
                session_token,
            } => serde_json::json!({
                "kind": "aws", "access_key_id": access_key_id, "secret_access_key": secret_access_key,
                "session_token": session_token,
            }),
            Credentials::Gcp {
                service_account_key,
            } => {
                serde_json::json!({ "kind": "gcp", "service_account_key": service_account_key })
            }
            Credentials::Azure {
                tenant_id,
                client_id,
                client_secret,
            } => serde_json::json!({
                "kind": "azure", "tenant_id": tenant_id, "client_id": client_id, "client_secret": client_secret,
            }),
        };
        v.to_string()
    }

    pub fn from_json(text: &str) -> Result<Credentials> {
        let v: serde_json::Value = serde_json::from_str(text)
            .map_err(|e| Error::Backend(format!("credentials are not JSON: {e}")))?;
        let field = |k: &str| -> Result<String> {
            v.get(k)
                .and_then(|x| x.as_str())
                .filter(|x| !x.trim().is_empty())
                .map(|x| x.trim().to_string())
                .ok_or_else(|| Error::Backend(format!("credentials need {k}")))
        };
        match v.get("kind").and_then(|k| k.as_str()) {
            Some("aws") => Ok(Credentials::Aws {
                access_key_id: field("access_key_id")?,
                secret_access_key: field("secret_access_key")?,
                session_token: field("session_token").ok(),
            }),
            Some("gcp") => {
                let key = field("service_account_key")?;
                let parsed: serde_json::Value = serde_json::from_str(&key).map_err(|_| {
                    Error::Backend("the service account key is not the JSON Google issues".into())
                })?;
                if parsed.get("type").and_then(|t| t.as_str()) != Some("service_account") {
                    return Err(Error::Backend(
                        "the key is not a service account key (its type is not service_account)"
                            .into(),
                    ));
                }
                Ok(Credentials::Gcp {
                    service_account_key: key,
                })
            }
            Some("azure") => Ok(Credentials::Azure {
                tenant_id: field("tenant_id")?,
                client_id: field("client_id")?,
                client_secret: field("client_secret")?,
            }),
            _ => Err(Error::Backend(
                "credentials need a kind: aws, gcp or azure".into(),
            )),
        }
    }
}

/// Finds the credentials a program keeps for a store URL.
pub type CredentialsLookup = fn(&str) -> Option<Credentials>;

static CREDENTIALS: Mutex<Option<CredentialsLookup>> = Mutex::new(None);

/// Have stores opened by URL (as a repository's `storage.url` is) use the
/// credentials `lookup` finds for it, where it finds any.
pub fn set_credentials_lookup(lookup: CredentialsLookup) {
    *CREDENTIALS.lock().unwrap_or_else(|p| p.into_inner()) = Some(lookup);
}

/// 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,
    /// Always there until the store is dropped.
    runtime: Option<tokio::runtime::Runtime>,
    /// One call at a time: the runtime runs on the calling thread.
    turn: Mutex<()>,
}

impl Drop for CloudStore {
    /// Without waiting, so a store can be dropped inside a runtime too.
    fn drop(&mut self) {
        if let Some(rt) = self.runtime.take() {
            rt.shutdown_background();
        }
    }
}

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> {
        Self::open_with(url, None)
    }

    /// [`CloudStore::open`] with credentials given, which must be for
    /// the URL's kind of store; then only they are used.
    pub fn open_with(url: &str, given: Option<&Credentials>) -> Result<CloudStore> {
        let parsed = StoreUrl::parse(url)?;
        if let Some(c) = given.filter(|c| c.scheme() != parsed.scheme) {
            return Err(Error::Backend(format!(
                "{}: the credentials given are for {}:// stores",
                parsed.plain(),
                c.scheme()
            )));
        }
        let aws_keys = match given {
            Some(Credentials::Aws {
                access_key_id,
                secret_access_key,
                session_token,
            }) => Some((
                access_key_id.as_str(),
                secret_access_key.as_str(),
                session_token.as_deref(),
            )),
            _ => None,
        };
        let gcp_key = match given {
            Some(Credentials::Gcp {
                service_account_key,
            }) => Some(service_account_key.as_str()),
            _ => None,
        };
        let principal = match given {
            Some(Credentials::Azure {
                tenant_id,
                client_id,
                client_secret,
            }) => Some((
                tenant_id.as_str(),
                client_id.as_str(),
                client_secret.as_str(),
            )),
            _ => None,
        };
        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, aws_keys)?),
                    parsed.path.clone(),
                    aws::HINT,
                ),
                "gs" => (
                    Arc::new(gcp::open(&parsed, gcp_key)?),
                    parsed.path.clone(),
                    gcp::HINT,
                ),
                "az" => {
                    let (store, prefix) = azure::open(&parsed, principal)?;
                    (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: Some(runtime),
            turn: Mutex::new(()),
        })
    }

    /// Run a request to completion. Called from a thread that is itself
    /// driving async tasks (a server's handler), where blocking on another
    /// runtime is not allowed, the request runs on a thread of its own.
    fn block<F>(&self, f: F) -> F::Output
    where
        F: Future + Send,
        F::Output: Send,
    {
        let _turn = self.turn.lock().unwrap_or_else(|p| p.into_inner());
        let runtime = self
            .runtime
            .as_ref()
            .expect("the runtime lives as long as the store");
        if tokio::runtime::Handle::try_current().is_ok() {
            return std::thread::scope(|s| {
                s.spawn(|| runtime.block_on(f))
                    .join()
                    .unwrap_or_else(|p| std::panic::resume_unwind(p))
            });
        }
        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 {
    #[tokio::test]
    async fn a_store_can_be_used_from_inside_a_runtime() {
        // As a server's handler would: blocking on the store's own runtime
        // from here must neither panic nor deadlock, and nor must dropping it.
        let store = CloudStore::open(
            "s3://oxy-no-such-bucket/p?region=eu-west-2&endpoint=http://127.0.0.1:9",
        )
        .unwrap();
        assert_eq!(store.block(async { 7 }), 7);
        drop(store);
    }

    #[test]
    fn credentials_round_trip_and_never_show_their_secret() {
        let aws = Credentials::Aws {
            access_key_id: "AKIDEXAMPLE".into(),
            secret_access_key: "wJalrXUtnFEMI".into(),
            session_token: None,
        };
        assert_eq!(Credentials::from_json(&aws.to_json()).unwrap(), aws);
        assert_eq!(format!("{aws:?}"), "Aws(AKIDEXAMPLE, secret hidden)");
        let az = Credentials::Azure {
            tenant_id: "t".into(),
            client_id: "c".into(),
            client_secret: "s3cr3t".into(),
        };
        assert_eq!(Credentials::from_json(&az.to_json()).unwrap(), az);
        assert!(!format!("{az:?}").contains("s3cr3t"));
        let key = r#"{"type":"service_account","client_email":"x@y","private_key":"k"}"#;
        let gcp = Credentials::Gcp {
            service_account_key: key.into(),
        };
        assert_eq!(Credentials::from_json(&gcp.to_json()).unwrap(), gcp);
        assert!(!format!("{gcp:?}").contains("private_key"));
        for (bad, why) in [
            (r#"{"kind":"aws","access_key_id":"a"}"#, "secret_access_key"),
            (
                r#"{"kind":"gcp","service_account_key":"{\"type\":\"authorized_user\"}"}"#,
                "not a service account key",
            ),
            (r#"{"kind":"ftp"}"#, "aws, gcp or azure"),
            ("nope", "not JSON"),
        ] {
            let e = Credentials::from_json(bad).unwrap_err().to_string();
            assert!(e.contains(why), "{bad}: {e}");
        }
        let e = CloudStore::open_with("gs://bucket/p", Some(&aws))
            .err()
            .unwrap()
            .to_string();
        assert!(e.contains("are for s3:// stores"), "{e}");
    }

    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")));
    }
}