//! knotless-serve against a reference PDS: real git clients over smart
//! HTTP, and the API.

#[path = "../../knotless/tests/common/mod.rs"]
mod common;
use common::*;

use knotless::commit::{Change, Spec};
use knotless_serve::{App, router};
use std::io::Write;
use std::sync::Arc;
use std::sync::atomic::Ordering;
use std::time::Duration;

struct Server {
    base: String,
    app: Arc<App>,
    rt: tokio::runtime::Runtime,
}

fn serve(s: &Sandbox, ttl: Duration) -> Server {
    let rt = tokio::runtime::Runtime::new().unwrap();
    let app = App::new(ctx(s), ttl);
    let listener = rt
        .block_on(tokio::net::TcpListener::bind("127.0.0.1:0"))
        .unwrap();
    let base = format!("http://{}", listener.local_addr().unwrap());
    let router = router(app.clone());
    rt.spawn(async move { axum::serve(listener, router).await.unwrap() });
    Server { base, app, rt }
}

fn put(path: &str, data: &str) -> Change {
    Change::Put {
        path: path.into(),
        data: data.as_bytes().to_vec(),
        executable: false,
    }
}

fn commit(s: &Sandbox, name: &str, changes: Vec<Change>, message: &str) -> String {
    let rt = tokio::runtime::Runtime::new().unwrap();
    let url = knotless::url::RepoUrl::parse(&s.url(name)).unwrap();
    let spec = Spec {
        branch: "main".into(),
        message: message.into(),
        changes,
        author: None,
        retries: 3,
        ..Default::default()
    };
    rt.block_on(knotless::commit::commit(&ctx(s), &url, &spec))
        .unwrap()
        .sha
}

fn get(url: &str) -> (u16, String, reqwest::header::HeaderMap) {
    let rt = tokio::runtime::Runtime::new().unwrap();
    rt.block_on(async {
        let res = reqwest::get(url).await.unwrap();
        let status = res.status().as_u16();
        let headers = res.headers().clone();
        (status, res.text().await.unwrap(), headers)
    })
}

#[test]
fn git_clients_over_smart_http() {
    let s = Sandbox::new();
    commit(
        &s,
        "web",
        vec![put("a.txt", "one\n"), put("dir/b.txt", "two\n")],
        "first",
    );
    let tip = commit(&s, "web", vec![put("a.txt", "three\n")], "second");
    let server = serve(&s, Duration::ZERO);
    let url = format!("{}/{}/web.git", server.base, s.account.handle);

    let cases: [(&str, &[&str], &[&str]); 4] = [
        ("v2", &[], &[]),
        ("v0", &["-c", "protocol.version=0"], &[]),
        ("partial", &[], &["--filter=blob:none"]),
        ("shallow", &[], &["--depth", "1"]),
    ];
    for (dir, config, flags) in cases {
        let path = s.path(dir);
        let mut args: Vec<&str> = config.to_vec();
        args.extend(["clone", "-q"]);
        args.extend(flags);
        args.extend([url.as_str(), path.to_str().unwrap()]);
        s.git(s.dir.path(), &args);
        assert_eq!(s.git(&path, &["rev-parse", "HEAD"]), tip, "{dir}");
        assert_eq!(
            std::fs::read_to_string(path.join("a.txt")).unwrap(),
            "three\n",
            "{dir}"
        );
    }
    assert_eq!(
        s.git(&s.path("shallow"), &["rev-list", "--count", "HEAD"]),
        "1"
    );

    // Pushing goes to the PDS, never here.
    let (status, _, _) = get(&format!("{url}/info/refs?service=git-receive-pack"));
    assert_eq!(status, 403);
    let (status, body, _) = get(&format!("{}/healthz", server.base));
    assert_eq!((status, body.as_str()), (200, "ok"));
    drop(server.rt);
}

#[test]
fn gzip_requests() {
    let s = Sandbox::new();
    commit(&s, "gz", vec![put("a", "a")], "c");
    let server = serve(&s, Duration::from_secs(60));
    let url = format!("{}/{}/gz/git-upload-pack", server.base, s.account.handle);
    let mut gz = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default());
    gz.write_all(b"0000").unwrap();
    let body = gz.finish().unwrap();
    server.rt.block_on(async {
        let client = reqwest::Client::new();
        let ok = client
            .post(&url)
            .header("content-encoding", "gzip")
            .body(body)
            .send()
            .await
            .unwrap();
        assert_eq!(ok.status(), 200);
        let bad = client
            .post(&url)
            .header("content-encoding", "gzip")
            .body("nope")
            .send()
            .await
            .unwrap();
        assert_eq!(bad.status(), 400);
    });
}

#[test]
fn api_reads_without_a_clone() {
    let s = Sandbox::new();
    let first = commit(
        &s,
        "api",
        vec![put("a.txt", "one\n"), put("dir/b.txt", "two\n")],
        "first",
    );
    let second = commit(&s, "api", vec![put("a.txt", "three\n")], "second\n\nbody");
    let server = serve(&s, Duration::from_secs(60));
    let api = format!("{}/api/{}/api", server.base, s.account.handle);

    let (status, body, _) = get(&format!("{api}/refs"));
    assert_eq!(status, 200);
    let v: serde_json::Value = serde_json::from_str(&body).unwrap();
    assert_eq!(v["refs"][0]["target"], second);
    assert_eq!(v["head"], "refs/heads/main");

    let (_, body, headers) = get(&format!("{api}/tree?rev={second}"));
    let v: serde_json::Value = serde_json::from_str(&body).unwrap();
    let names: Vec<&str> = v["entries"]
        .as_array()
        .unwrap()
        .iter()
        .map(|e| e["name"].as_str().unwrap())
        .collect();
    assert_eq!(names, vec!["a.txt", "dir"]);
    assert!(
        headers["cache-control"]
            .to_str()
            .unwrap()
            .contains("immutable")
    );
    let (_, body, _) = get(&format!("{api}/tree?path=/dir/"));
    assert!(body.contains("b.txt"));

    let (status, body, headers) = get(&format!("{api}/blob?path=a.txt"));
    assert_eq!((status, body.as_str()), (200, "three\n"));
    assert!(headers.contains_key("etag"));
    let (_, body, _) = get(&format!("{api}/blob?path=a.txt&rev={first}"));
    assert_eq!(body, "one\n");

    let (_, body, _) = get(&format!("{api}/log?limit=10"));
    let v: serde_json::Value = serde_json::from_str(&body).unwrap();
    assert_eq!(v["commits"].as_array().unwrap().len(), 2);
    assert_eq!(v["commits"][0]["message"], "second\n\nbody");
    assert_eq!(v["commits"][0]["parents"][0], first);
    let (_, body, _) = get(&format!("{api}/log?path=dir"));
    let v: serde_json::Value = serde_json::from_str(&body).unwrap();
    assert_eq!(v["commits"].as_array().unwrap().len(), 1);

    let (_, body, _) = get(&format!("{api}/diff?base={first}&head={second}"));
    assert!(body.contains("-one") && body.contains("+three"), "{body}");

    let (status, body, headers) = get(&format!("{api}/object/{second}"));
    assert_eq!(status, 200);
    assert!(body.starts_with("tree "));
    assert_eq!(headers["x-git-type"], "commit");

    for (path, want) in [
        ("/blob?rev=--output=x", 400),
        ("/blob?path=nope", 404),
        ("/object/xyz", 400),
        (&format!("/object/{}", "0".repeat(40)), 404),
    ] {
        assert_eq!(get(&format!("{api}{path}")).0, want, "{path}");
    }
    let (status, _, _) = get(&format!(
        "{}/api/{}/missing/refs",
        server.base, s.account.handle
    ));
    assert_eq!(status, 404);
    let (status, _, _) = get(&format!(
        "{}/api/{}/bad%20name/refs",
        server.base, s.account.handle
    ));
    assert_eq!(status, 400);
    let (status, _, _) = get(&format!("{}/api/nobody.invalid/x/refs", server.base));
    assert_eq!(status, 404);
}

#[test]
fn ttl_and_firehose_freshness() {
    let s = Sandbox::new();
    commit(&s, "fresh", vec![put("a", "1")], "one");
    let server = serve(&s, Duration::from_secs(3600));
    let blob = format!("{}/api/{}/fresh/blob?path=a", server.base, s.account.handle);
    assert_eq!(get(&blob).1, "1");
    // Within the TTL and unwatched, the mirror is trusted as is.
    commit(&s, "fresh", vec![put("a", "2")], "two");
    assert_eq!(get(&blob).1, "1");

    // The firehose marks it stale and refreshes it.
    let ws = format!(
        "{}/xrpc/com.atproto.sync.subscribeRepos",
        net().pds.replace("http", "ws")
    );
    server
        .rt
        .spawn(knotless_serve::firehose::watch(server.app.clone(), ws));
    wait_until(|| server.app.watching.load(Ordering::Relaxed));
    commit(&s, "fresh", vec![put("a", "3")], "three");
    wait_until(|| get(&blob).1 == "3");
    // Watched and quiet: served without asking the PDS.
    assert_eq!(get(&blob).1, "3");

    // A firehose that cannot connect leaves the TTL in charge.
    let lost = App::new(ctx(&s), Duration::ZERO);
    lost.watching.store(true, Ordering::Relaxed);
    server.rt.spawn(knotless_serve::firehose::watch(
        lost.clone(),
        "ws://127.0.0.1:9".into(),
    ));
    wait_until(|| !lost.watching.load(Ordering::Relaxed));
}

#[test]
fn binary_lifecycle() {
    let s = Sandbox::new();
    let exe = env!("CARGO_BIN_EXE_knotless-serve");
    let out = s.command(exe, s.dir.path()).arg("--nope").output().unwrap();
    assert_eq!(out.status.code(), Some(2));
    let out = s
        .command(exe, s.dir.path())
        .args(["--listen", "not-an-address"])
        .output()
        .unwrap();
    assert_eq!(out.status.code(), Some(1));
    let out = s
        .command(exe, s.dir.path())
        .args(["--public-url", "ftp://nope"])
        .output()
        .unwrap();
    assert_eq!(out.status.code(), Some(1));
    // `--private` is gone, and says what replaced it.
    let out = s
        .command(exe, s.dir.path())
        .args(["--private", "--listen", "127.0.0.1:0"])
        .output()
        .unwrap();
    assert_eq!(out.status.code(), Some(2));
    let said = String::from_utf8_lossy(&out.stderr);
    assert!(said.contains(knotless_serve::PRIVATE_GONE), "{said}");
    // Bots' logins are where bots reach it.
    let out = s
        .command(exe, s.dir.path())
        .args(["--bots", "bots"])
        .output()
        .unwrap();
    assert_eq!(out.status.code(), Some(1));
    assert!(String::from_utf8_lossy(&out.stderr).contains("--public-url"));
    // So are a site's owners sent from its root.
    let out = s
        .command(exe, s.dir.path())
        .args(["--pages-domain", "knotless.page"])
        .output()
        .unwrap();
    assert_eq!(out.status.code(), Some(1));
    assert!(String::from_utf8_lossy(&out.stderr).contains("--pages-domain takes --public-url"));
    // Apple's push key takes its id and team, and has to be one.
    let apns = |args: &[&str]| {
        let out = s.command(exe, s.dir.path()).args(args).output().unwrap();
        assert_eq!(out.status.code(), Some(1));
        String::from_utf8_lossy(&out.stderr).to_string()
    };
    let key = s.path("apns.p8");
    let key = key.to_string_lossy();
    assert!(apns(&["--apns-key", &key]).contains("--apns-key-id and --apns-team"));
    let named = [
        "--apns-key",
        &*key,
        "--apns-key-id",
        "K",
        "--apns-team",
        "T",
    ];
    assert!(apns(&named).contains("reading"));
    std::fs::write(s.path("apns.p8"), "not a key").unwrap();
    assert!(apns(&named).contains("apns.p8"));
    let signing = p256::ecdsa::SigningKey::from_bytes((&[7u8; 32]).into()).unwrap();
    let pem = p256::pkcs8::EncodePrivateKey::to_pkcs8_pem(&signing, Default::default()).unwrap();
    std::fs::write(s.path("apns.p8"), pem.as_bytes()).unwrap();

    std::fs::create_dir_all(s.path("releases")).unwrap();
    std::fs::write(s.path("releases/latest"), "0.1.0\n").unwrap();
    let mut child = s
        .command(exe, s.dir.path())
        .args([
            "--listen",
            "127.0.0.1:0",
            "--ttl",
            "1",
            "--firehose",
            "ws://127.0.0.1:9",
            "--jetstream",
            "ws://127.0.0.1:9",
            "--public-url",
            "http://localhost:1",
            "--oauth-url",
            "https://gateway.example/",
            "--index",
            &s.path("index.db").to_string_lossy(),
            "--backfill",
            "http://127.0.0.1:9",
            "--relay",
            "http://127.0.0.1:9",
            "--labeler",
            "did:web:127.0.0.1%3A9",
            "--web",
            "http://127.0.0.1:9",
            "--releases",
            &s.path("releases").to_string_lossy(),
            "--metrics",
            "127.0.0.1:0",
            "--usage",
            &s.path("usage.db").to_string_lossy(),
            "--logs",
            &s.path("logs.db").to_string_lossy(),
            "--logs-keep",
            "100000",
            "--code-index",
            &s.path("code").to_string_lossy(),
            "--pages-domain",
            "knotless.page",
            "--apns-key",
            &key,
            "--apns-key-id",
            "K",
            "--apns-team",
            "T",
            "--client-minimum",
            "0.1.0",
            "--client-minimum-reason",
            "Compaction lost objects.",
            "--ios-minimum-build",
            "7",
            "--ios-minimum-reason",
            "Older builds lose drafts.",
        ])
        .stdout(std::process::Stdio::piped())
        .stderr(std::process::Stdio::piped())
        .spawn()
        .unwrap();
    // The sweep at startup, told of on stderr, which is read to its end.
    let (swept, sweeping) = std::sync::mpsc::channel();
    let stderr = std::io::BufReader::new(child.stderr.take().unwrap());
    std::thread::spawn(move || {
        for line in std::io::BufRead::lines(stderr).map_while(Result::ok) {
            if line.contains("index sweep ended") {
                let _ = swept.send(());
            }
        }
    });
    let mut out = std::io::BufReader::new(child.stdout.take().unwrap());
    let mut said = |prefix: &str| {
        let mut line = String::new();
        std::io::BufRead::read_line(&mut out, &mut line).unwrap();
        line.trim().strip_prefix(prefix).unwrap().to_string()
    };
    let base = said("listening on ");
    let metrics = said("metrics on ");
    assert_eq!(get(&format!("{base}/healthz")).1, "ok");
    // The oldest client it asks for, and why.
    let described = get(&format!("{base}/xrpc/dev.knotless.describeServer")).1;
    let described: serde_json::Value = serde_json::from_str(&described).unwrap();
    assert_eq!(
        described["client"],
        serde_json::json!({"minimum": "0.1.0", "reason": "Compaction lost objects."})
    );
    assert_eq!(
        described["ios"],
        serde_json::json!({"minimumBuild": 7, "reason": "Older builds lose drafts."})
    );
    // Each request is counted, by its route, and given its trace.
    assert_eq!(get(&format!("{base}/healthz")).2["x-trace-id"].len(), 32);
    // Or the trace a traceparent names, as the pages hand theirs on.
    let id = "4bf92f3577b34da6a3ce929d0e0e4736";
    let asked = reqwest::blocking::Client::new()
        .get(format!("{base}/healthz"))
        .header("traceparent", format!("00-{id}-00f067aa0ba902b7-01"))
        .send()
        .unwrap();
    assert_eq!(asked.headers()["x-trace-id"], id);
    let counted = get(&format!("{metrics}/metrics")).1;
    assert!(counted.contains("http_route=\"/healthz\""), "{counted}");
    let http = reqwest::blocking::Client::new();
    let set = |level: &str| {
        let res = http.put(format!("{metrics}/log")).body(level.to_string());
        res.send().unwrap().status().as_u16()
    };
    assert_eq!(set("info,knotless::outgoing=debug"), 200);
    assert_eq!(set("[[["), 400);
    assert!(
        get(&format!("{base}/.well-known/did.json"))
            .1
            .contains("did:web:localhost%3A1")
    );
    assert!(
        get(&format!("{base}/install.sh"))
            .1
            .contains("http://localhost:1/releases")
    );
    // The script and the releases are answered here, even what is not
    // there, rather than handed on to the pages.
    assert_eq!(get(&format!("{base}/releases/latest")).1, "0.1.0\n");
    assert_eq!(get(&format!("{base}/releases/nope")).0, 404);
    assert!(
        get(&format!("{base}/.well-known/oauth-protected-resource/mcp"))
            .1
            .contains("http://localhost:1/mcp")
    );
    // The pages are handed on, here to nothing.
    assert_eq!(get(&format!("{base}/someone.test")).0, 502);
    // Stopped only once the sweep has finished, which it may not have
    // yet on a slow machine.
    sweeping.recv_timeout(Duration::from_secs(30)).unwrap();
    std::process::Command::new("kill")
        .args(["-TERM", &child.id().to_string()])
        .status()
        .unwrap();
    assert!(child.wait().unwrap().success());
    // Each line it logged was kept as it went to stderr, the last as it
    // stopped; the requests with their traces.
    let db = rusqlite::Connection::open(s.path("logs.db")).unwrap();
    let kept = |sql: &str| db.query_row(sql, [], |r| r.get::<_, i64>(0)).unwrap();
    assert_eq!(
        kept("SELECT count(*) FROM line WHERE message = 'stopped'"),
        1
    );
    let traced = "SELECT count(*) FROM line WHERE message = 'request' AND length(trace) = 32";
    assert!(kept(traced) >= 1);
    // Each request was counted, under the route it matched, and written as
    // serve stopped; the health check was not.
    let db = rusqlite::Connection::open(s.path("usage.db")).unwrap();
    let mut stmt = db
        .prepare(
            "SELECT surface, route, class, sum(n) FROM request GROUP BY 1, 2, 3 ORDER BY 1, 2, 3",
        )
        .unwrap();
    let counted: Vec<(String, String, i64, i64)> = stmt
        .query_map([], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?)))
        .unwrap()
        .map(Result::unwrap)
        .collect();
    let row = |surface: &str, route: &str, class, n| (surface.into(), route.into(), class, n);
    assert_eq!(
        counted,
        [
            row("install", "/install.sh", 2, 1),
            row("install", "/releases/*", 2, 1),
            row("install", "/releases/*", 4, 1),
            row("mcp", "/.well-known/oauth-protected-resource/mcp", 2, 1),
            row("other", "/.well-known/did.json", 2, 1),
            row("pages", "*", 5, 1),
            row("xrpc", "/xrpc/dev.knotless.describeServer", 2, 1),
        ]
    );
}

#[test]
fn an_index_is_checked_before_a_rolling_deploy() {
    let dir = tempfile::tempdir().unwrap();
    let exe = env!("CARGO_BIN_EXE_knotless-serve");
    let path = dir.path().join("index.db");
    let check = || {
        std::process::Command::new(exe)
            .arg("--check-index")
            .arg(&path)
            .output()
            .unwrap()
    };
    // None yet: nothing to rebuild, and checking makes none.
    assert!(check().status.success());
    assert!(!path.exists());
    knotless_serve::index::Index::open(&path).unwrap();
    let out = check();
    assert!(out.status.success());
    assert!(String::from_utf8_lossy(&out.stdout).contains("keeps it"));
    // Of other tables, starting would rebuild it, and checking leaves it.
    rusqlite::Connection::open(&path)
        .unwrap()
        .execute_batch("PRAGMA user_version = 1")
        .unwrap();
    let out = check();
    assert_eq!(out.status.code(), Some(1));
    assert!(String::from_utf8_lossy(&out.stderr).contains("would rebuild it"));
    assert_eq!(knotless_serve::index::version_at(&path).unwrap(), Some(1));
}

#[test]
fn arguments() {
    use knotless_serve::parse_args;
    let args = |a: &[&str]| {
        parse_args(
            &std::iter::once("x")
                .chain(a.iter().copied())
                .map(String::from)
                .collect::<Vec<_>>(),
        )
    };
    let o = args(&[
        "--listen",
        "0.0.0.0:1",
        "--ttl",
        "9",
        "--firehose",
        "wss://r",
        "--jetstream",
        "wss://jetstream.us-east.bsky.network",
    ])
    .unwrap();
    assert_eq!(o.listen, "0.0.0.0:1");
    assert_eq!(o.ttl, Duration::from_secs(9));
    assert_eq!(o.firehose.as_deref(), Some("wss://r"));
    assert_eq!(
        o.jetstream.as_deref(),
        Some("wss://jetstream.us-east.bsky.network")
    );
    assert_eq!(o.public_url, None);
    // Empty, a stream or backfill is off.
    let o = args(&["--firehose", "", "--jetstream", "", "--backfill", ""]).unwrap();
    assert_eq!((o.firehose, o.jetstream, o.backfill), (None, None, None));
    let o = args(&["--backfill", "https://r"]).unwrap();
    assert_eq!(o.backfill.as_deref(), Some("https://r"));
    // Directories in turn, and a mirror; empty, none.
    let o = args(&[
        "--directory",
        "https://a",
        "--directory",
        "",
        "--directory",
        "https://b",
        "--mirror",
        "https://m",
    ])
    .unwrap();
    assert_eq!(o.sources.directories, ["https://a", "https://b"]);
    assert_eq!(o.sources.mirror.as_deref(), Some("https://m"));
    assert_eq!(args(&["--mirror", ""]).unwrap().sources.mirror, None);
    let gone = args(&["--private"]).err();
    assert_eq!(gone.as_deref(), Some(knotless_serve::PRIVATE_GONE));
    let o = args(&["--public-url", "https://knot.example"]).unwrap();
    assert_eq!(o.public_url.as_deref(), Some("https://knot.example"));
    let o = args(&["--labeler", "did:web:a", "--labeler", "did:web:b"]).unwrap();
    assert_eq!(o.labelers, ["did:web:a", "did:web:b"]);
    let o = args(&["--default-runner", "did:plc:r"]).unwrap();
    assert_eq!(o.default_runner.as_deref(), Some("did:plc:r"));
    assert_eq!(
        args(&["--default-runner", ""]).unwrap().default_runner,
        None
    );
    assert_eq!(
        args(&[
            "--default-runner",
            "did:plc:r",
            "--default-runner",
            "did:plc:s"
        ])
        .unwrap_err(),
        "--default-runner names one runner"
    );
    assert_eq!(
        args(&["--default-runner", "runner.example"]).unwrap_err(),
        "--default-runner takes a DID, not runner.example"
    );
    assert_eq!(o.web, None);
    // The oldest client asked for is a version; empty, none is.
    let o = args(&["--client-minimum", "0.3.0", "--client-minimum-reason", ""]).unwrap();
    assert_eq!(o.client_minimum.as_deref(), Some("0.3.0"));
    assert_eq!(o.client_minimum_reason, None);
    let o = args(&["--client-minimum", "", "--client-minimum-reason", "Why"]).unwrap();
    assert_eq!(o.client_minimum, None);
    assert_eq!(o.client_minimum_reason.as_deref(), Some("Why"));
    assert!(args(&["--client-minimum", "soon"]).is_err());
    let o = args(&["--ios-minimum-build", "5", "--ios-minimum-reason", ""]).unwrap();
    assert_eq!(o.ios_minimum_build, Some(5));
    assert_eq!(o.ios_minimum_reason, None);
    let o = args(&["--ios-minimum-build", "", "--ios-minimum-reason", "Why"]).unwrap();
    assert_eq!(o.ios_minimum_build, None);
    assert_eq!(o.ios_minimum_reason.as_deref(), Some("Why"));
    let refused = args(&["--ios-minimum-build", "soon"]).err();
    assert_eq!(
        refused.as_deref(),
        Some("--ios-minimum-build takes a build number")
    );
    let o = args(&["--web", "http://127.0.0.1:2591"]).unwrap();
    assert_eq!(o.web.as_deref(), Some("http://127.0.0.1:2591"));
    assert_eq!(o.releases, None);
    let o = args(&["--releases", "/var/lib/knotless/releases"]).unwrap();
    let dir = knotless_serve::releases::Source::Dir("/var/lib/knotless/releases".into());
    assert_eq!(o.releases, Some(dir));
    // Or a repository's releases.
    let o = args(&["--releases", "knotless://alice.test/knotless"]).unwrap();
    let url = knotless::url::RepoUrl::parse("knotless://alice.test/knotless").unwrap();
    assert_eq!(
        o.releases,
        Some(knotless_serve::releases::Source::Repo(url))
    );
    assert!(args(&["--releases", "knotless://alice.test"]).is_err());
    assert_eq!(o.stale, None);
    let o = args(&["--stale", "300"]).unwrap();
    assert_eq!(o.stale, Some(Duration::from_secs(300)));
    assert!(args(&["--stale", "x"]).is_err());
    assert_eq!(
        (o.logs.as_deref(), o.logs_keep),
        (None, knotless_serve::logs::KEEP)
    );
    let o2 = args(&["--logs", "/data/logs.db", "--logs-keep", "9"]).unwrap();
    assert_eq!(
        (o2.logs.as_deref(), o2.logs_keep),
        (Some(std::path::Path::new("/data/logs.db")), 9)
    );
    assert!(args(&["--logs-keep", "x"]).is_err());
    assert_eq!(o.usage, None);
    let o = args(&["--usage", "/var/lib/knotless/usage.db"]).unwrap();
    assert_eq!(
        o.usage.as_deref(),
        Some(std::path::Path::new("/var/lib/knotless/usage.db"))
    );
    assert!(args(&["--ttl", "x"]).is_err());
    assert!(args(&["--ttl"]).is_err());
    assert_eq!(args(&[]).unwrap().metrics, None);
    let o = args(&["--metrics", "127.0.0.1:9464"]).unwrap();
    assert_eq!(o.metrics.as_deref(), Some("127.0.0.1:9464"));
    assert_eq!(o.bots, None);
    let o = args(&["--bots", "/var/lib/knotless/bots"]).unwrap();
    assert_eq!(
        o.bots.as_deref(),
        Some(std::path::Path::new("/var/lib/knotless/bots"))
    );
    assert_eq!(o.sign_up_pds, "https://selfhosted.social");
    let o = args(&["--sign-up-pds", "https://pds.example"]).unwrap();
    assert_eq!(o.sign_up_pds, "https://pds.example");
    assert!(args(&["--sign-up-pds"]).is_err());
    assert_eq!(o.oauth_url, None);
    assert_eq!(o.pages_domain, None);
    let o = args(&["--pages-domain", "knotless.page"]).unwrap();
    assert_eq!(o.pages_domain.as_deref(), Some("knotless.page"));
    assert_eq!(o.apns_key, None);
    let o = args(&[
        "--apns-key",
        "/data/apns.p8",
        "--apns-key-id",
        "K",
        "--apns-team",
        "T",
    ])
    .unwrap();
    assert_eq!(
        o.apns_key.as_deref(),
        Some(std::path::Path::new("/data/apns.p8"))
    );
    assert_eq!(o.apns_key_id.as_deref(), Some("K"));
    assert_eq!(o.apns_team.as_deref(), Some("T"));
    let o = args(&["--apns-key", "", "--apns-key-id", "", "--apns-team", ""]).unwrap();
    assert_eq!((o.apns_key, o.apns_key_id, o.apns_team), (None, None, None));
    let o = args(&["--oauth-url", "https://gateway.example"]).unwrap();
    assert_eq!(o.oauth_url.as_deref(), Some("https://gateway.example"));
}

#[test]
fn firehose_frames_of_every_kind() {
    use futures::SinkExt;
    use tokio_tungstenite::tungstenite::Message;
    let s = Sandbox::new();
    let rt = tokio::runtime::Runtime::new().unwrap();
    let listener = rt
        .block_on(tokio::net::TcpListener::bind("127.0.0.1:0"))
        .unwrap();
    let addr = listener.local_addr().unwrap();
    rt.spawn(async move {
        let (tcp, _) = listener.accept().await.unwrap();
        let mut ws = tokio_tungstenite::accept_async(tcp).await.unwrap();
        let value = |pairs: Vec<(&str, ciborium::Value)>| {
            ciborium::Value::Map(
                pairs
                    .into_iter()
                    .map(|(k, v)| (ciborium::Value::Text(k.into()), v))
                    .collect(),
            )
        };
        let mut frame = Vec::new();
        ciborium::into_writer(
            &value(vec![("t", ciborium::Value::Text("#identity".into()))]),
            &mut frame,
        )
        .unwrap();
        ciborium::into_writer(
            &value(vec![
                ("did", ciborium::Value::Text("did:plc:watched".into())),
                ("seq", ciborium::Value::Integer(5.into())),
            ]),
            &mut frame,
        )
        .unwrap();
        ws.send(Message::Text("hello".into())).await.unwrap();
        ws.send(Message::Binary(vec![0xff].into())).await.unwrap();
        ws.send(Message::Binary(frame.into())).await.unwrap();
        ws.close(None).await.unwrap();
    });
    let app = App::new(ctx(&s), Duration::ZERO);
    rt.spawn(knotless_serve::firehose::watch(
        app.clone(),
        format!("ws://{addr}"),
    ));
    wait_until(|| app.events_for("did:plc:watched") == 1);
    wait_until(|| !app.watching.load(Ordering::Relaxed));
}

#[test]
fn a_cache_it_cannot_write_is_a_server_error() {
    let mut s = Sandbox::new();
    commit(&s, "nocache", vec![put("a", "a")], "c");
    std::fs::write(s.path("blocked"), b"").unwrap();
    let blocked = s.path("blocked").join("cache");
    s.set("KNOTLESS_CACHE_DIR", blocked.to_str().unwrap());
    let server = serve(&s, Duration::ZERO);
    let (status, _, _) = get(&format!(
        "{}/api/{}/nocache/refs",
        server.base, s.account.handle
    ));
    assert_eq!(status, 500);
}