//! The five counterparts a Tangled command talks to, in miniature.
//!
//! Each is a plain function from a request to a reply over the shared
//! [`World`], with every call journalled first. They implement exactly the
//! methods atgc calls and refuse everything else by name — a test that
//! fails with `the mock PDS has no com.atproto.repo.describeRepo` is telling
//! you a command started asking for something new, which is worth knowing.
//!
//! **What is and is not asserted here.** These mocks check the things a real
//! PDS checks and atgc depends on: `swapRecord` and `swapCommit`
//! preconditions, that a record being updated exists, that a write carries
//! credentials. They do *not* verify DPoP proofs or validate records against
//! their lexicon — the first is jacquard's to get right and has no bearing
//! on any atgc bug, and the second would make the mock an authority on a
//! lexicon it does not have.

use super::cid;

use super::http::{Incoming, Reply};
use super::world::{Call, Record, World};

/// The `rev` every commit response carries.
///
/// Never read by atgc — nothing in the tree looks at a commit's rev — but it
/// is typed as a TID in the lexicon, so jacquard refuses a response whose
/// rev is not thirteen characters of the TID alphabet. A mock that answered
/// `"mockrev"` failed every write with a deserialization error, which is a
/// good demonstration of what these tests catch that a hand-written fixture
/// would not: the response shape is checked by the real client.
const MOCK_REV: &str = "3lmockrev2222";

/// Journal a call before answering it.
fn note(world: &mut World, service: &'static str, method: &str, req: &Incoming, json: bool) {
    let actor = req
        .token
        .as_ref()
        .and_then(|t| world.tokens.get(t))
        .cloned();
    world.journal.push(Call {
        service,
        method: method.to_string(),
        params: req.params.clone(),
        body: if json {
            req.json()
        } else {
            serde_json::Value::Null
        },
        body_bytes: req.body.len(),
        actor,
    });
}

/// Whether this service has been told to answer with a flood instead of a
/// reply, and how big a one. See [`World::flood`].
///
/// Consulted *after* the call is journalled, so a test can still assert
/// which read was the one that got flooded rather than inferring it from
/// the message.
fn flooding(world: &World, service: &str) -> Option<Reply> {
    match world.flood {
        Some((flooded, bytes)) if flooded == service => Some(Reply::Flood(bytes)),
        _ => None,
    }
}

/// A PDS: records, blobs, and the service-auth tokens a knot call needs.
/// A PDS that hosts exactly one account, which is what a PDS is.
///
/// Every per-account method names its subject — `repo` on the record calls,
/// `did` on `getBlob`, the bearer token on the rest — and a host answers for
/// nobody else. Asking Alice's PDS for Bob's records is not a slow path or a
/// redirect: it is a 404, and a command that does it is broken however
/// plausible its output looks.
///
/// The harness had one PDS for every account, so that whole class of mistake
/// was unrepresentable. `stack merge` had been reading a contributor's patch
/// blob from the *merging* account's host for as long as the command had
/// existed, and its own test passed.
pub fn pds_for(host: &'static str) -> impl Fn(&mut World, &Incoming) -> Reply {
    move |world, req| {
        if let Some(named) = subject_of(world, req)
            && named != host
        {
            let nsid = req.nsid().unwrap_or("(not xrpc)").to_string();
            note(world, "pds", &nsid, req, false);
            return Reply::not_found(format!(
                "{host} does not host {named}; this PDS answers for {host} only"
            ));
        }
        if let Some(reply) = oauth(world, req, host) {
            note(world, "pds", &format!("oauth:{}", req.path), req, false);
            return reply;
        }
        pds(world, req)
    }
}

/// A write into a repository the bearer does not own, which a real PDS
/// refuses however good the token is.
///
/// The mock checked only that *a* token was present, so a command writing
/// into somebody else's repository was accepted and the suite stayed green.
/// The host guard in [`pds_for`] already stops the ordinary shape of that
/// mistake — the write would have to be addressed to the other account's
/// host — but a token and a `repo` can disagree on one host too, and this is
/// the axis that says so rather than the one that says "wrong door".
fn refuse_foreign_write(world: &World, req: &Incoming, repo: &str) -> Option<Reply> {
    let bearer = req.token.as_ref().and_then(|t| world.tokens.get(t))?;
    (bearer != repo).then(|| {
        Reply::Error(
            hyper::StatusCode::FORBIDDEN,
            "AuthRequired",
            format!("{bearer} may not write into {repo}'s repository"),
        )
    })
}

/// Which account a request is about: the record calls name it, `getBlob`
/// names it, and everything else is the bearer's own business.
fn subject_of(world: &World, req: &Incoming) -> Option<String> {
    if let Some(did) = req.param("repo").or_else(|| req.param("did")) {
        return Some(did.to_string());
    }
    let body = req.json();
    if let Some(did) = body["repo"].as_str() {
        return Some(did.to_string());
    }
    req.token
        .as_ref()
        .and_then(|t| world.tokens.get(t))
        .cloned()
}

/// The OAuth endpoints a login discovers and calls.
///
/// **Nothing in this file served these before, which is why no test had ever
/// completed a login** — the account fixture planted sessions instead. That
/// left the one path that *creates* the on-disk layout unexercised, and a
/// directory atgc made was never compared against a directory a test made.
///
/// Deliberately not a conformant authorization server. It does not verify the
/// DPoP proof, the PKCE challenge or the client assertion, because the
/// property under test is what atgc writes when a login succeeds, and a mock
/// that re-implements the checks would be testing jacquard against itself.
/// What it does honour is the shape: PAR before authorize, a code exchanged
/// once at the token endpoint, and a `sub` the client is expected to match.
fn oauth(world: &mut World, req: &Incoming, host_did: &str) -> Option<Reply> {
    let base = world.pds_base.get(host_did).cloned().unwrap_or_default();
    let base = base.as_str();
    let did = world.login_as.clone();
    let did = did.as_deref().unwrap_or(host_did);
    Some(match req.path.as_str() {
        ".well-known/oauth-protected-resource" => Reply::ok(serde_json::json!({
            "resource": base,
            "authorization_servers": [base],
            // Required by the type, not optional: omitting it fails the
            // whole login with a bare "json error".
            "scopes_supported": ["atproto", "transition:generic"],
            "bearer_methods_supported": ["header"],
        })),
        ".well-known/oauth-authorization-server" => Reply::ok(serde_json::json!({
            "issuer": base,
            "authorization_endpoint": format!("{base}/oauth/authorize"),
            "token_endpoint": format!("{base}/oauth/token"),
            "pushed_authorization_request_endpoint": format!("{base}/oauth/par"),
            "require_pushed_authorization_requests": true,
            "scopes_supported": ["atproto", "transition:generic"],
            "response_types_supported": ["code"],
            "grant_types_supported": ["authorization_code", "refresh_token"],
            "code_challenge_methods_supported": ["S256"],
            "token_endpoint_auth_methods_supported": ["none"],
            "dpop_signing_alg_values_supported": ["ES256"],
            "client_id_metadata_document_supported": true,
        })),
        "oauth/par" => {
            // Remember what was asked for. The request is form-encoded, and
            // `scope` is the one field the token response has to agree with.
            let body = String::from_utf8_lossy(&req.body).into_owned();
            world.granted_scope = body.split('&').find_map(|pair| {
                pair.strip_prefix("scope=")
                    .map(|v| v.replace('+', " ").replace("%3A", ":").replace("%2F", "/"))
            });
            Reply::JsonStatus(
                hyper::StatusCode::CREATED,
                serde_json::json!({
                    "request_uri": "urn:ietf:params:oauth:request_uri:test",
                    "expires_in": 300,
                }),
            )
        }
        "oauth/token" => {
            let body = String::from_utf8_lossy(&req.body).into_owned();
            let field = |name: &str| {
                body.split('&').find_map(|pair| {
                    pair.strip_prefix(&format!("{name}="))
                        .map(|v| v.replace('+', " "))
                })
            };
            // A refresh token is single-use. Presenting a spent one is what a
            // real authorization server answers `invalid_grant` to, and what
            // two unserialised refreshes of one session produce.
            if field("grant_type").as_deref() == Some("refresh_token")
                && let Some(token) = field("refresh_token")
                && !world.spent_refresh.insert(token.clone())
            {
                return Some(Reply::Error(
                    hyper::StatusCode::BAD_REQUEST,
                    "invalid_grant",
                    format!("refresh token {token} has already been spent"),
                ));
            }
            world.tokens_issued += 1;
            let n = world.tokens_issued;
            Reply::ok(serde_json::json!({
            "access_token": format!("access-token-{n}"),
            "token_type": "DPoP",
            "refresh_token": format!("refresh-token-{n}"),
            "expires_in": 3600,
            "sub": did,
            // What was asked for at the pushed request, so the scope
            // pre-flight sees the grant a real server would have issued.
            "scope": world
                .granted_scope
                .clone()
                .unwrap_or_else(|| "atproto transition:generic".to_string()),
            }))
        }
        _ => return None,
    })
}

pub fn pds(world: &mut World, req: &Incoming) -> Reply {
    let Some(nsid) = req.nsid() else {
        return Reply::not_found(format!("the mock PDS serves /xrpc only, not /{}", req.path));
    };
    let nsid = nsid.to_string();
    // A blob upload's body is bytes, not JSON; everything else is JSON or
    // has no body at all.
    note(
        world,
        "pds",
        &nsid,
        req,
        nsid != "com.atproto.repo.uploadBlob",
    );
    if let Some(flood) = flooding(world, "pds") {
        return flood;
    }
    // A PDS that is unreachable, for the account this request names. Read
    // off the `repo` parameter, which every per-account method carries, so
    // one account's records can be unavailable while another's answer.
    if let Some(method) = world.pds_method_fails
        && nsid == method
    {
        return Reply::Error(
            hyper::StatusCode::BAD_GATEWAY,
            "BadGateway",
            format!("the PDS is unreachable for {method}"),
        );
    }
    if let Some(did) = req.param("repo")
        && world.pds_down.iter().any(|d| *d == did)
    {
        return Reply::Error(
            hyper::StatusCode::BAD_GATEWAY,
            "BadGateway",
            format!("{did}'s PDS is unreachable"),
        );
    }

    match nsid.as_str() {
        "com.atproto.repo.listRecords" => list_records(world, req),
        "com.atproto.repo.getRecord" => get_record(world, req),
        "com.atproto.repo.createRecord" => create_record(world, req),
        "com.atproto.repo.putRecord" => put_record(world, req),
        "com.atproto.repo.deleteRecord" => delete_record(world, req),
        "com.atproto.repo.applyWrites" => apply_writes(world, req),
        "com.atproto.repo.uploadBlob" => upload_blob(world, req),
        "com.atproto.sync.getBlob" => get_blob(world, req),
        "com.atproto.sync.getLatestCommit" => latest_commit(world, req),
        "com.atproto.server.getServiceAuth" => service_auth(world, req),
        other => Reply::not_found(format!("the mock PDS has no {other}")),
    }
}

fn list_records(world: &mut World, req: &Incoming) -> Reply {
    let (Some(did), Some(collection)) = (req.param("repo"), req.param("collection")) else {
        return Reply::bad("InvalidRequest", "listRecords needs repo and collection");
    };
    // **Newest first, which is what a real PDS does.** `listRecords` returns
    // a collection in descending record-key order, and record keys are TIDs,
    // so that is newest to oldest. This mock served them ascending for its
    // whole life, which made every paging test arrange its keys backwards
    // from reality — and made the one rule that reads the *order* rather
    // than the count untestable altogether: the backwards walks stop when a
    // page ends below the subject's own key, which against an ascending mock
    // is a condition that can never hold on the page it is meant to hold on.
    let mut all = world.collection(did, collection);
    all.reverse();
    // Paging by record key, which is how a real PDS cursors this collection
    // and what makes a multi-page listing testable at all.
    let limit: usize = req
        .param("limit")
        .and_then(|l| l.parse().ok())
        .unwrap_or(50)
        .min(100);
    let after = req.param("cursor").unwrap_or("").to_string();
    let page: Vec<(String, Record)> = all
        .into_iter()
        // The cursor walks *down* now, for the same reason: a real cursor is
        // the last key of the page just served and the next page continues
        // below it.
        .filter(|(rkey, _)| after.is_empty() || rkey.as_str() < after.as_str())
        .take(limit)
        .collect();
    let cursor = page.last().map(|(rkey, _)| rkey.clone());
    let records: Vec<serde_json::Value> = page
        .iter()
        .map(|(rkey, record)| {
            serde_json::json!({
                "uri": format!("at://{did}/{collection}/{rkey}"),
                "cid": record.cid,
                "value": record.value,
            })
        })
        .collect();
    let mut body = serde_json::json!({ "records": records });
    // A cursor only when the page was full — the documented end of a
    // listing is a page with no cursor, and handing one back forever is how
    // a mock teaches a client to spin.
    if records.len() == limit
        && let Some(cursor) = cursor
    {
        body["cursor"] = serde_json::json!(cursor);
    }
    Reply::ok(body)
}

fn get_record(world: &mut World, req: &Incoming) -> Reply {
    let (Some(did), Some(collection), Some(rkey)) = (
        req.param("repo"),
        req.param("collection"),
        req.param("rkey"),
    ) else {
        return Reply::bad(
            "InvalidRequest",
            "getRecord needs repo, collection and rkey",
        );
    };
    match world.get(did, collection, rkey) {
        Some(record) => Reply::ok(serde_json::json!({
            "uri": format!("at://{did}/{collection}/{rkey}"),
            "cid": record.cid,
            "value": record.value,
        })),
        None => Reply::not_found(format!("no {collection} record {rkey} in {did}")),
    }
}

/// `createRecord`, which is how `pr create` writes its one pull.
///
/// The record key is the *server's* to mint when the request does not name
/// one, and it must be a TID: the collection is ordered by key, and a
/// listing built on keys that do not sort by creation time would put a
/// pull's rounds and its neighbours in an order no PDS would. jacquard's own
/// ticker mints them, so the mock cannot drift from the client's idea of
/// what a record key is.
fn create_record(world: &mut World, req: &Incoming) -> Reply {
    let body = req.json();
    if req.token.is_none() {
        return Reply::Error(
            hyper::StatusCode::UNAUTHORIZED,
            "AuthMissing",
            "createRecord is authenticated".into(),
        );
    }
    let (Some(did), Some(collection)) = (body["repo"].as_str(), body["collection"].as_str()) else {
        return Reply::bad("InvalidRequest", "createRecord needs repo and collection");
    };
    if let Some(refusal) = refuse_foreign_write(world, req, did) {
        return refusal;
    }
    let rkey = match body["rkey"].as_str() {
        Some(rkey) => rkey.to_string(),
        None => next_tid(),
    };
    world.plant(did, collection, &rkey, body["record"].clone());
    let cid = world
        .get(did, collection, &rkey)
        .expect("just planted")
        .cid
        .clone();
    let commit = world.repo(did).commit_cid(did);
    Reply::ok(serde_json::json!({
        "uri": format!("at://{did}/{collection}/{rkey}"),
        "cid": cid,
        "commit": { "cid": commit, "rev": MOCK_REV },
        "validationStatus": "valid",
    }))
}

/// The next record key this process will mint, monotonic across every mock
/// in the test binary.
fn next_tid() -> String {
    use std::sync::Mutex;
    static TICKER: Mutex<Option<jacquard::types::tid::Ticker>> = Mutex::new(None);
    let mut guard = TICKER.lock().expect("the ticker's lock");
    let ticker = guard.get_or_insert_with(jacquard::types::tid::Ticker::new);
    ticker.next(None).as_str().to_string()
}

fn put_record(world: &mut World, req: &Incoming) -> Reply {
    let body = req.json();
    if req.token.is_none() {
        return Reply::Error(
            hyper::StatusCode::UNAUTHORIZED,
            "AuthMissing",
            "putRecord is authenticated".into(),
        );
    }
    let (Some(did), Some(collection), Some(rkey)) = (
        body["repo"].as_str(),
        body["collection"].as_str(),
        body["rkey"].as_str(),
    ) else {
        return Reply::bad(
            "InvalidRequest",
            "putRecord needs repo, collection and rkey",
        );
    };
    if let Some(refusal) = refuse_foreign_write(world, req, did) {
        return refusal;
    }
    // The compare-and-swap `clients/atproto/record.rs` exists to send. A
    // mock that ignored it would make every test of the lost-update guard
    // pass by doing nothing.
    if let Some(expected) = body["swapRecord"].as_str() {
        let current = world.get(did, collection, rkey).map(|r| r.cid.clone());
        if current.as_deref() != Some(expected) {
            return Reply::bad("InvalidSwap", "Record was at a different CID");
        }
    }
    let value = body["record"].clone();
    world.plant(did, collection, rkey, value);
    let record = world.get(did, collection, rkey).expect("just planted");
    let cid = record.cid.clone();
    let commit = world.repo(did).commit_cid(did);
    Reply::ok(serde_json::json!({
        "uri": format!("at://{did}/{collection}/{rkey}"),
        "cid": cid,
        "commit": { "cid": commit, "rev": MOCK_REV },
        "validationStatus": "valid",
    }))
}

fn delete_record(world: &mut World, req: &Incoming) -> Reply {
    let body = req.json();
    let (Some(did), Some(collection), Some(rkey)) = (
        body["repo"].as_str(),
        body["collection"].as_str(),
        body["rkey"].as_str(),
    ) else {
        return Reply::bad(
            "InvalidRequest",
            "deleteRecord needs repo, collection and rkey",
        );
    };
    if let Some(refusal) = refuse_foreign_write(world, req, did) {
        return refusal;
    }
    if let Some(expected) = body["swapRecord"].as_str() {
        let current = world.get(did, collection, rkey).map(|r| r.cid.clone());
        if current.as_deref() != Some(expected) {
            return Reply::bad("InvalidSwap", "Record was at a different CID");
        }
    }
    let did = did.to_string();
    let key = (collection.to_string(), rkey.to_string());
    let repo = world.repo(&did);
    repo.records.remove(&key);
    repo.rev += 1;
    let commit = repo.commit_cid(&did);
    Reply::ok(serde_json::json!({
        "commit": { "cid": commit, "rev": MOCK_REV },
    }))
}

/// `applyWrites`: all of it, or none of it.
///
/// The atomicity is not decoration. `stack create` and `stack resubmit` both
/// depend on it — every `dependentOn` in a chain points at a record the same
/// commit contains — so a mock that applied ops one at a time and stopped on
/// the first failure would let a half-written chain pass a test that the
/// real PDS would never have produced.
fn apply_writes(world: &mut World, req: &Incoming) -> Reply {
    let body = req.json();
    if req.token.is_none() {
        return Reply::Error(
            hyper::StatusCode::UNAUTHORIZED,
            "AuthMissing",
            "applyWrites is authenticated".into(),
        );
    }
    let Some(did) = body["repo"].as_str().map(String::from) else {
        return Reply::bad("InvalidRequest", "applyWrites needs a repo");
    };
    if let Some(refusal) = refuse_foreign_write(world, req, &did) {
        return refusal;
    }
    if world.fail_next_batch_swap {
        world.fail_next_batch_swap = false;
        return Reply::bad("InvalidSwap", "Commit was at a different CID");
    }
    if let Some(expected) = body["swapCommit"].as_str() {
        let current = world.repo(&did).commit_cid(&did);
        if current != expected {
            return Reply::bad("InvalidSwap", "Commit was at a different CID");
        }
    }
    let writes = body["writes"].as_array().cloned().unwrap_or_default();

    // Decide everything first, so a refusal writes nothing.
    let mut planned: Vec<(String, String, Option<serde_json::Value>)> = Vec::new();
    for write in &writes {
        let kind = write["$type"].as_str().unwrap_or_default();
        let (Some(collection), Some(rkey)) = (write["collection"].as_str(), write["rkey"].as_str())
        else {
            return Reply::bad("InvalidRequest", "a write named no collection or rkey");
        };
        match kind {
            "com.atproto.repo.applyWrites#create" => planned.push((
                collection.to_string(),
                rkey.to_string(),
                Some(write["value"].clone()),
            )),
            "com.atproto.repo.applyWrites#update" => {
                if world.get(&did, collection, rkey).is_none() {
                    return Reply::bad(
                        "InvalidRequest",
                        format!("cannot update {collection}/{rkey}: no such record"),
                    );
                }
                planned.push((
                    collection.to_string(),
                    rkey.to_string(),
                    Some(write["value"].clone()),
                ));
            }
            "com.atproto.repo.applyWrites#delete" => {
                planned.push((collection.to_string(), rkey.to_string(), None))
            }
            other => return Reply::bad("InvalidRequest", format!("unknown write type {other}")),
        }
    }

    let mut results = Vec::new();
    for ((collection, rkey, value), write) in planned.into_iter().zip(&writes) {
        let kind = write["$type"].as_str().unwrap_or_default();
        match value {
            Some(value) => {
                world.plant(&did, &collection, &rkey, value);
                let cid = world
                    .get(&did, &collection, &rkey)
                    .expect("just planted")
                    .cid
                    .clone();
                let result = if kind.ends_with("#create") {
                    "com.atproto.repo.applyWrites#createResult"
                } else {
                    "com.atproto.repo.applyWrites#updateResult"
                };
                results.push(serde_json::json!({
                    "$type": result,
                    "uri": format!("at://{did}/{collection}/{rkey}"),
                    "cid": cid,
                    "validationStatus": "valid",
                }));
            }
            None => {
                let repo = world.repo(&did);
                repo.records.remove(&(collection, rkey));
                repo.rev += 1;
                results.push(serde_json::json!({
                    "$type": "com.atproto.repo.applyWrites#deleteResult",
                }));
            }
        }
    }
    let commit = world.repo(&did).commit_cid(&did);
    Reply::ok(serde_json::json!({
        "commit": { "cid": commit, "rev": MOCK_REV },
        "results": results,
    }))
}

fn upload_blob(world: &mut World, req: &Incoming) -> Reply {
    if req.token.is_none() {
        return Reply::Error(
            hyper::StatusCode::UNAUTHORIZED,
            "AuthMissing",
            "uploadBlob is authenticated".into(),
        );
    }
    let cid = cid::of(&req.body);
    let size = req.body.len();
    if let Some(owner) = req
        .token
        .as_ref()
        .and_then(|t| world.tokens.get(t))
        .cloned()
    {
        world.blob_owner.insert(cid.clone(), owner);
    }
    world.blobs.insert(cid.clone(), req.body.clone());
    Reply::ok(serde_json::json!({
        "blob": {
            "$type": "blob",
            "ref": { "$link": cid },
            "mimeType": "application/gzip",
            "size": size,
        }
    }))
}

fn get_blob(world: &mut World, req: &Incoming) -> Reply {
    let Some(cid) = req.param("cid").map(String::from) else {
        return Reply::bad("InvalidRequest", "getBlob needs a cid");
    };
    let Some(did) = req.param("did").map(String::from) else {
        return Reply::bad("InvalidRequest", "getBlob needs a did");
    };
    // A blob is in one account's repository. Asking a different account's
    // PDS for it is a 404, exactly as a real PDS answers — which is the only
    // reason a caller that asks the wrong one can be caught.
    if let Some(owner) = world.blob_owner.get(&cid)
        && *owner != did
    {
        return Reply::not_found(format!("no blob {cid} in {did}"));
    }
    match world.blobs.get(&cid) {
        Some(bytes) => Reply::Bytes(bytes.clone()),
        None => Reply::not_found(format!("no blob {cid}")),
    }
}

fn latest_commit(world: &mut World, req: &Incoming) -> Reply {
    let Some(did) = req.param("did").map(String::from) else {
        return Reply::bad("InvalidRequest", "getLatestCommit needs a did");
    };
    let cid = world.repo(&did).commit_cid(&did);
    Reply::ok(serde_json::json!({ "cid": cid, "rev": MOCK_REV }))
}

/// A knot call's service-auth token.
///
/// A real PDS mints a signed JWT whose `aud` and `lxm` the knot verifies.
/// The mock knot verifies nothing, so what matters here is only that the
/// token is *distinguishable*: it carries the audience, the method and the
/// account, so a journal entry at the knot can be checked against the
/// account that asked the PDS for it. That is the identity question, and it
/// is the one this whole suite is about.
fn service_auth(world: &mut World, req: &Incoming) -> Reply {
    let actor = req
        .token
        .as_ref()
        .and_then(|t| world.tokens.get(t))
        .cloned()
        .unwrap_or_else(|| "anonymous".to_string());
    let aud = req.param("aud").unwrap_or_default();
    let lxm = req.param("lxm").unwrap_or_default();
    Reply::ok(serde_json::json!({ "token": format!("service-auth/{actor}/{aud}/{lxm}") }))
}

/// The PLC directory: DID documents, and a 404 for a DID nobody planted.
pub fn plc(world: &mut World, req: &Incoming) -> Reply {
    let did = req.path.clone();
    note(world, "plc", &did, req, false);
    if let Some(flood) = flooding(world, "plc") {
        return flood;
    }
    match world.dids.get(&did) {
        Some(doc) => Reply::ok(doc.clone()),
        None => Reply::not_found(format!("no DID document for {did}")),
    }
}

/// Bobbin, the appview index. Answers from its own list, which a test can
/// leave empty or stale on purpose.
pub fn bobbin(world: &mut World, req: &Incoming) -> Reply {
    let Some(nsid) = req.nsid().map(String::from) else {
        return Reply::not_found(format!(
            "the mock Bobbin serves /xrpc only, not /{}",
            req.path
        ));
    };
    note(world, "bobbin", &nsid, req, false);
    if let Some(flood) = flooding(world, "bobbin") {
        return flood;
    }
    if let Some(error) = world.bobbin_fails {
        return Reply::Error(
            hyper::StatusCode::INTERNAL_SERVER_ERROR,
            error,
            "the index is unavailable".into(),
        );
    }
    match nsid.as_str() {
        "sh.tangled.repo.listPulls" | "sh.tangled.repo.listPullsBy" => {
            Reply::ok(serde_json::json!({ "items": world.bobbin_pulls }))
        }
        other => Reply::not_found(format!("the mock Bobbin has no {other}")),
    }
}

/// Tangled's website, which atgc scrapes for one thing: a pull's number.
///
/// Every page 404s, and that is the interesting answer rather than a gap.
/// A number is assigned by the appview's own index, so a pull that was
/// written seconds ago has none — which is the normal state of every pull
/// these tests create, and the state a command has to survive without
/// pretending it has a link it does not. Serving invented pages here would
/// test the scraper, which `tests/fixtures/appview_pull_page_excerpt.html`
/// already does against real markup.
/// An appview, which for these tests means one thing: the HTML page a pull
/// *number* names.
///
/// The number is in no record and no XRPC response, so `pr diff 23` and every
/// write verb that takes a number resolve one by fetching `<repo>/pulls/<n>`
/// and reading the `data-aturi` off it. What the page says is therefore what
/// decides which record a command acts on, and [`World::pull_pages`] holds one
/// answer per `(repo path, number)` pair rather than per number.
pub fn appview(world: &mut World, req: &Incoming) -> Reply {
    note(world, "appview", &req.path, req, false);
    // `com.atproto.identity.resolveHandle`, the XRPC fallback the resolver
    // reaches when DNS and the well-known URL cannot answer -- which in a
    // test is always, since a handle like `alice.test` names no real host.
    // atgc points that fallback at whatever `ATGC_BSKY_APPVIEW` names, so
    // this is where a login's first lookup lands.
    // The DID document, for the same fallback. The resolver asks the XRPC
    // endpoint before the PLC directory, so a login's second lookup lands
    // here too.
    if req.nsid() == Some("com.atproto.identity.resolveDid") {
        let did = req.param("did").unwrap_or_default().to_string();
        // What the real appview does: it has no DID-document endpoint, so a
        // test may say so and force resolution through the PLC directory the
        // way production always does.
        if !world.appview_resolves_dids {
            return Reply::bad(
                "MethodNotImplemented",
                format!("this appview does not resolve DID documents ({did:?})"),
            );
        }
        return match world.dids.get(&did) {
            // `didDoc`, not the document bare: the lexicon wraps it.
            Some(doc) => Reply::ok(serde_json::json!({ "didDoc": doc })),
            None => Reply::bad("InvalidRequest", format!("unable to resolve did {did:?}")),
        };
    }
    if req.nsid() == Some("com.atproto.identity.resolveHandle") {
        let handle = req.param("handle").unwrap_or_default().to_string();
        // The DID documents the PLC mock serves are the same ones that carry
        // `alsoKnownAs`, so the mapping is already here rather than kept
        // twice.
        let found = world.dids.iter().find(|(_, doc)| {
            doc["alsoKnownAs"].as_array().is_some_and(|a| {
                a.iter()
                    .any(|v| v.as_str() == Some(&format!("at://{handle}")))
            })
        });
        return match found {
            Some((did, _)) => Reply::ok(serde_json::json!({ "did": did })),
            None => Reply::bad(
                "InvalidRequest",
                format!(
                    "unable to resolve handle {handle:?}; known: {:?}",
                    world
                        .dids
                        .iter()
                        .map(|(d, doc)| format!("{d}={}", doc["alsoKnownAs"]))
                        .collect::<Vec<_>>()
                ),
            ),
        };
    }
    if world.web_fails {
        return Reply::Error(
            hyper::StatusCode::BAD_GATEWAY,
            "BadGateway",
            "the appview is unavailable".into(),
        );
    }
    if let Some((repo, number)) = req.path.rsplit_once("/pulls/")
        && let Ok(number) = number.parse::<u32>()
        && let Some(uri) = world.pull_pages.get(&(repo.to_string(), number))
    {
        return Reply::Bytes(pull_page(uri).into_bytes());
    }
    // `/<repoDid>` redirects to `/<owner-handle>/<name>`, which is how the
    // real appview publishes who owns a repo and the only route
    // `resolve::owner_of` has to it — a repo's DID document names its knot
    // and nothing else.
    let path = req.path.trim_matches('/');
    if path.starts_with("did:")
        && let Some((owner, name)) = world.repo_owners.get(path)
    {
        // **The real appview redirects to the owner's *handle*; this uses
        // their DID.** `resolve::owner_of` accepts either, and the harness
        // has no way to answer a handle: `resolve_handle` is jacquard's and
        // goes to DNS and `.well-known` on the real network, neither of
        // which a loopback mock can stand in front of. So the half of that
        // path these tests exercise is "the appview names an owner and the
        // status walk reads their records", and the handle-to-DID hop is
        // covered nowhere — worth knowing before trusting a green run here
        // to mean the whole lookup works.
        return Reply::Redirect(format!("/{owner}/{name}"));
    }
    Reply::not_found(format!("the mock appview has no page at /{}", req.path))
}

/// A pull's page, cut down to the one thing atgc reads out of it.
///
/// The real page is up to 2.5 MB of rendered diff and this is four lines of
/// it, which is honest rather than lazy: a fixture is what proves the scrape
/// survives a real page (see `tests/fixtures/appview_pull_page_excerpt.html`),
/// and a mock's job here is only to give two repos different answers so a test
/// can tell which one was asked.
fn pull_page(uri: &str) -> String {
    format!(
        "<!doctype html><html><body>\
         <button data-aturi=\"{uri}\">copy record URI</button>\
         </body></html>"
    )
}

/// A knot: the merge check and the merge itself.
///
/// It performs no git operation. What a merge means to the commands under
/// test is "the knot accepted this patch and the branch moved", and the
/// records written afterwards are the observable part; a knot that actually
/// applied the patch would be testing git, which git already tests.
pub fn knot(world: &mut World, req: &Incoming) -> Reply {
    let Some(nsid) = req.nsid().map(String::from) else {
        return Reply::not_found(format!(
            "the mock knot serves /xrpc only, not /{}",
            req.path
        ));
    };
    // The knot authenticates with the PDS-minted service-auth token, which
    // carries the acting account; `note` resolves the actor off the token
    // table, so plant the service-auth spelling there too.
    let actor = req
        .token
        .as_ref()
        .and_then(|t| t.strip_prefix("service-auth/"))
        .and_then(|rest| rest.split('/').next())
        .map(String::from);
    note(world, "knot", &nsid, req, true);
    if let Some(actor) = &actor
        && let Some(last) = world.journal.last_mut()
    {
        last.actor = Some(actor.clone());
    }

    if let Some((method, error)) = world.knot_fails
        && nsid == method
    {
        return Reply::Error(
            hyper::StatusCode::BAD_GATEWAY,
            error,
            format!("the knot is unavailable for {method}"),
        );
    }
    match nsid.as_str() {
        "sh.tangled.repo.mergeCheck" => match world.merge_check.clone() {
            Ok(()) => Reply::ok(serde_json::json!({ "is_conflicted": false })),
            Err(detail) => Reply::ok(serde_json::json!({
                "is_conflicted": true,
                "message": detail,
                "conflicts": [{ "filename": "conflicted.txt", "reason": detail }],
            })),
        },
        // **The knot names a repository by its DID and decodes nothing
        // else.** `DeleteBranch` and `SetDefaultBranch` in
        // `knotserver/xrpc/` unmarshal `repo` into `repoident.RepoDid`,
        // whose `UnmarshalText` runs `syntax.ParseDID`, so an at-uri never
        // reaches a handler at all: it fails in `json.Decode` and comes back
        // `400` tagged `Generic` (read off tangled.org/tangled.org/core on
        // 2026-09-05). This mock used to accept whatever it was handed,
        // which is how atgc shipped the record's at-uri here and only found
        // out against a live knot — see issue `3mufarg7ust2w`.
        "sh.tangled.repo.deleteBranch" | "sh.tangled.repo.setDefaultBranch"
            if !req
                .json()
                .get("repo")
                .and_then(|repo| repo.as_str())
                .is_some_and(|repo| repo.starts_with("did:")) =>
        {
            let repo = req.json()["repo"].clone();
            Reply::bad(
                "Generic",
                format!("invalid request body: invalid repoDid {repo}: expected DID"),
            )
        }
        // The one call a knot authorizes: it merges for anyone with push
        // access to the repo, not only for its owner. A refusal carries the
        // `AccessControl` tag, which is the half atgc matches on.
        // `deleteBranch` and `setDefaultBranch` go through the same
        // `pushableRepoDID` guard in `knotserver/xrpc/` as `merge` does, so
        // the mock applies it to all three. It used to guard only `merge`,
        // which made the mock *permit* what a real knot refuses — a fake
        // service that is more permissive than the real one hides exactly
        // the bugs these tests exist to find.
        "sh.tangled.repo.merge"
        | "sh.tangled.repo.deleteBranch"
        | "sh.tangled.repo.setDefaultBranch"
            if matches!(
                (&world.knot_push_allowed, &actor),
                (Some(allowed), Some(actor)) if !allowed.contains(actor)
            ) =>
        {
            let actor = actor.clone().unwrap_or_default();
            Reply::denied(format!(
                "DID does not have sufficient access permissions for this operation: {actor}"
            ))
        }
        "sh.tangled.repo.merge" => Reply::ok(serde_json::json!({})),
        // The read `repo delete-branch` guards itself with, and the only
        // unauthenticated method here. A test makes it fail to stand in for
        // the ordinary case: a 5xx from a proxy, or a knot that renamed the
        // query, while the authenticated deletion beside it still works.
        "sh.tangled.repo.getDefaultBranch" => match world.default_branch.clone() {
            Ok(name) => Reply::ok(serde_json::json!({ "name": name })),
            Err(detail) => Reply::bad("InternalServerError", detail),
        },
        // Unauthenticated, like `getDefaultBranch` beside it: the knot
        // formats a diff for whoever asks, which is what lets `pr create`
        // take the patch from the knot without a scope or a session. A
        // refusal is tagged `RevisionNotFound`, the knot's own word for a
        // branch it has not got.
        "sh.tangled.repo.compare" => match world.compare.clone() {
            Ok((commits, patch)) => Reply::ok(serde_json::json!({
                "rev1": "1111111111111111111111111111111111111111",
                "rev2": "2222222222222222222222222222222222222222",
                "merge_base": "3333333333333333333333333333333333333333",
                "format_patch": vec![serde_json::json!({}); commits],
                "patch": patch,
            })),
            Err(detail) => Reply::bad("RevisionNotFound", detail),
        },
        "sh.tangled.repo.deleteBranch" => Reply::ok(serde_json::json!({})),
        // Accepted unconditionally, including for the branch that is already
        // the default: a knot has no reason to refuse that, which is why the
        // no-op check `repo default-branch` makes is worth testing at all.
        // Nothing here updates `World::default_branch` — a test that wants
        // the read to agree with a write it just made says so itself, and
        // one that does not is testing the command rather than the knot.
        "sh.tangled.repo.setDefaultBranch" => Reply::ok(serde_json::json!({})),
        // The two halves of atgc's irreversible pair. Neither touches git
        // here — what `repo create` and `repo delete` have to get right is
        // the *order* of the knot call and the record write, and the journal
        // is where that shows.
        //
        // A `repoDid` in the success body is the one thing `create` reads out
        // of this, and a knot old enough not to mint them is a case the
        // command carries a warning for, so the answer is a whole DID rather
        // than an empty object.
        "sh.tangled.repo.create" => match world.knot_create.clone() {
            Ok(repo_did) => Reply::ok(serde_json::json!({ "repoDid": repo_did })),
            Err(detail) => Reply::conflict(detail),
        },
        "sh.tangled.repo.delete" => match world.knot_delete.clone() {
            Ok(()) => Reply::ok(serde_json::json!({})),
            Err(detail) => Reply::denied(detail),
        },
        other => Reply::not_found(format!("the mock knot has no {other}")),
    }
}