//! Repositories kept in their owner's bucket (oxygenhub#96): the New
//! repository form refuses what the hub must not accept and creates
//! nothing then; with a real S3 test bucket configured (`OXY_TEST_S3` and
//! `OXY_TEST_S3_PROFILE`, keys read from that profile and handed over), a
//! repository is made in the bucket, takes a push, shows where it lives,
//! passes Settings → Storage → Check, and the test removes what it wrote.

use axum::body::Body;
use axum::http::{header, Request, StatusCode};
use http_body_util::BodyExt;
use oxy_core::repo::Repo;
use oxygenhub::config::Config;
use oxygenhub::repos::RepoName;
use oxygenhub::{db, Shared};
use tower::ServiceExt;

fn hermetic() {
    static ONCE: std::sync::Once = std::sync::Once::new();
    static DIR: std::sync::OnceLock<tempfile::TempDir> = std::sync::OnceLock::new();
    ONCE.call_once(|| {
        let d = tempfile::tempdir().unwrap();
        std::env::set_var("OXY_HOME", d.path());
        DIR.set(d).ok();
    });
}

struct Hub {
    app: Shared,
    base: String,
    _dir: tempfile::TempDir,
}

async fn hub() -> Hub {
    hermetic();
    let dir = tempfile::tempdir().unwrap();
    let mut config = Config::load(dir.path()).unwrap();
    config.auth.push_tokens = vec!["site".into()];
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let port = listener.local_addr().unwrap().port();
    config.server.base_url = format!("http://127.0.0.1:{port}");
    let app = oxygenhub::build(config).await.unwrap();
    let router = oxygenhub::router(app.clone());
    tokio::spawn(async move { axum::serve(listener, router).await.unwrap() });
    Hub { app, base: format!("http://127.0.0.1:{port}"), _dir: dir }
}

async fn send(app: &Shared, method: &str, path: &str, cookie: Option<&str>, form: Option<&str>) -> (StatusCode, String, Option<String>) {
    let mut req = Request::builder().method(method).uri(path);
    if let Some(c) = cookie {
        req = req.header(header::COOKIE, format!("oxyhub={c}"));
    }
    let body = match form {
        Some(f) => {
            req = req.header(header::CONTENT_TYPE, "application/x-www-form-urlencoded");
            Body::from(f.to_string())
        }
        None => Body::empty(),
    };
    let res = oxygenhub::router(app.clone()).oneshot(req.body(body).unwrap()).await.unwrap();
    let status = res.status();
    let cookie = res
        .headers()
        .get_all(header::SET_COOKIE)
        .iter()
        .filter_map(|v| v.to_str().ok())
        .filter_map(|c| c.split(';').next())
        .filter_map(|kv| kv.strip_prefix("oxyhub="))
        .map(String::from)
        .next();
    let bytes = res.into_body().collect().await.unwrap().to_bytes();
    (status, String::from_utf8_lossy(&bytes).into_owned(), cookie)
}

fn csrf_of(page: &str) -> String {
    let i = page.find("name=\"_csrf\" value=\"").expect("csrf field") + "name=\"_csrf\" value=\"".len();
    page[i..].split('"').next().unwrap().to_string()
}

fn enc(s: &str) -> String {
    s.bytes()
        .map(|b| if b.is_ascii_alphanumeric() || b"-_.~".contains(&b) { (b as char).to_string() } else { format!("%{b:02X}") })
        .collect()
}

async fn andy(app: &Shared) -> String {
    let (st, body, cookie) =
        send(app, "POST", "/register", None, Some("username=andy&email=andy%40example.com&password=correct-horse-battery&invite=")).await;
    assert_eq!(st, StatusCode::SEE_OTHER, "{body}");
    cookie.expect("session")
}

async fn new_repo(app: &Shared, me: &str, fields: &str) -> (StatusCode, String) {
    let csrf = csrf_of(&send(app, "GET", "/new", Some(me), None).await.1);
    let (st, body, _) = send(app, "POST", "/new", Some(me), Some(&format!("_csrf={csrf}&owner=andy&{fields}"))).await;
    (st, body)
}

#[tokio::test(flavor = "multi_thread")]
async fn a_bucket_the_hub_must_not_use_is_refused_and_nothing_is_made() {
    let h = hub().await;
    let me = andy(&h.app).await;
    for (url, why) in [
        ("s3://b/p?profile=default", "?profile= is not allowed"),
        ("s3://b/p?endpoint=http%3A%2F%2F169.254.169.254", "?endpoint= is not allowed"),
        ("gs://b/p?credentials=%2Fetc%2Fkey.json", "?credentials= is not allowed"),
        ("dir:///var/lib/oxygenhub", "a store is s3://"),
        ("az://account", "names the account and the container"),
    ] {
        let (st, body) =
            new_repo(&h.app, &me, &format!("name=cloudy&storage=bucket&store_url={}&access_key_id=a&secret_access_key=b", enc(url))).await;
        assert_eq!(st, StatusCode::OK, "{url}");
        assert!(body.contains(why), "{url}: {body}");
    }
    let (_, body) = new_repo(&h.app, &me, &format!("name=cloudy&storage=bucket&store_url={}&access_key_id=a", enc("s3://b/p"))).await;
    assert!(body.contains("Credentials: credentials need secret_access_key"), "{body}");
    let name = RepoName::new("andy", "cloudy").unwrap();
    assert!(!h.app.repos.exists(&name));
    assert!(db::repos::get(&h.app.db, "andy", "cloudy").await.unwrap().is_none());
    // The hub's own disk stays the default.
    let (st, body) = new_repo(&h.app, &me, "name=local").await;
    assert_eq!(st, StatusCode::SEE_OTHER, "{body}");
    let page = send(&h.app, "GET", "/andy/local", Some(&me), None).await.1;
    assert!(!page.contains("class=\"store store-"), "no store icon on a repository on this disk");
}

/// The S3 test bucket's URL with only its region kept, and the profile's keys.
fn s3_test() -> Option<(String, String, String)> {
    let url = std::env::var("OXY_TEST_S3").ok().filter(|u| !u.is_empty())?;
    let profile = std::env::var("OXY_TEST_S3_PROFILE").ok().filter(|p| !p.is_empty())?;
    let (base, query) = url.split_once('?').unwrap_or((&url, ""));
    let region: Vec<&str> = query.split('&').filter(|p| p.starts_with("region=")).collect();
    let home = std::path::PathBuf::from(std::env::var("HOME").ok()?);
    let text = std::fs::read_to_string(home.join(".aws/credentials")).ok()?;
    let (mut section, mut id, mut secret) = (String::new(), None, None);
    for line in text.lines().map(str::trim) {
        if let Some(n) = line.strip_prefix('[').and_then(|l| l.strip_suffix(']')) {
            section = n.to_string();
        } else if let Some((k, v)) = line.split_once('=') {
            if section == profile {
                match k.trim() {
                    "aws_access_key_id" => id = Some(v.trim().to_string()),
                    "aws_secret_access_key" => secret = Some(v.trim().to_string()),
                    _ => {}
                }
            }
        }
    }
    let base = format!("{base}/hub-{}", std::process::id());
    let url = if region.is_empty() { base } else { format!("{base}?{}", region.join("&")) };
    Some((url, id?, secret?))
}

#[tokio::test(flavor = "multi_thread")]
async fn a_repository_in_a_real_bucket_takes_a_push_and_passes_its_check() {
    let Some((url, id, secret)) = s3_test() else {
        eprintln!("skipped: OXY_TEST_S3 and OXY_TEST_S3_PROFILE are not set");
        return;
    };
    let h = hub().await;
    let me = andy(&h.app).await;
    let fields = format!(
        "name=cloudy&visibility=private&storage=bucket&store_url={}&access_key_id={}&secret_access_key={}",
        enc(&url),
        enc(&id),
        enc(&secret)
    );
    let (st, body) = new_repo(&h.app, &me, &fields).await;
    let credentials = oxy_strata::Credentials::Aws { access_key_id: id, secret_access_key: secret, session_token: None };
    let store = oxy_strata::CloudStore::open_with(&url, Some(&credentials)).unwrap();
    let cleanup = || {
        use oxy_core::strata::Store as _;
        for e in store.list("").unwrap() {
            store.delete(&e.key).unwrap();
        }
    };
    if st != StatusCode::SEE_OTHER {
        cleanup();
        panic!("{body}");
    }
    let result = std::panic::AssertUnwindSafe(async {
        // A push from a client lands in the bucket.
        let work = tempfile::tempdir().unwrap();
        let mut c = Repo::init(work.path()).unwrap();
        c.config.set("remote.origin.url", &format!("{}/andy/cloudy", h.base)).unwrap();
        c.config.set("remote.origin.token", "site").unwrap();
        c.save_config().unwrap();
        std::fs::write(work.path().join("a.txt"), "kept in a bucket\n").unwrap();
        let head = c.commit("First", &[]).unwrap();
        let pushed = tokio::task::spawn_blocking(move || c.push("origin", false).unwrap()).await.unwrap();
        assert!(pushed.iter().all(|r| r.ok), "{pushed:?}");
        let hosted = h.app.repos.open(&RepoName::new("andy", "cloudy").unwrap()).unwrap();
        assert_eq!(hosted.bookmark_heads().unwrap(), vec![head]);
        {
            use oxy_core::strata::Store as _;
            assert!(store.read_ref("ops/head").unwrap().is_some(), "the operation log is in the bucket");
        }
        // The pages say where it lives; Settings → Storage checks the bucket.
        let page = send(&h.app, "GET", "/andy/cloudy", Some(&me), None).await.1;
        assert!(page.contains("class=\"store store-s3\"") && page.contains(url.split('?').next().unwrap()), "{page}");
        let settings = send(&h.app, "GET", "/andy/cloudy/settings", Some(&me), None).await.1;
        assert!(settings.contains("id=\"storage\""), "{settings}");
        let csrf = csrf_of(&settings);
        let (_, after, _) =
            send(&h.app, "POST", "/andy/cloudy/settings/storage", Some(&me), Some(&format!("_csrf={csrf}&action=check"))).await;
        assert!(after.contains("The bucket passed the check: 8 checks passed."), "{after}");
        // A second repository cannot start in the same place.
        let (st, body) = new_repo(&h.app, &me, &fields.replace("name=cloudy", "name=again")).await;
        assert_eq!(st, StatusCode::OK);
        assert!(body.contains("already holds"), "{body}");
    });
    let outcome = futures_catch(result).await;
    cleanup();
    if let Err(p) = outcome {
        std::panic::resume_unwind(p);
    }
}

async fn futures_catch<F: std::future::Future<Output = ()>>(f: std::panic::AssertUnwindSafe<F>) -> std::thread::Result<()> {
    use futures::FutureExt;
    f.catch_unwind().await
}

#[tokio::test(flavor = "multi_thread")]
async fn an_organisation_can_require_its_own_buckets() {
    let h = hub().await;
    let me = andy(&h.app).await;
    let csrf = csrf_of(&send(&h.app, "GET", "/orgs/new", Some(&me), None).await.1);
    let (st, body, _) = send(&h.app, "POST", "/orgs/new", Some(&me), Some(&format!("_csrf={csrf}&name=acme&display_name=Acme"))).await;
    assert_eq!(st, StatusCode::SEE_OTHER, "{body}");
    let settings = send(&h.app, "GET", "/orgs/acme/settings", Some(&me), None).await.1;
    assert!(settings.contains("id=\"storage\""));
    let csrf = csrf_of(&settings);
    let set = |allowed: &str| {
        let form = format!("_csrf={csrf}&require_bucket=1&allowed={}", enc(allowed));
        let app = h.app.clone();
        let me = me.clone();
        async move { send(&app, "POST", "/orgs/acme/settings/storage", Some(&me), Some(&form)).await.1 }
    };
    let page = set("s3://acme-london/repos/?region=eu-west-2").await;
    assert!(page.contains("a prefix names a bucket and a path, with no settings"), "{page}");
    let page = set("s3://acme-london/repos/\ngs://acme-eu/").await;
    assert!(page.contains("Storage policy saved: every new repository in s3://acme-london/repos/, gs://acme-eu/."), "{page}");

    let new = |fields: String| {
        let app = h.app.clone();
        let me = me.clone();
        async move {
            let csrf = csrf_of(&send(&app, "GET", "/new", Some(&me), None).await.1);
            send(&app, "POST", "/new", Some(&me), Some(&format!("_csrf={csrf}&owner=acme&{fields}"))).await.1
        }
    };
    let page = new("name=ondisk".into()).await;
    assert!(page.contains("acme keeps every new repository in a bucket"), "{page}");
    let page = new(format!("name=elsewhere&storage=bucket&store_url={}&access_key_id=a&secret_access_key=b", enc("s3://other/x"))).await;
    assert!(page.contains("acme allows only these stores: s3://acme-london/repos/, gs://acme-eu/"), "{page}");
    assert!(db::repos::get(&h.app.db, "acme", "ondisk").await.unwrap().is_none());
    assert!(db::repos::get(&h.app.db, "acme", "elsewhere").await.unwrap().is_none());
    // A user's own repositories are not the organisation's business.
    let (st, body) = new_repo(&h.app, &me, "name=mine").await;
    assert_eq!(st, StatusCode::SEE_OTHER, "{body}");
}