//! 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", "[email protected]", "x", "salt", false).await.unwrap();

    db::accounts::create_account(&h.app.db, "bob", "[email protected]", "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", "[email protected]", "x", "salt", false).await.unwrap();

    db::accounts::create_account(&h.app.db, "bob", "[email protected]", "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);

}