//! Forks, pulls, issues and reviews, with an appview that follows the
//! reference PDS's firehose.

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

use futures::{SinkExt, StreamExt};
use knotless_serve::gateway::Gateway;
use knotless_serve::index::Index;
use knotless_serve::{App, router};
use serde_json::Value;
use std::sync::Arc;
use std::time::Duration;

struct Appview {
    base: String,
    did: String,
    app: Arc<App>,
    rt: tokio::runtime::Runtime,
    _proxy: TlsProxy,
    _sandbox: Sandbox,
}

/// An appview that is a gateway too, following the PDS.
fn appview() -> Appview {
    let mut s = Sandbox::new();
    s.set("KNOTLESS_PASSWORD", "");
    s.set("KNOTLESS_DID_WEB_SCHEME", "http");
    s.set("KNOTLESS_ADMIN_PASSWORD", "hunter2");
    let rt = tokio::runtime::Runtime::new().unwrap();
    let listener = rt
        .block_on(tokio::net::TcpListener::bind("127.0.0.1:0"))
        .unwrap();
    let port = listener.local_addr().unwrap().port();
    let base = format!("http://localhost:{port}");
    let mut gw = Gateway::new(&base).unwrap();
    // Authorization servers reach a gateway, an OAuth client, over https.
    let proxy = tls_proxy(port);
    gw.oauth_url = proxy.url.clone();
    let did = gw.did.clone();
    let index = Index::open(&s.path("index.db")).unwrap();
    let app = App::build(ctx(&s), Duration::ZERO, Some(gw), Some(index), false);
    let router = router(app.clone());
    rt.spawn(async move { axum::serve(listener, router).await.unwrap() });
    let ws = net().pds.replace("http://", "ws://");
    rt.spawn(knotless_serve::firehose::watch(app.clone(), ws));
    Appview {
        base,
        did,
        app,
        rt,
        _proxy: proxy,
        _sandbox: s,
    }
}

fn person(account: Account, appview: &Appview) -> Sandbox {
    let mut s = Sandbox::with(account);
    s.set("KNOTLESS_PASSWORD", "");
    s.set("KNOTLESS_DID_WEB_SCHEME", "http");
    s.set("KNOTLESS_APPVIEW", &appview.base);
    s.set("KNOTLESS_BROWSER", &browser(&s.account.clone(), ""));
    let password = s.account.password.clone();
    let handle = s.account.handle.clone();
    assert_eq!(
        s.cli(&["login", &handle, "--password-stdin"], password.as_bytes())
            .0,
        0
    );
    s
}

fn ok(s: &Sandbox, args: &[&str]) -> String {
    let (code, out) = s.cli(args, b"");
    assert_eq!(code, 0, "{args:?}: {out}");
    out
}

fn fails(s: &Sandbox, args: &[&str]) {
    let (code, out) = s.cli(args, b"");
    assert_eq!(code, 1, "{args:?} should fail: {out}");
}

/// Waits until `list` shows `line`.
fn listed(s: &Sandbox, url: &str, line: &str) {
    wait_until(|| s.cli(&["list", url], b"").1.contains(line));
}

#[test]
fn forks_pulls_reviews_and_merges() {
    let av = appview();
    let owner = person(account(), &av);
    let contrib = person(account(), &av);
    let url = owner.url("forkable");
    let work = owner.work("w", 1);
    owner.git(&work, &["push", "-q", &url, "main"]);

    // A fork in the contributor's account, with a branch to offer.
    let fork = ok(&contrib, &["fork", &url]);
    assert_eq!(fork, format!("knotless://{}/forkable", contrib.account.did));
    let clone = contrib.path("clone");
    contrib.git(contrib.dir.path(), &["clone", "-q", &fork, "clone"]);
    contrib.git(&clone, &["checkout", "-q", "-b", "feature"]);
    let feature = contrib.commit_file(&clone, "b", "from the fork\n");
    contrib.git(&clone, &["push", "-q", "origin", "feature"]);

    fails(
        &contrib,
        &["pull", &url, "main", &fork, "no-such-branch", "nope"],
    );
    let pull = ok(
        &contrib,
        &[
            "pull", &url, "main", &fork, "feature", "Add b", "-m", "please",
        ],
    );
    assert!(pull.starts_with(&format!("at://{}/dev.knotless.pull/", contrib.account.did)));
    listed(&owner, &url, &format!("pull\topen\t{pull}\tAdd b"));

    ok(
        &contrib,
        &["comment", &url, &pull, "looks right to me", "--approve"],
    );
    ok(&owner, &["comment", &url, &pull, "thanks"]);
    fails(&owner, &["comment", &url, &pull, "?", "--bogus"]);
    wait_until(|| owner.cli(&["show", &url, &pull], b"").1.contains("thanks"));
    let shown = ok(&owner, &["show", &url, &pull]);
    assert!(shown.starts_with("open pull by"), "{shown}");
    assert!(
        shown.contains("please") && shown.contains("[approve]: looks right"),
        "{shown}"
    );

    // The base has not moved: a fast-forward.
    let merged = ok(&owner, &["merge", &url, &pull]);
    assert_eq!(merged, feature);
    let (_, refs) = owner.cli(&["refs", &url], b"");
    assert!(
        refs.contains(&format!("{feature}\trefs/heads/main")),
        "{refs}"
    );
    listed(&owner, &url, &format!("pull\tmerged\t{pull}"));
    fails(&owner, &["merge", &url, &pull]);

    // The base moved: a merge commit with both parents.
    owner.git(&work, &["pull", "-q", &url, "main"]);
    owner.commit_file(&work, "c", "owner's\n");
    owner.git(&work, &["push", "-q", &url, "main"]);
    contrib.git(&clone, &["checkout", "-q", "-b", "two", &feature]);
    let two = contrib.commit_file(&clone, "d", "more\n");
    contrib.git(&clone, &["push", "-q", "origin", "two"]);
    let second = ok(&contrib, &["pull", &url, "main", &fork, "two", "Add d"]);
    listed(&owner, &url, &second);
    let merge = ok(&owner, &["merge", &url, &second]);
    let check = owner.path("check");
    owner.git(owner.dir.path(), &["clone", "-q", &url, "check"]);
    assert_eq!(owner.git(&check, &["rev-parse", "HEAD"]), merge);
    assert_eq!(owner.git(&check, &["rev-parse", "HEAD^2"]), two);
    assert!(check.join("c").exists() && check.join("d").exists());

    // Both sides changed one file: a conflict, and nothing moves.
    owner.git(&work, &["pull", "-q", &url, "main"]);
    owner.commit_file(&work, "f", "owner's f\n");
    owner.git(&work, &["push", "-q", &url, "main"]);
    contrib.git(&clone, &["checkout", "-q", "-b", "clash", &two]);
    contrib.commit_file(&clone, "f", "contributor's f\n");
    contrib.git(&clone, &["push", "-q", "origin", "clash"]);
    let clash = ok(
        &contrib,
        &["pull", &url, "main", &fork, "clash", "Change f"],
    );
    listed(&owner, &url, &clash);
    let (merged, _, err) = owner.knotless(&["merge", &url, &clash], b"");
    assert!(!merged && err.contains("conflicts in f"), "{err}");

    // Branches can go between opening a pull and merging it.
    contrib.git(&clone, &["push", "-q", "origin", ":clash"]);
    let (merged, _, err) = owner.knotless(&["merge", &url, &clash], b"");
    assert!(!merged && err.ends_with("has no branch clash"), "{err}");
    fails(
        &contrib,
        &["pull", &url, "release", &fork, "two", "Into release"],
    );
    owner.git(&work, &["push", "-q", &url, "main:release"]);
    let release = ok(
        &contrib,
        &["pull", &url, "release", &fork, "two", "Into release"],
    );
    owner.git(&work, &["push", "-q", &url, ":release"]);
    listed(&owner, &url, &release);
    let (merged, _, err) = owner.knotless(&["merge", &url, &release], b"");
    assert!(!merged && err.ends_with("has no branch release"), "{err}");

    // A pull outlives its branches by pointing at others: the target by
    // anyone who may close it, the head by its author alone.
    owner.git(&work, &["push", "-q", &url, "main:stable"]);
    for bad in [
        vec!["retarget", &url, &release],
        vec!["retarget", &url, &release, "--sideways", "x"],
        vec!["retarget", &url, &release, "--base", "nope"],
        vec!["retarget", &url, &release, "--head", "two"],
        vec!["retarget", &url, &pull, "--base", "stable"],
    ] {
        fails(&owner, &bad);
    }
    ok(&owner, &["retarget", &url, &release, "--base", "stable"]);
    fails(&contrib, &["retarget", &url, &clash, "--head", "nope"]);
    ok(&contrib, &["retarget", &url, &clash, "--head", "two"]);
    let issue = ok(&contrib, &["issue", &url, "Not a pull"]);
    fails(&owner, &["retarget", &url, &issue, "--base", "main"]);
    // Once the index has the new target, the pull merges into it.
    wait_until(|| owner.knotless(&["merge", &url, &release], b"").0);
    let check = owner.path("stable");
    owner.git(
        owner.dir.path(),
        &["clone", "-q", "-b", "stable", &url, "stable"],
    );
    owner.git(&check, &["merge-base", "--is-ancestor", &two, "stable"]);

    // A collaborator merges, through the gateway.
    ok(&owner, &["delegate", &owner.account.handle, &av.did]);
    ok(&owner, &["share", &url, &contrib.account.handle, "--write"]);
    contrib.git(&clone, &["fetch", "-q", &url, "main"]);
    contrib.git(&clone, &["checkout", "-q", "-b", "three", &merge]);
    contrib.commit_file(&clone, "e", "collaborating\n");
    contrib.git(&clone, &["push", "-q", "origin", "three"]);
    let third = ok(&contrib, &["pull", &url, "main", &fork, "three", "Add e"]);
    listed(&contrib, &url, &third);
    let sha = ok(&contrib, &["merge", &url, &third]);
    let (_, refs) = owner.cli(&["refs", &url], b"");
    assert!(refs.contains(&format!("{sha}\trefs/heads/main")), "{refs}");
    // And makes and deletes branches the same way, as the owner does.
    assert_eq!(ok(&contrib, &["branch", &url, "topic"]), sha);
    assert_eq!(
        ok(&contrib, &["branch", &url, "topic", "--delete"]),
        "deleted topic"
    );
    assert_eq!(ok(&owner, &["branch", &url, "pinned", &two]), two);
    ok(&owner, &["branch", &url, "pinned", "--delete"]);

    // Someone else's commit, elsewhere on the firehose, changes nothing.
    av.app.observe(knotless_serve::firehose::Event::default());
    assert_eq!(av.app.events_for("did:plc:nobody"), 0);

    // A record gone before the index fetched it is an error to skip.
    let gone = knotless_serve::firehose::Op {
        action: "create".into(),
        collection: "dev.knotless.issue".into(),
        rkey: "3zzzzzzzzzzzz".into(),
        record: None,
    };
    let index = av.app.index.as_ref().unwrap();
    assert!(
        av.rt
            .block_on(knotless_serve::index_op(
                &av.app.ctx,
                index,
                &owner.account.did,
                &gone
            ))
            .is_err()
    );

    // The index knows the fork, and finds repositories by name.
    let found: Value = reqwest::blocking::get(format!(
        "{}/xrpc/dev.knotless.listRepos?q=forkable",
        av.base
    ))
    .unwrap()
    .json()
    .unwrap();
    assert_eq!(found["repos"].as_array().unwrap().len(), 2);
    assert!(found.get("cursor").is_none());
    // A page at a time, the same two.
    let list = |cursor: Option<&str>| -> Value {
        let mut query = vec![("q", "forkable"), ("limit", "1")];
        query.extend(cursor.map(|c| ("cursor", c)));
        reqwest::blocking::Client::new()
            .get(format!("{}/xrpc/dev.knotless.listRepos", av.base))
            .query(&query)
            .send()
            .unwrap()
            .json()
            .unwrap()
    };
    let first = list(None);
    let second = list(first["cursor"].as_str());
    assert_eq!(
        [&first["repos"][0], &second["repos"][0]],
        [&found["repos"][0], &found["repos"][1]]
    );
    assert!(
        list(second["cursor"].as_str())["repos"]
            .as_array()
            .unwrap()
            .is_empty()
    );
    let uri = format!("at://{}/dev.knotless.repo/forkable", owner.account.did);
    let indexed = av.app.index.as_ref().unwrap();
    let key = indexed.repo(&uri).unwrap().unwrap()["key"]
        .as_str()
        .unwrap()
        .to_string();
    let forks = indexed.forks(&owner.account.did, &key);
    assert_eq!(forks.unwrap().len(), 1);

    // A second appview, backfilled, sees the same.
    let s = Sandbox::new();
    let index = Index::open(&s.path("i.db")).unwrap();
    let n = av
        .rt
        .block_on(knotless_serve::backfill(&ctx(&s), &index, &net().pds))
        .unwrap();
    assert!(n >= 10, "{n}");
    let report = av.rt.block_on(knotless_serve::backfilled(
        &ctx(&s),
        &index,
        "http://127.0.0.1:9",
    ));
    assert!(
        report.starts_with("backfilling from http://127.0.0.1:9: "),
        "{report}"
    );
    let report = av
        .rt
        .block_on(knotless_serve::backfilled(&ctx(&s), &index, &net().pds));
    assert!(report.starts_with("backfilled "), "{report}");
    let did = owner.account.did.as_str();
    let backfilled = index.forge(did, &key).unwrap().len();
    wait_until(|| indexed.forge(did, &key).unwrap().len() == backfilled);

    // A deletion takes the record out of the index.
    let pull_rkey = third.rsplit('/').next().unwrap();
    let deleted = knotless_serve::firehose::Op {
        action: "delete".into(),
        collection: "dev.knotless.pull".into(),
        rkey: pull_rkey.into(),
        record: None,
    };
    av.rt
        .block_on(knotless_serve::index_op(
            &ctx(&s),
            &index,
            &contrib.account.did,
            &deleted,
        ))
        .unwrap();
    assert_eq!(index.forge(did, &key).unwrap().len(), backfilled - 1);
}

#[test]
fn pulls_squash_into_a_linear_branch() {
    let av = appview();
    let owner = person(account(), &av);
    let contrib = person(account(), &av);
    let url = owner.url("squashed");
    let work = owner.work("w", 1);
    owner.git(&work, &["push", "-q", &url, "main"]);
    let base = owner.git(&work, &["rev-parse", "HEAD"]);
    let fork = ok(&contrib, &["fork", &url]);
    let clone = contrib.path("clone");
    contrib.git(contrib.dir.path(), &["clone", "-q", &fork, "clone"]);
    contrib.git(&clone, &["checkout", "-q", "-b", "two"]);
    contrib.commit_file(&clone, "a", "one\n");
    contrib.commit_file(&clone, "b", "two\n");
    contrib.git(&clone, &["push", "-q", "origin", "two"]);
    let pull = ok(
        &contrib,
        &[
            "pull",
            &url,
            "main",
            &fork,
            "two",
            "Add a and b",
            "-m",
            "both",
        ],
    );
    listed(&owner, &url, &pull);

    // A fast-forward would bring both commits; protected, it is one.
    ok(&owner, &["protect", &url, "main"]);
    let sha = ok(&owner, &["merge", &url, &pull]);
    let check = owner.path("check");
    owner.git(owner.dir.path(), &["clone", "-q", &url, "check"]);
    assert_eq!(owner.git(&check, &["rev-parse", "HEAD"]), sha);
    assert_eq!(owner.git(&check, &["rev-parse", "HEAD^"]), base);
    assert_eq!(owner.git(&check, &["rev-list", "--count", "HEAD"]), "2");
    assert!(check.join("a").exists() && check.join("b").exists());
    let message = owner.git(&check, &["log", "-1", "--format=%an|%cn|%B"]);
    // The head's author wrote it, and the owner, who merged it, committed it.
    let committer = &owner.account.handle;
    assert_eq!(
        message,
        format!("Test|{committer}|Add a and b\n\nboth\n\n{pull}")
    );
}

#[test]
fn issues_and_who_may_close_them() {
    let av = appview();
    let owner = person(account(), &av);
    let author = person(account(), &av);
    let stranger = person(account(), &av);
    let url = owner.url("tracked");
    ok(&owner, &["create", &url]);
    let issue = ok(
        &author,
        &["issue", &url, "It breaks", "-m", "when I do this"],
    );
    let quiet = ok(&author, &["issue", &url, "Also this"]);
    listed(&owner, &url, &issue);
    listed(&owner, &url, &quiet);
    assert_eq!(ok(&author, &["close", &url, &issue]), "closed");
    listed(&owner, &url, &format!("issue\tclosed\t{issue}"));
    // A stranger's reopening counts for nothing; the owner's does.
    ok(&stranger, &["reopen", &url, &issue]);
    std::thread::sleep(Duration::from_millis(500));
    let listing = ok(&owner, &["list", &url, "--issues"]);
    assert!(
        listing.contains(&format!("issue\tclosed\t{issue}")),
        "{listing}"
    );
    ok(&owner, &["reopen", &url, &issue]);
    listed(&owner, &url, &format!("issue\topen\t{issue}"));
    assert_eq!(ok(&owner, &["list", &url, "--pulls"]), "");
    assert!(ok(&owner, &["show", &url, &quiet]).ends_with(": Also this"));
    fails(&owner, &["list", &url, "--bogus"]);
    fails(&owner, &["merge", &url, &issue]);
    fails(
        &owner,
        &["show", &url, "at://did:plc:nobody/dev.knotless.issue/x"],
    );
    // A comment is about a pull or an issue on this repository.
    fails(
        &owner,
        &[
            "comment",
            &url,
            "at://did:plc:x/dev.knotless.state/3abc",
            "?",
        ],
    );
    fails(
        &owner,
        &[
            "comment",
            &url,
            &format!("{}x", issue.trim_end_matches(|c: char| c.is_alphanumeric())),
            "?",
        ],
    );
    ok(&owner, &["create", &owner.url("elsewhere")]);
    fails(&owner, &["comment", &owner.url("elsewhere"), &issue, "?"]);
    fails(&owner, &["issue", &url, "t", "-x"]);
    fails(&owner, &["issue", &url, "t", "-m"]);
    fails(&owner, &["fork", &owner.url("nothing")]);

    // Without an appview there is nowhere to find them.
    let mut lost = person(owner.account.clone(), &av);
    lost.set("KNOTLESS_APPVIEW", "");
    let (code, out) = lost.cli(&["list", &url], b"");
    assert_eq!(code, 1, "{out}");

    // A server without an index says so.
    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 base = format!("http://{}", listener.local_addr().unwrap());
    let bare = App::new(ctx(&s), Duration::ZERO);
    let router = router(bare.clone());
    rt.spawn(async move { axum::serve(listener, router).await.unwrap() });
    for path in [
        "dev.knotless.getForge?repo=at://did:plc:a/dev.knotless.repo/r&key=k",
        "dev.knotless.listRepos",
        "com.atproto.label.queryLabels?uriPatterns=*",
    ] {
        let res = reqwest::blocking::get(format!("{base}/xrpc/{path}")).unwrap();
        assert_eq!(res.status().as_u16(), 404);
    }
    // Nor can it follow a labeler, and an account is not one: each try
    // fails, and it tries again.
    for (app, did) in [
        (&bare, av.did.clone()),
        (&av.app, owner.account.did.clone()),
    ] {
        let following = knotless_serve::moderation::follow(app.clone(), did);
        let _ =
            rt.block_on(async { tokio::time::timeout(Duration::from_secs(1), following).await });
    }
    // A repository is named by its record's URI.
    let res = reqwest::blocking::get(format!(
        "{}/xrpc/dev.knotless.getForge?repo=x&key=k",
        av.base
    ))
    .unwrap();
    assert_eq!(res.status().as_u16(), 400);

    // Every record comes back, a page at a time.
    let key = owner.record("dev.knotless.repo", "tracked")["key"].clone();
    let repo = format!("at://{}/dev.knotless.repo/tracked", owner.account.did);
    let forge = |cursor: &str| -> Value {
        reqwest::blocking::get(format!(
            "{}/xrpc/dev.knotless.getForge?repo={repo}&key={}&limit=1&cursor={cursor}",
            av.base,
            key.as_str().unwrap()
        ))
        .unwrap()
        .json()
        .unwrap()
    };
    let (mut paged, mut cursor) = (0, String::new());
    loop {
        let page = forge(&cursor);
        assert!(page["entries"].as_array().unwrap().len() <= 1);
        paged += page["entries"].as_array().unwrap().len();
        match page["cursor"].as_str() {
            Some(next) => cursor = next.to_string(),
            None => break,
        }
    }
    assert!(paged > 1, "{paged}");
}

#[test]
fn private_repositories_keep_their_issues_in_the_space() {
    if !net().spaces {
        eprintln!("skipped: the PDS is not on the spaces alpha");
        return;
    }
    let av = appview();
    let owner = person(account(), &av);
    let reader = person(account(), &av);
    let url = owner.url("inner");
    ok(&owner, &["create", &url, "--private"]);
    let work = owner.work("w", 1);
    owner.git(&work, &["push", "-q", &url, "main"]);
    ok(&owner, &["share", &url, &reader.account.handle]);

    let issue = ok(&reader, &["issue", &url, "Private trouble"]);
    owner.git(&work, &["checkout", "-q", "-b", "fix"]);
    let fix = owner.commit_file(&work, "fix", "fixed\n");
    owner.git(&work, &["push", "-q", &url, "fix"]);
    let pull = ok(&owner, &["pull", &url, "main", &url, "fix", "Fix it"]);
    ok(&reader, &["comment", &url, &pull, "ship it", "--approve"]);
    // No appview involved: the members' repos in the space are read directly.
    let listing = ok(&owner, &["list", &url]);
    assert!(
        listing.contains(&format!("issue\topen\t{issue}")),
        "{listing}"
    );
    assert!(
        listing.contains(&format!("pull\topen\t{pull}")),
        "{listing}"
    );
    // Its pulls and issues are in the space, which cannot move.
    fails(&owner, &["transfer", &url, &reader.account.handle]);
    // The owner's gate and blocks hold in the space, with no appview.
    ok(&owner, &["hide", &url, &issue]);
    assert!(!ok(&reader, &["list", &url]).contains(&issue));
    ok(&owner, &["unhide", &url, &issue]);
    ok(&owner, &["block", &reader.account.handle]);
    assert!(!ok(&owner, &["list", &url]).contains(&issue));
    ok(&owner, &["unblock", &reader.account.handle]);
    assert!(ok(&owner, &["list", &url]).contains(&issue));
    // A block from this repository alone is in the space with it.
    ok(&owner, &["block", &reader.account.handle, &url]);
    assert!(!ok(&owner, &["list", &url]).contains(&issue));
    ok(&owner, &["unblock", &reader.account.handle, &url]);
    assert!(ok(&owner, &["list", &url]).contains(&issue));
    assert_eq!(ok(&owner, &["merge", &url, &pull]), fix);
    let listing = ok(&reader, &["list", &url]);
    assert!(
        listing.contains(&format!("pull\tmerged\t{pull}")),
        "{listing}"
    );
    // The appview never saw any of it.
    let uri = format!("at://{}/dev.knotless.repo/inner", owner.account.did);
    assert_eq!(av.app.index.as_ref().unwrap().repo(&uri).unwrap(), None);

    // A private repository forks into a private one.
    let fork = ok(&reader, &["fork", &url, "inner-fork"]);
    let (_, refs) = reader.cli(&["refs", &fork], b"");
    assert!(refs.contains(&format!("{fix}\trefs/heads/main")), "{refs}");
    let anonymous = Sandbox::new();
    assert_eq!(anonymous.cli(&["refs", &fork], b"").0, 1);
    // Its source's owner reads it, to take a pull from it.
    let (code, refs) = owner.cli(&["refs", &fork], b"");
    assert_eq!(code, 0, "{refs}");
    let readers = ok(&reader, &["members", &fork]);
    assert!(readers.contains(owner.account.did.as_str()), "{readers}");
    // A fork of one's own is shared with nobody more.
    let own = ok(&owner, &["fork", &url, "inner-copy"]);
    let readers = ok(&owner, &["members", &own]);
    assert_eq!(readers.lines().count(), 1, "{readers}");
}

#[test]
fn moving_to_another_account() {
    let av = appview();
    let owner = person(account(), &av);
    let heir = person(account(), &av);
    let stranger = person(account(), &av);
    let url = owner.url("moving");
    let work = owner.work("w", 1);
    owner.git(&work, &["push", "-q", &url, "main"]);
    let before = ok(&owner, &["refs", &url]);
    let closed = ok(&stranger, &["issue", &url, "Closed before the move"]);
    let open = ok(&stranger, &["issue", &url, "Still open"]);
    ok(&owner, &["close", &url, &closed]);
    listed(&owner, &url, &format!("issue\tclosed\t{closed}"));

    // Nothing moves until it is offered, and only to whom it is offered.
    fails(&heir, &["accept", &url]);
    fails(&owner, &["transfer", &url, &owner.account.handle]);
    ok(&owner, &["transfer", &url, &heir.account.handle]);
    fails(&stranger, &["accept", &url]);
    ok(&owner, &["transfer", &url, "--cancel"]);
    fails(&heir, &["accept", &url]);
    let elsewhere = format!("knotless://{}/moving", heir.account.handle);
    ok(&owner, &["transfer", &url, &elsewhere]);
    ok(&owner, &["transfer", &url, "--cancel"]);
    fails(
        &owner,
        &["transfer", &owner.url("nothing"), &heir.account.handle],
    );
    let offered = ok(&owner, &["transfer", &url, &heir.account.handle]);
    assert!(offered.contains("knotless accept"), "{offered}");
    let moved = ok(&heir, &["accept", &url]);
    let new = format!("knotless://{}/moving", heir.account.did);
    assert_eq!(moved, format!("moved to {new}"));

    // The old URL leads to the new place, for reads and pushes alike.
    assert_eq!(ok(&heir, &["refs", &url]), before);
    heir.git(heir.dir.path(), &["clone", "-q", &url, "clone"]);
    let clone = heir.path("clone");
    heir.commit_file(&clone, "g", "after the move\n");
    let (pushed, _, err) = heir.git_raw(&clone, &["push", "-q", "origin", "main"]);
    assert!(
        pushed && err.contains(&format!("has moved to {new}")),
        "{err}"
    );
    assert_ne!(ok(&heir, &["refs", &new]), before);

    // Its issues came with it, and what the old owner did before the move
    // still counts; nothing they do after it does.
    listed(&heir, &new, &format!("issue\tclosed\t{closed}"));
    ok(&owner, &["close", &url, &open]);
    std::thread::sleep(Duration::from_secs(1));
    listed(&heir, &new, &format!("issue\topen\t{open}"));
    ok(&heir, &["comment", &new, &open, "moved along"]);
    ok(&heir, &["close", &new, &open]);
    listed(&heir, &new, &format!("issue\tclosed\t{open}"));

    // The old copy can go, leaving a redirect in its place; its old names'
    // redirects lead to the new place too, and others' stay as they were.
    let key = owner.record("dev.knotless.repo", "moving")["key"].clone();
    let at = format!("at://{}/dev.knotless.repo/moving", owner.account.did);
    owner.put_record(
        "dev.knotless.redirect",
        "once-moving",
        serde_json::json!({"$type": "dev.knotless.redirect", "to": {"uri": at, "key": key},
            "createdAt": "2026-10-21T00:00:00.000Z"}),
    );
    ok(&owner, &["create", &owner.url("spare")]);
    ok(&owner, &["rename", &owner.url("spare"), "spare2"]);
    ok(&owner, &["delete", &url, "--yes"]);
    assert!(
        owner.record("dev.knotless.redirect", "once-moving")["to"]["uri"]
            .as_str()
            .unwrap()
            .starts_with(&format!("at://{}/", heir.account.did))
    );
    assert_eq!(
        owner.record("dev.knotless.redirect", "spare")["to"]["uri"],
        format!("at://{}/dev.knotless.repo/spare2", owner.account.did)
    );
    assert!(
        owner.record("dev.knotless.redirect", "moving")["to"]["uri"]
            .as_str()
            .unwrap()
            .starts_with(&format!("at://{}/", heir.account.did))
    );
    assert_eq!(
        ok(&stranger, &["refs", &url]),
        ok(&stranger, &["refs", &new])
    );
    listed(&stranger, &url, &format!("issue\tclosed\t{closed}"));
    let direct = reqwest::blocking::Client::builder()
        .redirect(reqwest::redirect::Policy::none())
        .build()
        .unwrap();
    let old_page = || {
        direct
            .get(format!(
                "{}/_data/{}/moving/issues",
                av.base, owner.account.handle
            ))
            .send()
            .unwrap()
    };
    wait_until(|| old_page().status().as_u16() == 307);
    assert_eq!(
        old_page().headers()["location"],
        format!("/_data/{}/moving/issues", heir.account.did)
    );

    // Saying a repository came from another proves nothing without its
    // owner's word: none of its issues come along.
    let other = owner.url("other");
    owner.git(&work, &["push", "-q", &other, "main"]);
    let theirs = ok(&stranger, &["issue", &other, "Not yours"]);
    listed(&owner, &other, &theirs);
    let key = owner.record("dev.knotless.repo", "other")["key"].clone();
    let from = serde_json::json!({"uri": format!("at://{}/dev.knotless.repo/other", owner.account.did), "key": key});
    stranger.put_record(
        "dev.knotless.repo",
        "claimed",
        serde_json::json!({"$type": "dev.knotless.repo", "key": "3kabcdefghijk", "name": "claimed",
            "transferredFrom": from, "createdAt": "2026-10-21T00:00:00.000Z"}),
    );
    let claimed = stranger.url("claimed");
    std::thread::sleep(Duration::from_secs(1));
    assert!(!ok(&stranger, &["list", &claimed]).contains(&theirs));
}

/// Waits until `list` no longer shows `line`.
fn unlisted(s: &Sandbox, url: &str, line: &str) {
    wait_until(|| {
        let (code, out) = s.cli(&["list", url], b"");
        code == 0 && !out.contains(line)
    });
}

#[test]
fn moderating() {
    let av = appview();
    let owner = person(account(), &av);
    let troll = person(account(), &av);
    let reporter = person(account(), &av);
    let url = owner.url("moderated");
    let work = owner.work("w", 1);
    owner.git(&work, &["push", "-q", &url, "main"]);
    let trolling = ok(&troll, &["issue", &url, "Trolling"]);
    let trolling = trolling.trim();
    let fair = ok(&reporter, &["issue", &url, "A fair point"]);
    let fair = fair.trim();
    let reply = ok(&troll, &["comment", &url, fair, "nonsense"]);
    let reply = reply.trim();
    listed(&owner, &url, trolling);

    // The owner hides an issue, and what was said about it goes with it.
    assert_eq!(ok(&owner, &["hide", &url, fair]), "hidden");
    assert_eq!(ok(&owner, &["hide", &url, fair]), "hidden");
    unlisted(&reporter, &url, fair);
    assert_eq!(reporter.cli(&["show", &url, fair], b"").0, 1);
    ok(&owner, &["hide", &url, reply]);
    assert_eq!(ok(&owner, &["unhide", &url, fair]), "shown");
    listed(&reporter, &url, fair);
    assert!(!ok(&reporter, &["show", &url, fair]).contains("nonsense"));
    ok(&owner, &["unhide", &url, reply]);
    wait_until(|| ok(&reporter, &["show", &url, fair]).contains("nonsense"));
    // Only the owner, and only what is said about a repository.
    fails(&troll, &["hide", &url, fair]);
    let repo_uri = format!("at://{}/dev.knotless.repo/moderated", owner.account.did);
    fails(&owner, &["hide", &url, &repo_uri]);

    // A block keeps out everything the account writes.
    let blocked = ok(&owner, &["block", &troll.account.handle]);
    assert_eq!(blocked, format!("blocked {}", troll.account.did));
    ok(&owner, &["block", &troll.account.handle]);
    unlisted(&reporter, &url, trolling);
    wait_until(|| !ok(&reporter, &["show", &url, fair]).contains("nonsense"));
    fails(&owner, &["block", &owner.account.handle]);
    ok(&owner, &["unblock", &troll.account.handle]);
    listed(&reporter, &url, trolling);
    // Or keeps them out of one repository only, and only its owner can.
    ok(&owner, &["block", &troll.account.handle, &url]);
    ok(&owner, &["block", &troll.account.handle, &url]);
    unlisted(&reporter, &url, trolling);
    // Nor does it show by its URI, however new.
    assert_eq!(reporter.cli(&["show", &url, trolling], b"").0, 1);
    fails(&reporter, &["block", &troll.account.handle, &url]);
    ok(&owner, &["unblock", &troll.account.handle, &url]);
    listed(&reporter, &url, trolling);

    // Reports go to the appview's moderation service, from an OAuth login
    // too, whose scope names the call.
    let handle = reporter.account.handle.clone();
    assert_eq!(reporter.cli(&["login", &handle], b"").0, 0);
    let sent = ok(
        &reporter,
        &["report", trolling, "spam", "-m", "the same thing again"],
    );
    assert_eq!(sent, "report 1 sent");
    ok(&reporter, &["report", &troll.account.handle, "rude"]);
    ok(&reporter, &["report", &url, "other"]);
    fails(&reporter, &["report", &url, "boring"]);
    fails(&reporter, &["report", &format!("{trolling}x"), "spam"]);
    let mut admin = person(account(), &av);
    fails(&admin, &["admin", "reports"]);
    admin.set("KNOTLESS_ADMIN_PASSWORD", "wrong");
    fails(&admin, &["admin", "reports"]);
    admin.set("KNOTLESS_ADMIN_PASSWORD", "hunter2");
    let reports = ok(&admin, &["admin", "reports"]);
    let lines: Vec<&str> = reports.lines().collect();
    assert_eq!(lines.len(), 3, "{reports}");
    assert!(
        lines[0].starts_with("3\t") && lines[0].contains(&repo_uri),
        "{reports}"
    );
    assert!(
        lines[1].contains(&format!("rude\t{}", troll.account.did)),
        "{reports}"
    );
    assert_eq!(
        lines[2],
        format!(
            "1\t{}\tspam\t{trolling}\tthe same thing again",
            reporter.account.did
        )
    );
    assert_eq!(ok(&admin, &["admin", "reports", "1"]).lines().count(), 1);
    fails(&admin, &["admin", "reports", "x"]);
    fails(&admin, &["admin", "nothing"]);
    // Without an appview, reports go where KNOTLESS_MODERATOR says.
    let mut direct = person(reporter.account.clone(), &av);
    direct.set("KNOTLESS_APPVIEW", "");
    fails(&direct, &["report", trolling, "spam"]);
    direct.set("KNOTLESS_MODERATOR", &owner.account.did);
    fails(&direct, &["report", trolling, "spam"]);
    fails(&reporter, &["report", &owner.url("nothing"), "spam"]);
    direct.set("KNOTLESS_MODERATOR", &av.did);
    assert_eq!(ok(&direct, &["report", trolling, "spam"]), "report 4 sent");

    // A second appview follows this one's labels.
    let other = appview();
    other.rt.spawn(knotless_serve::moderation::follow(
        other.app.clone(),
        av.did.clone(),
    ));
    let down_there = |subject: &str| other.app.index.as_ref().unwrap().down(&[subject]).unwrap();

    // The operator takes things down.
    assert_eq!(ok(&admin, &["admin", "status", trolling]), "up");
    assert_eq!(ok(&admin, &["admin", "takedown", trolling]), "taken down");
    assert_eq!(ok(&admin, &["admin", "status", trolling]), "taken down");
    unlisted(&reporter, &url, trolling);
    wait_until(|| down_there(trolling));
    fails(&admin, &["admin", "takedown", trolling, "--later"]);
    assert_eq!(
        ok(&admin, &["admin", "takedown", trolling, "--undo"]),
        "restored"
    );
    listed(&reporter, &url, trolling);
    wait_until(|| !down_there(trolling));

    // An account taken down takes its repositories with it.
    let page = || {
        reqwest::blocking::get(format!(
            "{}/_data/{}/moderated",
            av.base, owner.account.handle
        ))
        .unwrap()
        .status()
        .as_u16()
    };
    assert_eq!(page(), 200);
    ok(&admin, &["admin", "takedown", &owner.account.handle]);
    assert_eq!(
        ok(&admin, &["admin", "status", &owner.account.handle]),
        "taken down"
    );
    assert_eq!(page(), 404);
    let listing = reqwest::blocking::get(format!("{}/xrpc/dev.knotless.listRepos", av.base))
        .unwrap()
        .text()
        .unwrap();
    assert!(!listing.contains(&owner.account.did), "{listing}");
    let labels: Value = reqwest::blocking::get(format!(
        "{}/xrpc/com.atproto.label.queryLabels?uriPatterns=did:*&uriPatterns={}",
        av.base, trolling
    ))
    .unwrap()
    .json()
    .unwrap();
    assert_eq!(labels["labels"][0]["uri"], owner.account.did.as_str());
    assert_eq!(labels["labels"][0]["val"], "!takedown");
    assert_eq!(labels["labels"].as_array().unwrap().len(), 1);
    wait_until(|| down_there(&owner.account.did));
    ok(&admin, &["admin", "takedown", &url]);
    assert_eq!(ok(&admin, &["admin", "status", &url]), "taken down");
    ok(
        &admin,
        &["admin", "takedown", &owner.account.handle, "--undo"],
    );
    assert_eq!(page(), 404);
    ok(&admin, &["admin", "takedown", &url, "--undo"]);
    assert_eq!(page(), 200);

    // The calls themselves, by hand.
    let client = reqwest::blocking::Client::new();
    let xrpc = |nsid: &str| format!("{}/xrpc/{nsid}", av.base);
    let post = |nsid: &str, auth: Option<&str>, body: Value| {
        let mut req = client.post(xrpc(nsid)).json(&body);
        if let Some(auth) = auth {
            req = req.header("authorization", auth);
        }
        req.send().unwrap().status().as_u16()
    };
    let basic = "Basic YWRtaW46aHVudGVyMg==";
    let status = "com.atproto.admin.updateSubjectStatus";
    assert_eq!(post(status, None, serde_json::json!({})), 401);
    assert_eq!(
        post(status, Some(basic), serde_json::json!({"subject": {}})),
        400
    );
    let account =
        serde_json::json!({"$type": "com.atproto.admin.defs#repoRef", "did": "did:plc:x"});
    assert_eq!(
        post(status, Some(basic), serde_json::json!({"subject": account})),
        200
    );
    let get_status = |q: &str| {
        client
            .get(xrpc(&format!("com.atproto.admin.getSubjectStatus{q}")))
            .header("authorization", basic)
            .send()
            .unwrap()
            .status()
            .as_u16()
    };
    assert_eq!(get_status(""), 400);
    assert_eq!(get_status("?did=did:plc:x"), 200);
    let report = "com.atproto.moderation.createReport";
    assert_eq!(post(report, None, serde_json::json!({})), 401);
    let token = |ctx: &knotless::env::Ctx| {
        av.rt.block_on(async {
            let me = knotless::session::me(ctx).await.unwrap();
            knotless::auth::service_token(ctx, &me, &av.did, report)
                .await
                .unwrap()
        })
    };
    let bearer = format!("Bearer {}", token(&ctx(&reporter)));
    assert_eq!(
        post(
            report,
            Some(&bearer),
            serde_json::json!({"subject": account})
        ),
        400
    );
    let bearer = format!("Bearer {}", token(&ctx(&reporter)));
    assert_eq!(
        post(report, Some(&bearer), serde_json::json!({"subject": {}})),
        400
    );
    let page2 = client
        .get(xrpc("dev.knotless.listReports?limit=2"))
        .header("authorization", basic)
        .send()
        .unwrap()
        .json::<Value>()
        .unwrap();
    assert_eq!(page2["cursor"], "3");

    // The label stream, from the start: everything given out so far, then
    // what comes; from past the end, only what comes after it.
    let stream = |cursor: Option<i64>| {
        let url = format!(
            "{}/xrpc/com.atproto.label.subscribeLabels{}",
            av.base.replace("http", "ws"),
            cursor.map(|c| format!("?cursor={c}")).unwrap_or_default()
        );
        av.rt
            .block_on(tokio_tungstenite::connect_async(url))
            .unwrap()
            .0
    };
    let (mut from_start, mut from_end) = (stream(Some(0)), stream(Some(1000)));
    let next = |ws: &mut tokio_tungstenite::WebSocketStream<_>| {
        let msg = av.rt.block_on(ws.next()).unwrap().unwrap();
        knotless_serve::moderation::parse(&msg.into_data()).unwrap()
    };
    assert_eq!(next(&mut from_start).0, 1);
    for _ in 2..=6 {
        next(&mut from_start);
    }
    ok(&admin, &["admin", "takedown", trolling]);
    let (seq, labels) = next(&mut from_start);
    assert_eq!((seq, labels[0]["uri"].as_str()), (7, Some(trolling)));
    let ping = tokio_tungstenite::tungstenite::Message::Ping(vec![].into());
    av.rt.block_on(from_start.send(ping)).unwrap();
    av.rt.block_on(from_start.close(None)).unwrap();
    av.rt.block_on(from_end.close(None)).unwrap();
    av.rt.block_on(stream(None).close(None)).unwrap();
    ok(&admin, &["admin", "takedown", trolling, "--undo"]);
    // Labels follow from where they were read to.
    let labels = other.app.index.as_ref().unwrap();
    assert!(labels.labeler_seq(&av.did).unwrap().is_some());
    // One taken down as soon as it is opened does not show by its URI,
    // indexed yet or not.
    let fresh = ok(&troll, &["issue", &url, "Again"]);
    let fresh = fresh.trim();
    ok(&admin, &["admin", "takedown", fresh]);
    assert_eq!(reporter.cli(&["show", &url, fresh], b"").0, 1);
}