chadtmiller.com/knotless

//! 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, 502);
}

#[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 repositories are served to this machine alone.
    let out = s
        .command(exe, s.dir.path())
        .args(["--private", "--listen", "0.0.0.0:0"])
        .output()
        .unwrap();
    assert_eq!(out.status.code(), Some(1));
    assert!(String::from_utf8_lossy(&out.stderr).contains("loopback"));

    let mut child = s
        .command(exe, s.dir.path())
        .args([
            "--listen",
            "127.0.0.1:0",
            "--private",
            "--ttl",
            "1",
            "--firehose",
            "ws://127.0.0.1:9",
            "--jetstream",
            "ws://127.0.0.1:9",
            "--public-url",
            "http://localhost:1",
            "--index",
            &s.path("index.db").to_string_lossy(),
            "--backfill",
            "http://127.0.0.1:9",
            "--labeler",
            "did:web:127.0.0.1%3A9",
            "--web",
            "http://127.0.0.1:9",
        ])
        .stdout(std::process::Stdio::piped())
        .spawn()
        .unwrap();
    let mut line = String::new();
    std::io::BufRead::read_line(
        &mut std::io::BufReader::new(child.stdout.take().unwrap()),
        &mut line,
    )
    .unwrap();
    let base = line
        .trim()
        .strip_prefix("listening on ")
        .unwrap()
        .to_string();
    assert_eq!(get(&format!("{base}/healthz")).1, "ok");
    assert!(
        get(&format!("{base}/.well-known/did.json"))
            .1
            .contains("did:web:localhost%3A1")
    );
    // The pages are handed on, here to nothing.
    assert_eq!(get(&format!("{base}/someone.test")).0, 502);
    std::process::Command::new("kill")
        .args(["-TERM", &child.id().to_string()])
        .status()
        .unwrap();
    assert!(child.wait().unwrap().success());
}

#[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);
    assert!(!o.private);
    assert!(args(&["--private"]).unwrap().private);
    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"]);
    assert_eq!(o.web, None);
    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!(args(&["--ttl", "x"]).is_err());
    assert!(args(&["--ttl"]).is_err());
}

#[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);
}