//! 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();
    // The first account's setup code (oxygenhub#245).
    config.registration.setup_code = "test-setup".into();
    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', $1, '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);
}

/// docs/backups.md §7 (BP1): a restore brings back the records as well as
/// the stores: issues with their comments, labels, milestone and
/// assignees, proposals with their reviews and events, releases with their
/// asset files, and webhooks without their secrets, every id mapped and
/// every account found again by name; what this hub lacks is said.
#[tokio::test(flavor = "multi_thread")]
async fn a_restore_brings_back_issues_proposals_and_releases() {
    let store = tempfile::tempdir().unwrap();
    let h = hub(store.path()).await;
    let andy = db::accounts::create_account(&h.app.db, "andy", "a@example.com", "x", "salt", false).await.unwrap();
    let bob = 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")]);
    let row = db::repos::get(&h.app.db, "andy", "demo").await.unwrap().unwrap();
    db::repos::set_diffusion(&h.app.db, row.id, "green").await.unwrap();
    // Ids on the first hub start further on than on the fresh one.
    for n in 0..3 {
        db::issues::upsert_label(&h.app.db, row.id, &format!("spare{n}"), "#000000", "").await.unwrap();
    }
    let bug = db::issues::upsert_label(&h.app.db, row.id, "bug", "#d73a4a", "Something is wrong").await.unwrap();
    let m = db::issues::upsert_milestone(&h.app.db, row.id, "1.0", "the first", None).await.unwrap();
    let issue = db::issues::create(
        &h.app.db,
        db::issues::NewIssue { repo_id: row.id, title: "Sprites flicker", body: "on level two", author_id: Some(andy.id), author: "andy" },
    )
    .await
    .unwrap();
    db::issues::add_comment(&h.app.db, issue.id, Some(bob.id), "bob", "seen it too").await.unwrap();
    db::issues::set_issue_label(&h.app.db, issue.id, bug.id, true).await.unwrap();
    db::issues::set_milestone(&h.app.db, issue.id, Some(m.id)).await.unwrap();
    db::issues::set_assignee(&h.app.db, issue.id, andy.id, true).await.unwrap();
    db::issues::set_assignee(&h.app.db, issue.id, bob.id, true).await.unwrap();
    let p = db::proposals::create(
        &h.app.db,
        db::proposals::NewProposal {
            repo_id: row.id,
            title: "Double-buffer the sprites",
            body: "fixes the flicker",
            author_id: Some(bob.id),
            author: "bob",
            draft: false,
            target: "main",
            source_kind: "bookmark",
            source_repo_id: Some(row.id),
            source_bookmark: "fix",
            source_changes: "",
        },
    )
    .await
    .unwrap();
    db::proposals::add_event(&h.app.db, p.id, "bob", "opened", "").await.unwrap();
    db::proposals::add_review(&h.app.db, p.id, andy.id, "andy", "approve", "nice").await.unwrap();
    let release = db::releases::create(&h.app.db, row.id, "v1.0", "One", "notes", false, Some((andy.id, "andy"))).await.unwrap();
    let asset = b"a game image".to_vec();
    let hash = blake3::hash(&asset).to_hex().to_string();
    let path = h.app.config.data.join("assets").join(&hash[..2]).join(&hash);
    std::fs::create_dir_all(path.parent().unwrap()).unwrap();
    std::fs::write(&path, &asset).unwrap();
    db::releases::add_asset(&h.app.db, release.id, "game.prg", asset.len() as i64, &hash, "application/octet-stream").await.unwrap();
    db::hooks::create_hook(&h.app.db, Some(row.id), None, "https://example.com/hook", "s3cret-value", "push").await.unwrap();

    let mut s = Setting::off("andy", "user");
    s.every = "hourly".into();
    s.target = store.path().display().to_string();
    backups::save(&h.app.db, &s).await.unwrap();
    let run = backups::run(&h.app, "andy", "now").await.unwrap();
    assert_eq!(run.status, "ok", "{}", run.detail);
    let dest = Directory { root: store.path().join("andy") };
    let text = String::from_utf8_lossy(&dest.get(&format!("runs/{}/records/demo.json", run.at)).unwrap()).into_owned();
    assert!(!text.contains("s3cret-value"), "never a secret");
    assert!(text.contains("\"_names\"") && text.contains("\"bob\""), "accounts by name");

    // A fresh hub that knows andy but not bob.
    let empty = tempfile::tempdir().unwrap();
    let fresh = hub(empty.path()).await;
    let andy2 = db::accounts::create_account(&fresh.app.db, "andy", "a@example.com", "x", "salt", false).await.unwrap();
    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[0], "andy/demo (repos)");
    let notes = done[1..].join("\n");
    assert!(notes.contains("webhook https://example.com/hook: enter its secret again"), "{notes}");
    assert!(notes.contains("issue_assignees: 1 row(s) left out"), "bob is not on this hub: {notes}");
    let r2 = db::repos::get(&fresh.app.db, "andy", "demo").await.unwrap().unwrap();
    assert_eq!(r2.diffusion, "green", "settings come back");
    // The issue, its comment, label, milestone and the assignee this hub has.
    let issues: Vec<(i64, i64, String, Option<i64>)> = sqlx::query_as("SELECT id, number, title, author_id FROM issues WHERE repo_id = $1")
        .bind(r2.id)
        .fetch_all(&fresh.app.db)
        .await
        .unwrap();
    assert_eq!(issues.len(), 1);
    let (iid, number, title, author) = issues[0].clone();
    assert_eq!((number, title.as_str(), author), (issue.number, "Sprites flicker", Some(andy2.id)));
    let comments = db::issues::comments(&fresh.app.db, iid).await.unwrap();
    assert_eq!(comments.len(), 1);
    assert_eq!((comments[0].author.as_str(), comments[0].body.as_str()), ("bob", "seen it too"));
    let labels: Vec<String> =
        sqlx::query_scalar("SELECT l.name FROM issue_labels il JOIN labels l ON l.id = il.label_id WHERE il.issue_id = $1")
            .bind(iid)
            .fetch_all(&fresh.app.db)
            .await
            .unwrap();
    assert_eq!(labels, vec!["bug"], "the label, its id mapped");
    let milestone: Option<String> =
        sqlx::query_scalar("SELECT m.title FROM issues i JOIN milestones m ON m.id = i.milestone_id WHERE i.id = $1")
            .bind(iid)
            .fetch_one(&fresh.app.db)
            .await
            .unwrap();
    assert_eq!(milestone.as_deref(), Some("1.0"));
    assert_eq!(db::issues::assignees(&fresh.app.db, iid).await.unwrap(), vec!["andy"]);
    // The proposal, its review and its timeline.
    let p2 = db::proposals::get(&fresh.app.db, r2.id, p.number).await.unwrap().expect("the proposal");
    assert_eq!((p2.title.as_str(), p2.author.as_str(), p2.author_id), ("Double-buffer the sprites", "bob", None));
    let reviews: Vec<(String, String)> = sqlx::query_as("SELECT author, verdict FROM proposal_reviews WHERE proposal_id = $1")
        .bind(p2.id)
        .fetch_all(&fresh.app.db)
        .await
        .unwrap();
    assert_eq!(reviews, vec![("andy".to_string(), "approve".to_string())]);
    assert!(db::proposals::events(&fresh.app.db, p2.id).await.unwrap().iter().any(|e| e.event == "opened"));
    // The release and its asset's bytes.
    let rel = db::releases::get(&fresh.app.db, r2.id, "v1.0").await.unwrap().expect("the release");
    let assets = db::releases::assets(&fresh.app.db, rel.id).await.unwrap();
    assert_eq!(assets.len(), 1);
    assert_eq!(std::fs::read(fresh.app.config.data.join("assets").join(&hash[..2]).join(&hash)).unwrap(), asset);
    // Search finds what came back.
    let hits = db::notify::search(&fresh.app.db, "flicker", Some(r2.id), 10).await.unwrap();
    assert!(hits.iter().any(|h| h.kind == "issue" && h.target_id == iid), "{:?}", hits.iter().map(|h| &h.title).collect::<Vec<_>>());
    // New issues number on from where the old hub was.
    let next = db::issues::create(
        &fresh.app.db,
        db::issues::NewIssue { repo_id: r2.id, title: "After", body: "", author_id: Some(andy2.id), author: "andy" },
    )
    .await
    .unwrap();
    assert!(next.number > p.number, "{} after {}", next.number, p.number);
}

/// docs/backups.md §5 (BP3): a repository left out is in no run; one with
/// its own schedule makes its owner's backup run at least that often, and
/// alone when its owner's is off; a run by hand takes all but those left
/// out.
#[tokio::test(flavor = "multi_thread")]
async fn repositories_choose_their_own_backup() {
    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();
    for n in ["game", "notes", "secret"] {
        create(&h.app, "andy", n, hub_api::Visibility::Public).await;
        let work = tempfile::tempdir().unwrap();
        push(&format!("{}/andy/{n}", h.base), work.path(), &[("README.md", n)]);
    }
    let id = |n: &'static str| {
        let db = h.app.db.clone();
        async move { db::repos::get(&db, "andy", n).await.unwrap().unwrap().id }
    };
    db::repos::set_backup(&h.app.db, id("secret").await, "off").await.unwrap();
    let mut s = Setting::off("andy", "user");
    s.every = "daily".into();
    s.target = store.path().display().to_string();
    backups::save(&h.app.db, &s).await.unwrap();
    let names = |at: String| {
        let dest = Directory { root: store.path().join("andy") };
        let mut v = dest.list(&format!("runs/{at}/records")).unwrap();
        v.sort();
        v
    };
    let first = backups::run(&h.app, "andy", "schedule").await.unwrap();
    assert_eq!((first.status.as_str(), first.repos), ("ok", 2), "{}", first.detail);
    assert_eq!(names(first.at.clone()), vec!["game.json", "notes.json"], "the one left out is not in it");
    // Daily, and it just ran: not due, until a repository asks for hourly.
    assert!(backups::due(&h.app.db).await.unwrap().is_empty());
    db::repos::set_backup(&h.app.db, id("game").await, "hourly").await.unwrap();
    sqlx::query("UPDATE backup_runs SET started_ms = started_ms - 2 * 3600 * 1000").execute(&h.app.db).await.unwrap();
    assert_eq!(backups::due(&h.app.db).await.unwrap().len(), 1, "hourly for the game");
    // The owner's schedule off: only the game, which asks for its own.
    s.every = "off".into();
    backups::save(&h.app.db, &s).await.unwrap();
    assert_eq!(backups::due(&h.app.db).await.unwrap().len(), 1);
    tokio::time::sleep(std::time::Duration::from_millis(1100)).await;
    let second = backups::run(&h.app, "andy", "schedule").await.unwrap();
    assert_eq!(names(second.at.clone()), vec!["game.json"]);
    // By hand: everything but the one left out.
    tokio::time::sleep(std::time::Duration::from_millis(1100)).await;
    let third = backups::run(&h.app, "andy", "now").await.unwrap();
    assert_eq!(names(third.at.clone()), vec!["game.json", "notes.json"]);
    // Admin › Backups counts what the next scheduled run covers.
    let (owners, uncovered) = backups::overview(&h.app.db).await.unwrap();
    assert_eq!((owners[0].repos, uncovered.len()), (1, 0));
}

/// AWS STS and S3, as much as backups use (path style, under `/s3`): an
/// AssumeRoleWithWebIdentity that admits `backup:` subjects with audience
/// `sts.amazonaws.com`, and buckets (`OWNER-backup`) that check every
/// request's Signature Version 4 and admit only their owner's session.
async fn fake_aws() -> (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();
        let header = |h: &str| parts.headers.get(h).and_then(|v| v.to_str().ok()).unwrap_or("").to_string();
        if path == "/" && parts.method == "POST" {
            let form: Vec<(String, String)> = String::from_utf8_lossy(&bytes)
                .split('&')
                .filter_map(|kv| kv.split_once('=').map(|(k, v)| (k.to_string(), decode(&v.replace('+', " ")))))
                .collect();
            let field = |k: &str| form.iter().find(|(f, _)| f == k).map(|(_, v)| v.clone()).unwrap_or_default();
            let jwt = field("WebIdentityToken");
            let claims: serde_json::Value = serde_json::from_slice(&base64url(jwt.split('.').nth(1).unwrap())).unwrap();
            let (sub, aud) = (claims["sub"].as_str().unwrap().to_string(), claims["aud"].as_str().unwrap().to_string());
            g.lock().unwrap().exchanged.push((sub.clone(), aud.clone()));
            if field("Action") != "AssumeRoleWithWebIdentity" || !sub.starts_with("backup:") || aud != "sts.amazonaws.com" {
                return (StatusCode::FORBIDDEN, "<ErrorResponse><Error><Code>AccessDenied</Code><Message>Not authorized to perform sts:AssumeRoleWithWebIdentity</Message></Error></ErrorResponse>").into_response();
            }
            let owner = sub.trim_start_matches("backup:");
            return format!(
                "<AssumeRoleWithWebIdentityResponse><AssumeRoleWithWebIdentityResult><Credentials>\
                 <AccessKeyId>ASIA{owner}</AccessKeyId><SecretAccessKey>secret/{owner}</SecretAccessKey>\
                 <SessionToken>session-{owner}</SessionToken><Expiration>2030-01-01T00:00:00Z</Expiration>\
                 </Credentials></AssumeRoleWithWebIdentityResult></AssumeRoleWithWebIdentityResponse>"
            )
            .into_response();
        }
        // /s3/{bucket}/{key…}
        let rest = path.strip_prefix("/s3/").unwrap_or("");
        let (bucket, key) = rest.split_once('/').unwrap_or((rest, ""));
        let (bucket, key) = (bucket.to_string(), decode(key));
        let owner = bucket.trim_end_matches("-backup").to_string();
        // Check the signature as AWS would, with the session's secret.
        let session = header("x-amz-security-token");
        let who = session.trim_start_matches("session-").to_string();
        let query: Vec<(String, String)> = parts
            .uri
            .query()
            .unwrap_or("")
            .split('&')
            .filter(|kv| !kv.is_empty())
            .map(|kv| kv.split_once('=').map(|(k, v)| (decode(k), decode(v))).unwrap_or((decode(kv), String::new())))
            .collect();
        let signed = vec![
            ("host".to_string(), header("host")),
            ("x-amz-content-sha256".to_string(), header("x-amz-content-sha256")),
            ("x-amz-date".to_string(), header("x-amz-date")),
            ("x-amz-security-token".to_string(), session.clone()),
        ];
        let id = format!("ASIA{who}");
        let secret = format!("secret/{who}");
        let want = oxygenhub::backups::sigv4(
            parts.method.as_str(),
            &path,
            &query,
            &signed,
            &header("x-amz-content-sha256"),
            (&id, &secret),
            "eu-west-2",
            "s3",
            &header("x-amz-date"),
        );
        use sha2::Digest;
        let body_hash = hex::encode(sha2::Sha256::digest(&bytes));
        if header("authorization") != want || header("x-amz-content-sha256") != body_hash {
            return (
                StatusCode::FORBIDDEN,
                "<Error><Code>SignatureDoesNotMatch</Code><Message>The request signature we calculated does not match</Message></Error>",
            )
                .into_response();
        }
        if who != owner {
            return (StatusCode::FORBIDDEN, "<Error><Code>AccessDenied</Code><Message>Access Denied</Message></Error>").into_response();
        }
        let mut g = g.lock().unwrap();
        match (parts.method.as_str(), key.is_empty()) {
            ("PUT", false) => {
                g.objects.insert((bucket, key), bytes.to_vec());
                StatusCode::OK.into_response()
            }
            ("GET", false) | ("HEAD", false) => match g.objects.get(&(bucket, key)) {
                Some(b) => b.clone().into_response(),
                None => (StatusCode::NOT_FOUND, "<Error><Code>NoSuchKey</Code></Error>").into_response(),
            },
            ("DELETE", false) => {
                g.objects.remove(&(bucket, key));
                StatusCode::NO_CONTENT.into_response()
            }
            ("GET", true) => {
                let prefix = query.iter().find(|(k, _)| k == "prefix").map(|(_, v)| v.clone()).unwrap_or_default();
                let delimited = query.iter().any(|(k, _)| k == "delimiter");
                let (mut keys, mut prefixes) = (String::new(), std::collections::BTreeSet::new());
                for (b, n) in g.objects.keys() {
                    if *b != bucket || !n.starts_with(&prefix) {
                        continue;
                    }
                    match n[prefix.len()..].find('/') {
                        Some(i) if delimited => {
                            prefixes.insert(format!("{prefix}{}", &n[prefix.len()..prefix.len() + i + 1]));
                        }
                        _ => keys.push_str(&format!("<Contents><Key>{n}</Key></Contents>")),
                    }
                }
                let common: String = prefixes.iter().map(|p| format!("<CommonPrefixes><Prefix>{p}</Prefix></CommonPrefixes>")).collect();
                format!("<ListBucketResult><Prefix>{prefix}</Prefix><IsTruncated>false</IsTruncated>{keys}{common}</ListBucketResult>")
                    .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)
}

async fn hub_aws(backups: &Path, aws: &str) -> Hub {
    hermetic();
    let dir = tempfile::tempdir().unwrap();
    let mut config = Config::load(dir.path()).unwrap();
    // The first account's setup code (oxygenhub#245).
    config.registration.setup_code = "test-setup".into();
    config.auth.push_tokens = vec!["site".into()];
    config.backups.directories = vec![backups.to_path_buf()];
    config.backups.aws_sts_url = aws.to_string();
    config.backups.s3_url = format!("{aws}/s3/{{bucket}}");
    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 }
}

const ROLE: &str = "arn:aws:iam::123456789012:role/oxygenhub-backup";

fn s3(owner: &str, bucket: &str) -> Setting {
    let mut s = Setting::off(owner, "user");
    s.every = "hourly".into();
    s.destination = "s3".into();
    s.target = bucket.into();
    s.provider = ROLE.into();
    s.region = "eu-west-2".into();
    s.prefix = "oxygen hub".into();
    s
}

/// docs/backups.md §3 (BP4): an S3 bucket, keylessly: the hub's own token
/// for AWS STS, then signed requests; a second run sends nothing; the
/// probe leaves the bucket as it was; a fresh hub restores the same
/// repository from it; another owner's session is refused, saying what the
/// role needs.
#[tokio::test(flavor = "multi_thread")]
async fn an_s3_backup_needs_no_key_and_restores_the_same_repository() {
    let (aws, seen) = fake_aws().await;
    let store = tempfile::tempdir().unwrap();
    let h = hub_aws(store.path(), &aws).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"), ("src/a b+c.rs", "fn main() {}\n")]);

    backups::save(&h.app.db, &s3("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!("oxygen hub/runs/{}/DONE", first.at)),
            "under the prefix"
        );
        assert_eq!(g.exchanged.first().map(|(s, a)| (s.as_str(), a.as_str())), Some(("backup:andy", "sts.amazonaws.com")));
    }
    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);
    let before = seen.lock().unwrap().objects.len();
    let dest = backups::destination(&h.app, &s3("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, "the probe cleans up");

    let empty = tempfile::tempdir().unwrap();
    let fresh = hub_aws(empty.path(), &aws).await;
    let dest: Arc<dyn Destination> = backups::destination(&fresh.app, &s3("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");

    backups::save(&h.app.db, &s3("bob", "andy-backup")).await.unwrap();
    let refused = backups::run(&h.app, "bob", "now").await.unwrap();
    assert_eq!(refused.status, "failed");
    assert!(refused.detail.contains("s3:PutObject") && refused.detail.contains(ROLE), "{}", refused.detail);
}

/// One API call with a bearer token: the status and the JSON answer.
async fn api(
    app: &Shared,
    method: &str,
    path: &str,
    token: &str,
    body: Option<serde_json::Value>,
) -> (axum::http::StatusCode, serde_json::Value) {
    use axum::body::Body;
    use http_body_util::BodyExt;
    use tower::ServiceExt;
    let mut req =
        axum::http::Request::builder().method(method).uri(format!("/api/v1{path}")).header("authorization", format!("Bearer {token}"));
    let body = match body {
        Some(b) => {
            req = req.header("content-type", "application/json");
            Body::from(b.to_string())
        }
        None => Body::empty(),
    };
    let res = oxygenhub::router(app.clone()).oneshot(req.body(body).unwrap()).await.unwrap();
    let st = res.status();
    let bytes = res.into_body().collect().await.unwrap().to_bytes();
    (st, serde_json::from_slice(&bytes).unwrap_or(serde_json::Value::Null))
}

/// oxygenhub#179: the API does what the Backups pages do, for the caller's
/// own repositories and an organisation it owns, with repo:admin: the
/// setting (checked as the page checks it), TEST, back up now, the runs.
#[tokio::test(flavor = "multi_thread")]
async fn backups_are_set_up_and_run_through_the_api() {
    let store = tempfile::tempdir().unwrap();
    let h = hub(store.path()).await;
    let andy = db::accounts::create_account(&h.app.db, "andy", "a@example.com", "x", "salt", false).await.unwrap();
    let bob = db::accounts::create_account(&h.app.db, "bob", "b@example.com", "x", "salt", false).await.unwrap();
    db::accounts::create_org(&h.app.db, "club", "Club", andy.id).await.unwrap();
    create(&h.app, "andy", "demo", hub_api::Visibility::Public).await;
    create(&h.app, "club", "game", hub_api::Visibility::Private).await;
    let token = |id: i64, scopes: &'static str| {
        let db = h.app.db.clone();
        async move {
            let (t, hash) = oxygenhub::auth::new_token();
            db::accounts::create_token(&db, id, "t", &hash, scopes, "", None).await.unwrap();
            t
        }
    };
    let (andy_t, bob_t, weak) = (token(andy.id, "repo:admin").await, token(bob.id, "repo:admin").await, token(andy.id, "repo:read").await);
    let dir = store.path().display().to_string();

    let (st, empty) = api(&h.app, "GET", "/user/backups", &andy_t, None).await;
    assert_eq!(st, 200);
    assert!(empty["setting"].is_null() && empty["runs"].as_array().unwrap().is_empty(), "{empty}");
    let (st, _) = api(&h.app, "POST", "/user/backups/run", &andy_t, None).await;
    assert_eq!(st, 409, "nothing saved yet");
    let (st, _) =
        api(&h.app, "PUT", "/user/backups", &weak, Some(serde_json::json!({ "every": "hourly", "destination": "dir", "dir": dir }))).await;
    assert_eq!(st, 403, "repo:admin only");
    let (st, bad) = api(
        &h.app,
        "PUT",
        "/user/backups",
        &andy_t,
        Some(serde_json::json!({ "every": "hourly", "destination": "gcs", "bucket": "andy-backup", "provider": "nope" })),
    )
    .await;
    assert_eq!(st, 400, "checked as the page checks it: {bad}");
    let (st, saved) = api(
        &h.app,
        "PUT",
        "/user/backups",
        &andy_t,
        Some(serde_json::json!({ "every": "hourly", "destination": "dir", "dir": dir, "keep_days": 30 })),
    )
    .await;
    assert_eq!(st, 200, "{saved}");
    assert_eq!((saved["setting"]["every"].as_str(), saved["place"].as_str()), (Some("hourly"), Some(format!("{dir}/andy").as_str())));
    let (st, probe) = api(&h.app, "POST", "/user/backups/test", &andy_t, None).await;
    assert_eq!((st.as_u16(), probe["ok"].as_bool()), (200, Some(true)), "{probe}");
    let (st, started) = api(&h.app, "POST", "/user/backups/run", &andy_t, None).await;
    assert_eq!(st, 202, "{started}");
    let mut done = serde_json::Value::Null;
    for _ in 0..200 {
        let (_, runs) = api(&h.app, "GET", "/user/backups/runs", &andy_t, None).await;
        if runs[0]["status"].as_str().is_some_and(|s| s != "running") {
            done = runs[0].clone();
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }
    assert_eq!((done["status"].as_str(), done["repos"].as_i64(), done["trigger"].as_str()), (Some("ok"), Some(1), Some("now")), "{done}");
    // A whole run after a failed one says it works again; the day's
    // summary tells the owner what ran (#183).
    sqlx::query("INSERT INTO backup_runs (owner, started_ms, finished_ms, status, at, trigger, detail) VALUES ('andy', 1, 2, 'failed', 'x', 'schedule', 'GCS refused')")
        .execute(&h.app.db)
        .await
        .unwrap();
    tokio::time::sleep(std::time::Duration::from_millis(1100)).await;
    let again = backups::run(&h.app, "andy", "now").await.unwrap();
    assert_eq!(again.status, "ok");
    let told = |db: oxygenhub::db::Pool| async move { db::notify::inbox(&db, andy.id, 20).await.unwrap() };
    assert!(told(h.app.db.clone()).await.iter().any(|n| n.kind == "backup.recovered" && n.title.contains("works again")));
    assert_eq!(backups::daily_summary(&h.app, db::now_ms(), true).await.unwrap(), 1);
    let summary = told(h.app.db.clone()).await.into_iter().find(|n| n.kind == "backup.summary").expect("a summary");
    assert!(summary.title.contains("Backups of andy today") && summary.title.contains("runs"), "{}", summary.title);

    // An organisation's, by its owner; another account sees no such thing.
    let (st, _) =
        api(&h.app, "PUT", "/orgs/club/backups", &bob_t, Some(serde_json::json!({ "every": "hourly", "destination": "dir", "dir": dir })))
            .await;
    assert_eq!(st, 404);
    let (st, org) = api(
        &h.app,
        "PUT",
        "/orgs/club/backups",
        &andy_t,
        Some(serde_json::json!({ "every": "daily", "destination": "dir", "dir": dir, "keep_days": 90 })),
    )
    .await;
    assert_eq!(st, 200, "{org}");
    assert_eq!((org["setting"]["kind"].as_str(), org["setting"]["keep_days"].as_i64()), (Some("org"), Some(90)));
    let (st, _) = api(&h.app, "GET", "/orgs/club/backups", &bob_t, None).await;
    assert_eq!(st, 404);
}