//! Amazon S3: `s3://BUCKET/PREFIX`.

//!

//! Credentials, in the order the AWS tools use them:

//! 1. `AWS_ACCESS_KEY_ID` and `AWS_SECRET_ACCESS_KEY` (and

//!    `AWS_SESSION_TOKEN`) in the environment, unless the URL names a

//!    profile;

//! 2. the profile (`?profile=NAME`, else `AWS_PROFILE`, else `default`):

//!    its keys from `~/.aws/credentials`;

//! 3. what the AWS CLI can produce for that profile

//!    (`aws configure export-credentials`): SSO, login sessions, assumed

//!    roles and credential processes, so anything that works for `aws`

//!    works for `oxy`;

//! 4. the instance's or container's role, through the provider's own

//!    lookup.

//!

//! The region comes from `?region=`, the environment, the profile, or is

//! asked of S3 itself; the endpoint (`?endpoint=`, for MinIO and other

//! S3-alikes) from the URL, the environment or the profile.

use crate::ini;

use crate::StoreUrl;

use object_store::aws::{AmazonS3, AmazonS3Builder, S3ConditionalPut};

use object_store::ClientOptions;

use oxy_core::strata::{Error, Result};

use std::path::PathBuf;

pub const HINT: &str = "set AWS_PROFILE (or add ?profile=NAME to the remote URL) to a profile in ~/.aws/credentials, run `aws sso login` or `aws login` for it, or set AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY";

struct Found {

    key_id: String,

    secret: String,

    token: Option<String>,

}

fn env(name: &str) -> Option<String> {

    std::env::var(name).ok().filter(|v| !v.is_empty())

}

fn home() -> Option<PathBuf> {

    env("HOME")

        .or_else(|| env("USERPROFILE"))

        .map(PathBuf::from)

}

fn read_ini(env_override: &str, default_rel: &str) -> ini::Sections {

    let path = env(env_override)

        .map(PathBuf::from)

        .or_else(|| home().map(|h| h.join(default_rel)));

    path.and_then(|p| std::fs::read_to_string(p).ok())

        .map(|t| ini::parse(&t))

        .unwrap_or_default()

}

/// A profile's section in `~/.aws/config`: `[profile NAME]`, or

/// `[default]`; `~/.aws/credentials` uses `[NAME]` for both.

fn config_section<'a>(

    config: &'a ini::Sections,

    profile: &str,

) -> Option<&'a std::collections::BTreeMap<String, String>> {

    if profile == "default" {

        config.get("default")

    } else {

        config

            .get(&format!("profile {profile}"))

            .or_else(|| config.get(profile))

    }

}

/// Ask the AWS CLI for the profile's credentials, if it is installed and

/// knows the profile.

fn from_cli(profile: &str) -> Option<Found> {

    let out = std::process::Command::new("aws")

        .args([

            "configure",

            "export-credentials",

            "--profile",

            profile,

            "--format",

            "process",

        ])

        .stdin(std::process::Stdio::null())

        .stderr(std::process::Stdio::null())

        .output()

        .ok()?;

    if !out.status.success() {

        return None;

    }

    let v: serde_json::Value = serde_json::from_slice(&out.stdout).ok()?;

    Some(Found {

        key_id: v.get("AccessKeyId")?.as_str()?.to_string(),

        secret: v.get("SecretAccessKey")?.as_str()?.to_string(),

        token: v

            .get("SessionToken")

            .and_then(|t| t.as_str())

            .map(str::to_string),

    })

}

pub fn open(url: &StoreUrl, runtime: &tokio::runtime::Runtime) -> Result<AmazonS3> {

    let bucket = url.host.clone();

    let explicit_profile = url.param("profile").map(str::to_string);

    let profile = explicit_profile

        .clone()

        .or_else(|| env("AWS_PROFILE"))

        .unwrap_or_else(|| "default".into());

    let config = read_ini("AWS_CONFIG_FILE", ".aws/config");

    let creds = read_ini("AWS_SHARED_CREDENTIALS_FILE", ".aws/credentials");

    let section = config_section(&config, &profile);

    let known_profile = section.is_some() || creds.contains_key(&profile);

    // from_env picks up the environment's keys, region, endpoint, and the

    // container and instance roles; what follows fills in or overrides.

    let mut b = AmazonS3Builder::from_env()

        .with_bucket_name(&bucket)

        .with_conditional_put(S3ConditionalPut::ETagMatch);

    let env_keys = match (env("AWS_ACCESS_KEY_ID"), env("AWS_SECRET_ACCESS_KEY")) {

        (Some(k), Some(s)) if explicit_profile.is_none() => Some(Found {

            key_id: k,

            secret: s,

            token: env("AWS_SESSION_TOKEN"),

        }),

        _ => None,

    };

    let found = env_keys

        .or_else(|| {

            let c = creds.get(&profile)?;

            Some(Found {

                key_id: c.get("aws_access_key_id")?.clone(),

                secret: c.get("aws_secret_access_key")?.clone(),

                token: c.get("aws_session_token").cloned(),

            })

        })

        .or_else(|| {

            if known_profile {

                from_cli(&profile)

            } else {

                None

            }

        });

    match found {

        Some(f) => {

            b = b

                .with_access_key_id(f.key_id)

                .with_secret_access_key(f.secret);

            if let Some(t) = f.token {

                b = b.with_token(t);

            }

        }

        None if explicit_profile.is_some() && !known_profile => {

            return Err(Error::Backend(format!(

                "{}: profile {profile:?} is not in ~/.aws/config or ~/.aws/credentials: {HINT}",

                url.plain()

            )));

        }

        None => {}

    }

    let endpoint = url

        .param("endpoint")

        .map(str::to_string)

        .or_else(|| env("AWS_ENDPOINT_URL_S3"))

        .or_else(|| env("AWS_ENDPOINT_URL"))

        .or_else(|| env("AWS_ENDPOINT"))

        .or_else(|| section.and_then(|s| s.get("endpoint_url").cloned()));

    if let Some(e) = &endpoint {

        b = b

            .with_endpoint(e.clone())

            .with_allow_http(e.starts_with("http://"));

        // S3-alikes serve buckets by path, not by host name.

        b = b.with_virtual_hosted_style_request(false);

    }

    let region = url

        .param("region")

        .map(str::to_string)

        .or_else(|| env("AWS_REGION"))

        .or_else(|| env("AWS_DEFAULT_REGION"))

        .or_else(|| section.and_then(|s| s.get("region").cloned()));

    let region = match region {

        Some(r) => r,

        None if endpoint.is_some() => "us-east-1".into(),

        None => runtime

            .block_on(object_store::aws::resolve_bucket_region(&bucket, &ClientOptions::new()))

            .map_err(|e| {

                Error::Backend(format!(

                    "{}: no region was given and S3 did not say which region holds the bucket ({e}): add ?region=NAME to the remote URL, set AWS_REGION, or put region in the profile",

                    url.plain()

                ))

            })?,

    };

    b = b.with_region(region);

    b.build()

        .map_err(|e| Error::Backend(format!("{}: {e}: {HINT}", url.plain())))

}

#[cfg(test)]

mod tests {

    use super::*;

    #[test]

    fn profile_sections_are_found_under_either_heading() {

        let config =

            ini::parse("[default]\nregion = a\n[profile work]\nregion = b\n[plain]\nregion = c\n");

        assert_eq!(config_section(&config, "default").unwrap()["region"], "a");

        assert_eq!(config_section(&config, "work").unwrap()["region"], "b");

        assert_eq!(config_section(&config, "plain").unwrap()["region"], "c");

        assert!(config_section(&config, "none").is_none());

    }

}