//! The cloud stores against real services or emulators, when the
//! environment names them; otherwise each test says so and passes.
//!
//! - `OXY_TEST_S3=s3://BUCKET/PREFIX[?params]`, with `OXY_TEST_S3_PROFILE`
//!   added as `?profile=`;
//! - `OXY_TEST_GCS=gs://BUCKET/PREFIX[?params]`, with
//!   `OXY_TEST_GCS_CREDENTIALS` added as `?credentials=`;
//! - `OXY_TEST_AZURE=az://ACCOUNT/CONTAINER/PREFIX[?params]`, with the
//!   `AZURE_*` variables read from the file `OXY_TEST_AZURE_CREDENTIALS`.
//!
//! Every test works under its own prefix below the one given, and deletes
//! what it wrote.

use oxy_core::repo::Repo;
use oxy_core::strata::{conformance, Store};
use oxy_strata::CloudStore;
use std::path::Path;

fn configured(scheme: &str) -> Option<String> {
    let (var, extra, key) = match scheme {
        "s3" => ("OXY_TEST_S3", "OXY_TEST_S3_PROFILE", "profile"),
        "gs" => ("OXY_TEST_GCS", "OXY_TEST_GCS_CREDENTIALS", "credentials"),
        "az" => ("OXY_TEST_AZURE", "OXY_TEST_AZURE_CREDENTIALS", ""),
        _ => unreachable!(),
    };
    let url = std::env::var(var).ok().filter(|u| !u.is_empty())?;
    let extra_value = std::env::var(extra).ok().filter(|v| !v.is_empty());
    if scheme == "az" {
        if let Some(file) = &extra_value {
            for line in std::fs::read_to_string(file).unwrap_or_default().lines() {
                if let Some((k, v)) = line.split_once('=') {
                    std::env::set_var(k.trim(), v.trim());
                }
            }
        }
        return Some(url);
    }
    Some(match extra_value {
        Some(v) if !url.contains(&format!("{key}=")) => {
            let sep = if url.contains('?') { '&' } else { '?' };
            format!("{url}{sep}{key}={v}")
        }
        _ => url,
    })
}

fn each_store() -> Vec<(String, CloudStore)> {
    oxy_strata::register();
    ["s3", "gs", "az"]
        .into_iter()
        .filter_map(|s| {
            let Some(url) = configured(s) else {
                eprintln!("{s}: not configured, skipped");
                return None;
            };
            let store = CloudStore::open(&url).unwrap_or_else(|e| panic!("{url}: {e}"));
            Some((url, store))
        })
        .collect()
}

fn remove_all(store: &dyn Store, prefix: &str) {
    for e in store.list(prefix).unwrap() {
        store.delete(&e.key).unwrap();
    }
}

/// What earlier runs that failed may have left under the test prefix:
/// other processes' `probe-*` and `repo-*` keys.
fn remove_stale(store: &dyn Store) {
    let mine = format!("-{}", std::process::id());
    for e in store.list("").unwrap() {
        let first = e.key.split('/').next().unwrap_or("");
        if (first.starts_with("probe-") || first.starts_with("repo-")) && !first.ends_with(&mine) {
            store.delete(&e.key).unwrap();
        }
    }
}

#[test]
fn every_configured_store_conforms() {
    for (url, store) in each_store() {
        remove_stale(&store);
        let prefix = format!("probe-{}", std::process::id());
        let passed = conformance(&store, &prefix).unwrap_or_else(|e| {
            remove_all(&store, &prefix);
            panic!("{url}: {e}")
        });
        eprintln!("{url}: {} checks passed", passed.len());
        assert_eq!(passed.len(), 8);
    }
}

/// A small repository pushed to the store, cloned back, and pulled, by
/// URL: what a user does. Then every key the test wrote is removed.
#[test]
fn a_repository_round_trips_through_every_configured_store() {
    for (url, base_store) in each_store() {
        remove_stale(&base_store);
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        std::fs::create_dir_all(root.join("home")).unwrap();
        std::env::set_var("OXY_HOME", root.join("home"));
        let (base, query) = match url.split_once('?') {
            Some((b, q)) => (b.to_string(), format!("?{q}")),
            None => (url.clone(), String::new()),
        };
        let remote_url = format!("{base}/repo-{}{query}", std::process::id());
        let plain = format!("{base}/repo-{}", std::process::id());
        let store = CloudStore::open(&remote_url).unwrap();

        let mut a = Repo::init(&root.join("a")).unwrap();
        a.config.set("remote.cloud.url", &remote_url).unwrap();
        a.save_config().unwrap();
        write(root, "a/f.txt", b"hello from a\n");
        let h1 = a.commit("first", &[]).unwrap();
        let r = a.push("cloud", false).unwrap();
        assert!(r.iter().all(|x| x.ok), "{plain}: {r:?}");
        assert!(store.has("oxygen.toml").unwrap());
        assert_eq!(store.list("bundles/").unwrap().len(), 1, "{plain}");

        let b = Repo::clone(&remote_url, &root.join("b"))
            .unwrap_or_else(|e| panic!("{plain}: clone: {e}"));
        assert_eq!(
            std::fs::read(root.join("b/f.txt")).unwrap(),
            b"hello from a\n"
        );
        assert_eq!(b.bookmark_heads().unwrap(), vec![h1]);
        write(root, "b/f.txt", b"hello from a\nand b\n");
        b.commit("second", &[]).unwrap();
        assert!(b.push("origin", false).unwrap()[0].ok);
        let pr = a.pull("cloud").unwrap();
        assert!(pr.changed, "{plain}");
        assert_eq!(
            std::fs::read(root.join("a/f.txt")).unwrap(),
            b"hello from a\nand b\n"
        );
        let total: u64 = store.list("").unwrap().iter().map(|e| e.size).sum();
        eprintln!("{plain}: round trip ok, {total} bytes in the store before cleanup");
        remove_all(&store, "");
        assert!(store.list("").unwrap().is_empty(), "{plain}: cleanup");
    }
}

/// oxygen#68: a small repository kept on each store (`storage.url`), with
/// the local cache in front: commits, a reopen with the cache dropped, and
/// the time each step took. Then every key the test wrote is removed.
#[test]
fn a_repository_runs_on_every_configured_store() {
    for (url, base_store) in each_store() {
        remove_stale(&base_store);
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        std::fs::create_dir_all(root.join("home")).unwrap();
        std::env::set_var("OXY_HOME", root.join("home"));
        let (base, query) = match url.split_once('?') {
            Some((b, q)) => (b.to_string(), format!("?{q}")),
            None => (url.clone(), String::new()),
        };
        let store_url = format!("{base}/repo-{}{query}", std::process::id());
        let plain = format!("{base}/repo-{}", std::process::id());
        let store = CloudStore::open(&store_url).unwrap();
        let t = std::time::Instant::now();
        let repo = Repo::init_new(&root.join("w"), false, Some(&store_url))
            .unwrap_or_else(|e| panic!("{plain}: init: {e}"));
        let init_ms = t.elapsed().as_millis();
        let t = std::time::Instant::now();
        let mut last = None;
        for i in 0..5 {
            write(
                root,
                "w/f.txt",
                format!("line {i}\n").repeat(i + 1).as_bytes(),
            );
            last = Some(repo.commit(&format!("commit {i}"), &[]).unwrap());
        }
        let commit_ms = t.elapsed().as_millis() / 5;
        drop(repo);
        std::fs::remove_dir_all(root.join("w/.oxy/cache")).unwrap();
        let t = std::time::Instant::now();
        let repo = Repo::open(&root.join("w")).unwrap();
        let heads = repo.bookmark_heads().unwrap();
        let state = repo.load_state(&heads).unwrap().materialize();
        let cold_ms = t.elapsed().as_millis();
        assert_eq!(heads, vec![last.unwrap()], "{plain}");
        assert_eq!(state.files.len(), 1);
        assert!(
            store.read_ref("ops/head").unwrap().is_some(),
            "{plain}: the head is in the store"
        );
        let keys = store.list("").unwrap();
        let total: u64 = keys.iter().map(|e| e.size).sum();
        eprintln!(
            "{plain}: init {init_ms} ms, commit {commit_ms} ms each, cold open and read {cold_ms} ms; {} keys, {total} bytes",
            keys.len()
        );
        remove_all(&store, "");
        assert!(store.list("").unwrap().is_empty(), "{plain}: cleanup");
    }
}

/// Each store opened with credentials handed over (`CloudStore::open_with`,
/// as a hub does with the ones it keeps sealed), the profile and key file
/// parameters left off the URL, passes the conformance check.
#[test]
fn every_configured_store_conforms_with_credentials_handed_over() {
    use oxy_strata::Credentials;
    let env = |k: &str| std::env::var(k).ok().filter(|v| !v.is_empty());
    let mut tried = 0;
    if let (Some(url), Some(profile)) = (env("OXY_TEST_S3"), env("OXY_TEST_S3_PROFILE")) {
        let home = std::path::PathBuf::from(env("HOME").unwrap());
        let text = std::fs::read_to_string(home.join(".aws/credentials")).unwrap_or_default();
        let mut section = String::new();
        let mut keys = std::collections::BTreeMap::new();
        for line in text.lines().map(str::trim) {
            if let Some(name) = line.strip_prefix('[').and_then(|l| l.strip_suffix(']')) {
                section = name.to_string();
            } else if let Some((k, v)) = line.split_once('=') {
                if section == profile {
                    keys.insert(k.trim().to_string(), v.trim().to_string());
                }
            }
        }
        if let (Some(id), Some(secret)) = (
            keys.get("aws_access_key_id"),
            keys.get("aws_secret_access_key"),
        ) {
            let c = Credentials::Aws {
                access_key_id: id.clone(),
                secret_access_key: secret.clone(),
                session_token: keys.get("aws_session_token").cloned(),
            };
            check_handed_over(&url, &c);
            tried += 1;
        }
    }
    if let (Some(url), Some(file)) = (env("OXY_TEST_GCS"), env("OXY_TEST_GCS_CREDENTIALS")) {
        let c = Credentials::Gcp {
            service_account_key: std::fs::read_to_string(file).unwrap(),
        };
        check_handed_over(&url, &c);
        tried += 1;
    }
    if let (Some(url), Some(file)) = (env("OXY_TEST_AZURE"), env("OXY_TEST_AZURE_CREDENTIALS")) {
        let mut vars = std::collections::BTreeMap::new();
        for line in std::fs::read_to_string(file).unwrap().lines() {
            if let Some((k, v)) = line.split_once('=') {
                vars.insert(k.trim().to_string(), v.trim().to_string());
            }
        }
        let c = Credentials::Azure {
            tenant_id: vars["AZURE_TENANT_ID"].clone(),
            client_id: vars["AZURE_CLIENT_ID"].clone(),
            client_secret: vars["AZURE_CLIENT_SECRET"].clone(),
        };
        check_handed_over(&url, &c);
        tried += 1;
    }
    eprintln!("{tried} stores checked with credentials handed over");
}

fn check_handed_over(url: &str, c: &oxy_strata::Credentials) {
    let store =
        oxy_strata::CloudStore::open_with(url, Some(c)).unwrap_or_else(|e| panic!("{url}: {e}"));
    let prefix = format!("probe-{}", std::process::id());
    let passed = conformance(&store, &prefix).unwrap_or_else(|e| {
        remove_all(&store, &prefix);
        panic!("{url} with {c:?}: {e}")
    });
    eprintln!("{url} with {c:?}: {} checks passed", passed.len());
    assert_eq!(passed.len(), 8);
}

fn write(root: &Path, rel: &str, content: &[u8]) {
    let p = root.join(rel);
    std::fs::create_dir_all(p.parent().unwrap()).unwrap();
    std::fs::write(p, content).unwrap();
}