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