//! Backups (docs/backups.md, oxygenhub#169), end to end: a real `oxy` push,
//! a backup to a directory, a second run that sends nothing new, and a
//! restore into a fresh hub that gives back the same repository, byte for
//! byte; and an owner's backup never holding another's repositories.

use oxy_core::repo::Repo;
use oxygenhub::backups::{self, Destination, Directory, Setting};
use oxygenhub::config::Config;
use oxygenhub::repos::RepoName;
use oxygenhub::{db, Shared};
use std::path::Path;
use std::sync::Arc;

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(backups: &Path) -> Hub {
    hub_with(backups, None).await
}

async fn hub_with(backups: &Path, google: Option<&str>) -> Hub {
    hermetic();
    let dir = tempfile::tempdir().unwrap();
    let mut config = Config::load(dir.path()).unwrap();
    config.auth.push_tokens = vec!["site".into()];
    config.backups.directories = vec![backups.to_path_buf()];
    if let Some(g) = google {
        config.backups.sts_url = g.to_string();
        config.backups.storage_url = g.to_string();
    }
    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 }
}

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

async fn create(app: &Shared, owner: &str, name: &str, visibility: hub_api::Visibility) {
    app.repos.create(&RepoName::parse(&format!("{owner}/{name}")).unwrap(), None).unwrap();
    db::repos::create(&app.db, owner, name, "a demo", visibility).await.unwrap();
}

fn push(url: &str, work: &Path, files: &[(&str, &str)]) {
    let mut c = Repo::init(work).unwrap();
    c.config.set("remote.origin.url", url).unwrap();
    c.config.set("remote.origin.token", "site").unwrap();
    c.save_config().unwrap();
    for (f, t) in files {
        write(work, f, t);
    }
    c.snapshot("t").unwrap();
    c.commit("Some work", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
}

/// Every file under `dir`, with its bytes.
fn files(dir: &Path) -> Vec<(String, Vec<u8>)> {
    let mut out = Vec::new();
    let mut stack = vec![dir.to_path_buf()];
    while let Some(d) = stack.pop() {
        for e in std::fs::read_dir(&d).unwrap().flatten() {
            let p = e.path();
            if p.is_dir() {
                stack.push(p);
            } else {
                out.push((p.strip_prefix(dir).unwrap().to_string_lossy().into_owned(), std::fs::read(&p).unwrap()));
            }
        }
    }
    out.sort();
    out
}

#[tokio::test(flavor = "multi_thread")]
async fn a_backup_restores_the_same_repository_and_sends_only_what_changed() {
    let store = tempfile::tempdir().unwrap();
    let h = hub(store.path()).await;
    db::accounts::create_account(&h.app.db, "andy", "a@example.com", "x", "salt", false).await.unwrap();
    db::accounts::create_account(&h.app.db, "bob", "b@example.com", "x", "salt", false).await.unwrap();
    create(&h.app, "andy", "demo", hub_api::Visibility::Private).await;
    create(&h.app, "bob", "other", hub_api::Visibility::Public).await;
    let work = tempfile::tempdir().unwrap();
    push(&format!("{}/andy/demo", h.base), work.path(), &[("README.md", "hello\n"), ("src/main.rs", "fn main() {}\n")]);
    let other = tempfile::tempdir().unwrap();
    push(&format!("{}/bob/other", h.base), other.path(), &[("x.txt", "bob's\n")]);

    // Andy backs up to the directory the hub allows.
    let target = store.path().display().to_string();
    let mut s = Setting::off("andy", "user");
    s.every = "hourly".into();
    s.target = target.clone();
    backups::save(&h.app.db, &s).await.unwrap();
    let first = backups::run(&h.app, "andy", "now").await.unwrap();
    assert_eq!((first.status.as_str(), first.repos), ("ok", 1), "{}", first.detail);
    assert!(first.sent > 0 && first.bytes > 0);
    let dest = Directory { root: store.path().join("andy") };
    assert!(dest.exists(&format!("runs/{}/DONE", first.at)).unwrap());
    let records: serde_json::Value = serde_json::from_slice(&dest.get(&format!("runs/{}/records/demo.json", first.at)).unwrap()).unwrap();
    assert_eq!(records["repositories"][0]["visibility"], "private");
    let text = String::from_utf8_lossy(&dest.get(&format!("runs/{}/records/demo.json", first.at)).unwrap()).into_owned();
    assert!(!text.contains("bob/other") && !text.contains("\"other\""), "nothing of another owner");
    assert!(!store.path().join("bob").exists(), "bob set up no backup");

    // Nothing changed: the next run sends nothing new.
    tokio::time::sleep(std::time::Duration::from_millis(1100)).await;
    let second = backups::run(&h.app, "andy", "now").await.unwrap();
    assert_eq!((second.status.as_str(), second.sent), ("ok", 0), "{}", second.detail);

    // A fresh hub, restored from the backup, has the same repository.
    let empty = tempfile::tempdir().unwrap();
    let fresh = hub(empty.path()).await;
    let dest: Arc<dyn Destination> = Arc::new(Directory { root: store.path().join("andy") });
    let done = backups::restore_into_hub(&fresh.app, dest, None).await.unwrap();
    assert_eq!(done, vec!["andy/demo (repos)"]);
    let name = RepoName::parse("andy/demo").unwrap();
    assert_eq!(files(&fresh.app.repos.dir(&name)), files(&h.app.repos.dir(&name)), "byte for byte");
    let row = db::repos::get(&fresh.app.db, "andy", "demo").await.unwrap().expect("listed on the hub");
    assert_eq!((row.visibility.as_str(), row.description.as_str()), ("private", "a demo"));
    let (a, b) = (h.app.repos.open(&name).unwrap(), fresh.app.repos.open(&name).unwrap());
    assert_eq!(
        a.bookmarks().unwrap().iter().map(|b| b.heads.clone()).collect::<Vec<_>>(),
        b.bookmarks().unwrap().iter().map(|b| b.heads.clone()).collect::<Vec<_>>()
    );

    // A directory the hub does not allow is refused.
    let mut bad = Setting::off("andy", "user");
    bad.every = "hourly".into();
    bad.target = "/tmp/somewhere-else".into();
    backups::save(&h.app.db, &bad).await.unwrap();
    let refused = backups::run(&h.app, "andy", "now").await.unwrap();
    assert_eq!(refused.status, "failed");
    assert!(refused.detail.contains("not a backup directory this hub allows"), "{}", refused.detail);
}

#[tokio::test(flavor = "multi_thread")]
async fn backups_fall_due_on_their_schedule() {
    let store = tempfile::tempdir().unwrap();
    let h = hub(store.path()).await;
    let mut s = Setting::off("andy", "user");
    s.target = store.path().display().to_string();
    backups::save(&h.app.db, &s).await.unwrap();
    assert!(backups::due(&h.app.db).await.unwrap().is_empty(), "off is never due");
    s.every = "hourly".into();
    backups::save(&h.app.db, &s).await.unwrap();
    assert_eq!(backups::due(&h.app.db).await.unwrap().len(), 1, "never run: due");
    sqlx::query("INSERT INTO backup_runs (owner, started_ms, status) VALUES ('andy', ?, 'ok')")
        .bind(db::now_ms())
        .execute(&h.app.db)
        .await
        .unwrap();
    assert!(backups::due(&h.app.db).await.unwrap().is_empty(), "ran just now");
}

// ------------------------------------------------------------ a fake Google

/// What the fake has seen and holds.
#[derive(Default)]
struct Google {
    /// Objects by bucket and name.
    objects: std::collections::BTreeMap<(String, String), Vec<u8>>,
    /// The subjects and audiences of the tokens exchanged.
    exchanged: Vec<(String, String)>,
}

type Shared2 = Arc<std::sync::Mutex<Google>>;

fn decode(s: &str) -> String {
    let b = s.as_bytes();
    let mut out = Vec::new();
    let mut i = 0;
    while i < b.len() {
        if b[i] == b'%' && i + 2 < b.len() {
            out.push(u8::from_str_radix(&s[i + 1..i + 3], 16).unwrap());
            i += 3;
        } else {
            out.push(b[i]);
            i += 1;
        }
    }
    String::from_utf8(out).unwrap()
}

fn query(uri: &axum::http::Uri, key: &str) -> Option<String> {
    uri.query()?.split('&').find_map(|kv| kv.split_once('=').filter(|(k, _)| *k == key).map(|(_, v)| decode(v)))
}

/// Google's STS and Cloud Storage, as much as backups use: a token
/// exchange that admits `backup:` subjects for the provider's audience,
/// and buckets that admit only their owner's token (bucket `OWNER-backup`).
async fn fake_google() -> (String, Shared2) {
    use axum::extract::{Request, State};
    use axum::http::StatusCode;
    use axum::response::IntoResponse;
    let state: Shared2 = Default::default();
    async fn handle(State(g): State<Shared2>, req: Request) -> axum::response::Response {
        let (parts, body) = req.into_parts();
        let bytes = http_body_util::BodyExt::collect(body).await.unwrap().to_bytes();
        let path = parts.uri.path().to_string();
        if path == "/v1/token" {
            let v: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
            let jwt = v["subjectToken"].as_str().unwrap();
            let payload = jwt.split('.').nth(1).unwrap();
            let claims: serde_json::Value = serde_json::from_slice(&base64url(payload)).unwrap();
            let (sub, aud) = (claims["sub"].as_str().unwrap().to_string(), claims["aud"].as_str().unwrap().to_string());
            let provider = v["audience"].as_str().unwrap();
            g.lock().unwrap().exchanged.push((sub.clone(), aud.clone()));
            if !sub.starts_with("backup:") || aud != format!("https:{provider}") {
                return (
                    StatusCode::BAD_REQUEST,
                    axum::Json(serde_json::json!({"error": "invalid_grant", "error_description": "not admitted"})),
                )
                    .into_response();
            }
            return axum::Json(serde_json::json!({ "access_token": format!("tok-{sub}"), "expires_in": 3600 })).into_response();
        }
        let auth = parts.headers.get("authorization").and_then(|v| v.to_str().ok()).unwrap_or("").to_string();
        let segs: Vec<&str> = path.split('/').collect();
        // /storage/v1/b/{bucket}/o[/{name}]  or  /upload/storage/v1/b/{bucket}/o
        let upload = segs.get(1) == Some(&"upload");
        let at = if upload { 5 } else { 4 };
        let bucket = segs.get(at).unwrap_or(&"").to_string();
        let owner = bucket.trim_end_matches("-backup");
        if auth != format!("Bearer tok-backup:{owner}") {
            return (StatusCode::FORBIDDEN, "caller does not have storage.objects access").into_response();
        }
        let mut g = g.lock().unwrap();
        if upload {
            let name = query(&parts.uri, "name").unwrap();
            g.objects.insert((bucket, name), bytes.to_vec());
            return axum::Json(serde_json::json!({})).into_response();
        }
        let name = segs.get(at + 2).map(|n| decode(n));
        match (parts.method.as_str(), name) {
            ("GET", Some(n)) => match g.objects.get(&(bucket, n.clone())) {
                Some(b) if query(&parts.uri, "alt").as_deref() == Some("media") => b.clone().into_response(),
                Some(_) => axum::Json(serde_json::json!({ "name": n })).into_response(),
                None => StatusCode::NOT_FOUND.into_response(),
            },
            ("DELETE", Some(n)) => {
                g.objects.remove(&(bucket, n));
                StatusCode::NO_CONTENT.into_response()
            }
            ("GET", None) => {
                let prefix = query(&parts.uri, "prefix").unwrap_or_default();
                let delimited = query(&parts.uri, "delimiter").is_some();
                let (mut items, mut prefixes) = (Vec::new(), std::collections::BTreeSet::new());
                for (b, n) in g.objects.keys() {
                    if *b != bucket || !n.starts_with(&prefix) {
                        continue;
                    }
                    let rest = &n[prefix.len()..];
                    match rest.find('/') {
                        Some(i) if delimited => {
                            prefixes.insert(format!("{prefix}{}", &rest[..=i]));
                        }
                        _ => items.push(serde_json::json!({ "name": n })),
                    }
                }
                axum::Json(serde_json::json!({ "items": items, "prefixes": prefixes })).into_response()
            }
            _ => StatusCode::METHOD_NOT_ALLOWED.into_response(),
        }
    }
    let app = axum::Router::new().fallback(handle).with_state(state.clone());
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let port = listener.local_addr().unwrap().port();
    tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
    (format!("http://127.0.0.1:{port}"), state)
}

fn base64url(s: &str) -> Vec<u8> {
    let table = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_";
    let mut bits = 0u32;
    let mut n = 0;
    let mut out = Vec::new();
    for c in s.bytes() {
        let v = table.iter().position(|t| *t == c).unwrap() as u32;
        bits = (bits << 6) | v;
        n += 6;
        if n >= 8 {
            n -= 8;
            out.push((bits >> n) as u8);
            bits &= (1 << n) - 1;
        }
    }
    out
}

const PROVIDER: &str = "//iam.googleapis.com/projects/1/locations/global/workloadIdentityPools/oxygenhub/providers/oxygenhub-io";

fn gcs(owner: &str, bucket: &str) -> Setting {
    let mut s = Setting::off(owner, "user");
    s.every = "hourly".into();
    s.destination = "gcs".into();
    s.target = bucket.into();
    s.provider = PROVIDER.into();
    s
}

#[tokio::test(flavor = "multi_thread")]
async fn a_gcs_backup_needs_no_key_and_restores_the_same_repository() {
    let (google, seen) = fake_google().await;
    let store = tempfile::tempdir().unwrap();
    let h = hub_with(store.path(), Some(&google)).await;
    db::accounts::create_account(&h.app.db, "andy", "a@example.com", "x", "salt", false).await.unwrap();
    db::accounts::create_account(&h.app.db, "bob", "b@example.com", "x", "salt", false).await.unwrap();
    create(&h.app, "andy", "demo", hub_api::Visibility::Public).await;
    let work = tempfile::tempdir().unwrap();
    push(&format!("{}/andy/demo", h.base), work.path(), &[("README.md", "hello\n")]);

    backups::save(&h.app.db, &gcs("andy", "andy-backup")).await.unwrap();
    let first = backups::run(&h.app, "andy", "now").await.unwrap();
    assert_eq!((first.status.as_str(), first.repos), ("ok", 1), "{}", first.detail);
    {
        let g = seen.lock().unwrap();
        assert!(g.objects.keys().any(|(b, n)| b == "andy-backup" && n == &format!("runs/{}/DONE", first.at)));
        assert!(g.objects.keys().all(|(b, _)| b == "andy-backup"));
        let (sub, aud) = g.exchanged.first().unwrap();
        assert_eq!(
            (sub.as_str(), aud.as_str()),
            ("backup:andy", format!("https:{PROVIDER}").as_str()),
            "the hub signs in as the owner, for the provider"
        );
    }
    tokio::time::sleep(std::time::Duration::from_millis(1100)).await;
    let second = backups::run(&h.app, "andy", "now").await.unwrap();
    assert_eq!((second.status.as_str(), second.sent), ("ok", 0), "{}", second.detail);
    // The probe writes, reads and removes; the bucket is as it was.
    let before = seen.lock().unwrap().objects.len();
    let dest = backups::destination(&h.app, &gcs("andy", "andy-backup")).unwrap();
    tokio::task::spawn_blocking(move || backups::probe(dest.as_ref())).await.unwrap().unwrap();
    assert_eq!(seen.lock().unwrap().objects.len(), before);

    // Restored from the bucket into a fresh hub: the same repository.
    let empty = tempfile::tempdir().unwrap();
    let fresh = hub_with(empty.path(), Some(&google)).await;
    let dest: Arc<dyn Destination> = backups::destination(&fresh.app, &gcs("andy", "andy-backup")).unwrap().into();
    assert_eq!(backups::restore_into_hub(&fresh.app, dest, None).await.unwrap(), vec!["andy/demo (repos)"]);
    let name = RepoName::parse("andy/demo").unwrap();
    assert_eq!(files(&fresh.app.repos.dir(&name)), files(&h.app.repos.dir(&name)), "byte for byte");

    // Bob cannot write Andy's bucket: Google refuses his token, and the
    // run says what to grant.
    backups::save(&h.app.db, &gcs("bob", "andy-backup")).await.unwrap();
    let refused = backups::run(&h.app, "bob", "now").await.unwrap();
    assert_eq!(refused.status, "failed");
    assert!(refused.detail.contains("roles/storage.objectAdmin") && refused.detail.contains("backup:bob"), "{}", refused.detail);
}