//! A read accelerator for knotless repositories.
//!
//! A PDS can hand out records and blobs and nothing else: no partial clone,
//! no shallow fetch, no "what is in this file at this commit". This server
//! keeps a bare mirror per repository and answers those from it:
//!
//! - git smart HTTP (protocol v0 and v2, `--filter`, `--depth`) at
//!   `/<handle or did>/<name>`, read only;
//! - a JSON and raw API at `/api/<handle or did>/<name>/...` for agents that
//!   want a file, a tree, a log or a diff without a clone.
//!
//! Everything it holds is a cache of the owners' records, which stay the
//! truth: each mirror is revalidated against them (one round trip) when
//! older than the TTL, or kept warm by the firehose or Jetstream, and the
//! index (a SQLite file) is built from them and can be rebuilt by a backfill.
//! Every object it serves is named by its hash, so a client can check it.

pub mod accounts;
mod api;
mod appview;
pub mod archive;
pub mod card;
pub mod code;
pub mod diff;
pub mod firehose;
pub mod gateway;
pub mod highlight;
pub mod index;
mod invites;
pub mod jetstream;
pub mod languages;
pub mod live;
pub mod login;
pub mod logs;
pub mod moderation;
pub mod objects;
pub mod oci;
mod orgs;
pub mod plans;
pub mod private;
pub mod releases;
mod secrets;
pub mod shared;
pub mod sites;
mod smart;
pub mod spaces;
pub mod status;
pub mod telemetry;
pub mod usage;
pub mod viewer;
pub mod web;

use anyhow::Result;
use axum::Router;
use axum::http::StatusCode;
use axum::response::{IntoResponse, Response};
use axum::routing::{get, post};
use knotless::env::Ctx;
use knotless::git::Git;
use knotless::mirror::Mirror;
use knotless::records::RepoRecord;
use knotless::records::nsid::PROFILES;
use knotless::remote::Remote;
use knotless::telemetry::{Fields, Level, millis, note, slow};
use knotless::url::RepoUrl;
use std::collections::HashMap;
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

pub struct App {
    pub ctx: Ctx,
    /// How long a mirror is trusted without asking the PDS, when the
    /// firehose is not watching.
    pub ttl: Duration,
    repos: Mutex<HashMap<(String, String), Shared<Entry>>>,
    /// The private repositories viewers were shown, each viewer's apart, by
    /// viewer, owner and name ([`private`]).
    viewed: Mutex<HashMap<(String, String, String), Shared<Entry>>>,
    /// How long, in milliseconds, a viewer's reading of a private
    /// repository is taken as allowed before its space is asked again.
    pub reprove: std::sync::atomic::AtomicU64,
    /// The domain repositories' sites are served under, as `knotless.page`,
    /// once a [`sites::Sites`] serves them: their pages show the address.
    pub pages_domain: std::sync::OnceLock<String>,
    /// Firehose events seen per DID.
    events: Mutex<HashMap<String, u64>>,
    /// Of those, the ones that could change any of the account's
    /// repositories, per DID ([`App::stamp`]).
    account_events: Mutex<HashMap<String, u64>>,
    /// And those that change one repository, by its owner's DID and key.
    repo_events: Mutex<HashMap<(String, String), u64>>,
    /// Writes made from the pages (or by this server for an account), per
    /// DID: a read after one waits for its sync ([`App::served`]).
    writes: Mutex<HashMap<String, u64>>,
    /// Each public repository's mirror as last synced, by owner and name,
    /// for a read to take without waiting on the lock a sync holds.
    last: Mutex<HashMap<(String, String), Entry>>,
    /// Each repository's storage as last counted, kept until it moves
    /// (`dev.knotless.getRepoStorage`).
    pub(crate) counts: web::Counts,
    /// The last commit, by its rev, that wrote to each DID's repository
    /// collections ([`first_of`]).
    revs: Mutex<HashMap<String, String>>,
    pub watching: AtomicBool,
    /// Syncs of each space that found something written there, by URI:
    /// shared with a sync run aside from the read that asked for it.
    moves: Arc<Mutex<HashMap<String, u64>>>,
    /// Set while the spaces kept are synced ([`spaces::follow`]): a
    /// private repository's mirror is then fresh until its space moves.
    pub following: AtomicBool,
    /// The feeds followed, as their watchers tell of them, by name.
    feeds: Mutex<std::collections::BTreeMap<&'static str, status::Followed>>,
    /// How long a feed may stay silent before `/_status` calls it dead,
    /// rather than each feed's own default.
    pub stale: Option<Duration>,
    /// The directory whose disk `/_status` watches for room: the cache's,
    /// where mirrors go. None in tests, which share whatever disk they get.
    pub disk: Option<std::path::PathBuf>,
    started: Instant,
    /// Set when this server also takes writes for owners who delegate them.
    pub gateway: Option<gateway::Gateway>,
    /// Set when this server is an appview, indexing what the firehose says.
    pub index: Option<Arc<index::Index>>,
    /// Where people log in, this server's own index of the private
    /// repositories read here, synced from their spaces ([`spaces`]): kept
    /// apart from `index`, so nothing of a space reaches an appview's.
    pub spaces: Option<Arc<index::Index>>,
    /// When each account's spaces were last listed, for private
    /// repositories of theirs not yet in `spaces`, and those shared with
    /// them asked after ([`spaces::discover`]).
    pub discovered: Mutex<HashMap<String, Instant>>,
    /// Whether each owner's PDS has spaces, as their listing last found.
    pub with_spaces: Mutex<HashMap<String, bool>>,
    /// The spaces their authorities told of writes in since they were last
    /// synced, to sync now ([`spaces::written`]).
    due: Mutex<std::collections::HashSet<String>>,
    /// Wakes [`spaces::follow`] for what is due.
    told: tokio::sync::Notify,
    /// When each space was last synced in full, as anyone here synced it:
    /// a read syncs it itself, aside unless it never was, once that is
    /// [`spaces::STALE`] ago, as when
    /// the session of the member it is synced as no longer reads it. Its
    /// entries taken out, every space counts as not synced lately.
    pub last_synced: Arc<Mutex<HashMap<String, Instant>>>,
    /// When this server last asked each space's authority to tell it of
    /// writes there.
    registered: tokio::sync::Mutex<HashMap<String, Instant>>,
    /// Set when this server searches the code of public repositories
    /// ([`code`]).
    pub code: Option<Arc<code::Code>>,
    /// Set when this server counts how it is used, for its operator.
    pub usage: Option<Arc<usage::Usage>>,
    /// Set when this server keeps the lines it logs, for its operator.
    pub logs: Option<Arc<logs::Logs>>,
    /// The labels this server gives out, to those following them.
    pub labels: tokio::sync::broadcast::Sender<(i64, serde_json::Value)>,
    /// When each check still running was last read again from its PDS.
    rechecked: Mutex<HashMap<String, Instant>>,
    /// The delegation tokens queries brought, by `jti`, each until it ends:
    /// spent once ([`App::spend`]).
    spent: Mutex<HashMap<String, u64>>,
    /// The runner, by DID, a new public repository names unless its
    /// owner names others, as `dev.knotless.describeServer` says.
    pub default_runner: Option<String>,
    /// The oldest client version this appview asks for, and why, as
    /// `dev.knotless.describeServer` says.
    pub client_minimum: Option<(String, Option<String>)>,
    /// The oldest build of the iOS app this appview lets run, and why, as
    /// `dev.knotless.describeServer` says.
    pub ios_minimum: Option<(u64, Option<String>)>,
    /// The output of the checks running now, and the spaces written to,
    /// for whoever follows them.
    pub live: live::Live,
    /// One lock per organization, held while this gateway changes its
    /// records: a space's records take no compare and swap, so two changes
    /// at once would each write over the other.
    orgs: Mutex<HashMap<String, Shared<()>>>,
    /// One lock per account, held while its spaces are listed
    /// ([`spaces::discover`]): two listings at once could each keep what
    /// the other's authority said after it, a removal undone by an answer
    /// asked before it.
    discovers: Mutex<HashMap<String, Shared<()>>>,
}

type Shared<T> = Arc<tokio::sync::Mutex<T>>;

#[derive(Default, Clone)]
struct Entry {
    git: Option<Git>,
    /// The repository's owner, key and name now, which differ from what
    /// was asked for after a move or a rename.
    found: Option<(String, String, String)>,
    record: RepoRecord,
    /// The space it is in, if it is private.
    space: Option<String>,
    synced: Option<Instant>,
    /// When its viewer was last found allowed to read it, for a private
    /// repository a viewer was shown.
    proved: Option<Instant>,
    /// Where a private repository was then found, read until it is
    /// proved again.
    remote: Option<Remote>,
    /// The events it was last synced as of ([`App::stamp`]).
    events: (u64, u64),
    /// Its space's moves when it was last synced, for a private one.
    moves: u64,
    /// The writes made from the pages for its owner when it was last
    /// synced ([`App::writes_for`]).
    writes: u64,
    /// When the first event came that made it stale, if one has since it
    /// was synced: kept with [`App::last`] only.
    stale_since: Option<Instant>,
}

/// How long a mirror that events have made stale is still served, to a
/// read that wrote nothing, while its sync catches up.
const STALE_FOR: Duration = Duration::from_secs(10);

/// Whether a read is served `entry`, a mirror as it was last synced,
/// without waiting on a sync, when its events now stamp as `stamp` and
/// `wrote` writes were made from the pages: if no event has come since it
/// synced, or the first came within [`STALE_FOR`], and the sync it started
/// is then still catching up; never after a write from the pages that it
/// may not have seen.
fn serves(entry: &Entry, stamp: (u64, u64), wrote: u64) -> bool {
    let catching_up = entry.stale_since.is_some_and(|at| at.elapsed() < STALE_FOR);
    entry.writes == wrote && (entry.events == stamp || catching_up)
}

/// A repository found by name, and its mirror.
pub struct Located {
    pub git: Git,
    pub did: String,
    pub key: String,
    /// Its name now: after a rename, not the name it was asked for by.
    pub name: String,
    /// Whether it is in another account than the one asked for: it moved.
    pub moved: bool,
    /// Its record, as the PDS had it when the mirror last synced.
    pub record: RepoRecord,
    /// The space it is in, if it is private.
    pub space: Option<String>,
    /// What reads it: this server, or for a private repository the viewer
    /// it was found for, so that its records are read as theirs.
    pub reader: Ctx,
}

/// The index of the private repositories this server may keep one of,
/// kept with its cache: none if it cannot be opened.
fn spaces_index(ctx: &Ctx) -> Option<Arc<index::Index>> {
    let dir = ctx.env.cache_dir.join("serve");
    std::fs::create_dir_all(&dir).ok()?;
    let index = index::Index::open(&dir.join("private.sqlite")).ok()?;
    Some(Arc::new(index))
}

impl App {
    pub fn new(ctx: Ctx, ttl: Duration) -> Arc<App> {
        App::build(ctx, ttl, None, None)
    }

    pub fn build(
        ctx: Ctx,
        ttl: Duration,
        gateway: Option<gateway::Gateway>,
        index: Option<index::Index>,
    ) -> Arc<App> {
        let spaces = gateway.as_ref().and_then(|_| spaces_index(&ctx));
        // Asking itself, as the pages ask its appview and gateway for a
        // logged-in account, it signs the token: none from their PDS. Failing
        // that, it asks their PDS as anyone else would.
        if let Some(gw) = &gateway {
            let kept = ctx.env.config_dir.join("loopback.json");
            let _ = knotless::auth::loopback(&kept, &gw.did);
        }
        Arc::new(App {
            ctx,
            ttl,
            repos: Mutex::default(),
            viewed: Mutex::default(),
            reprove: std::sync::atomic::AtomicU64::new(private::REPROVE.as_millis() as u64),
            pages_domain: std::sync::OnceLock::new(),
            events: Mutex::default(),
            account_events: Mutex::default(),
            repo_events: Mutex::default(),
            writes: Mutex::default(),
            last: Mutex::default(),
            counts: web::Counts::default(),
            revs: Mutex::default(),
            watching: AtomicBool::new(false),
            moves: Default::default(),
            following: AtomicBool::new(false),
            feeds: Mutex::default(),
            stale: None,
            disk: None,
            started: Instant::now(),
            gateway,
            index: index.map(Arc::new),
            spaces,
            discovered: Mutex::default(),
            with_spaces: Mutex::default(),
            due: Mutex::default(),
            told: tokio::sync::Notify::new(),
            last_synced: Default::default(),
            registered: tokio::sync::Mutex::default(),
            code: None,
            usage: None,
            logs: None,
            labels: tokio::sync::broadcast::channel(256).0,
            rechecked: Mutex::default(),
            spent: Mutex::default(),
            default_runner: None,
            client_minimum: None,
            ios_minimum: None,
            live: live::Live::default(),
            orgs: Mutex::default(),
            discovers: Mutex::default(),
        })
    }

    /// Whether `did`'s spaces are to be listed now: not within
    /// [`spaces::DISCOVER`] of the last time.
    pub fn discovering(&self, did: &str) -> bool {
        let mut discovered = self.discovered.lock().unwrap();
        let recent = discovered
            .get(did)
            .is_some_and(|at| at.elapsed() < spaces::DISCOVER);
        if !recent {
            discovered.insert(did.to_string(), Instant::now());
        }
        !recent
    }

    /// The lock held while the organization `org`'s records change.
    pub fn org_lock(&self, org: &str) -> Shared<()> {
        let mut orgs = self.orgs.lock().unwrap();
        orgs.entry(org.to_string()).or_default().clone()
    }

    /// The lock held while `did`'s spaces are listed.
    pub(crate) fn discover_lock(&self, did: &str) -> Shared<()> {
        let mut discovers = self.discovers.lock().unwrap();
        discovers.entry(did.to_string()).or_default().clone()
    }

    /// Whether `did`'s PDS has spaces, for private repositories: unknown
    /// until their spaces are first listed ([`spaces::discover`]).
    pub fn has_spaces(&self, did: &str) -> Option<bool> {
        self.with_spaces.lock().unwrap().get(did).copied()
    }

    pub(crate) fn found_spaces(&self, did: &str, has: bool) {
        let mut with_spaces = self.with_spaces.lock().unwrap();
        with_spaces.insert(did.to_string(), has);
    }

    /// Reads again from its PDS each of `checks` the index still has
    /// queued or running a minute after it was filed so, once a minute at
    /// most: a stream can miss the write that started or finished it, as one written while serve is
    /// itself rolling over to a new build. In the background, so a later
    /// read has it.
    pub fn recheck(&self, index: &Arc<index::Index>, checks: &[knotless::forge::Entry]) {
        let ago = std::time::SystemTime::now() - RECHECK;
        let before = humantime::format_rfc3339_millis(ago).to_string();
        let mut rechecked = self.rechecked.lock().unwrap();
        rechecked.retain(|_, at| at.elapsed() < RECHECK);
        let stale = checks.iter().filter(|c| {
            let started = c.value["startedAt"].as_str().unwrap_or_default();
            let going = c.value["status"]
                .as_str()
                .is_some_and(knotless::records::status::going);
            going && started < before.as_str()
        });
        // Not again within the minute, however often it is read meanwhile.
        let uris = stale.filter(|c| {
            !rechecked.contains_key(&c.uri)
                && rechecked.insert(c.uri.clone(), Instant::now()).is_none()
        });
        let uris: Vec<String> = uris.map(|c| c.uri.clone()).collect();
        drop(rechecked);
        for (did, collection, rkey) in uris
            .iter()
            .filter_map(|u| knotless::forge::parse_uri(u).ok())
        {
            let op = firehose::Op {
                action: "update".into(),
                collection: collection.into(),
                rkey: rkey.into(),
                record: None,
            };
            let (ctx, index, did) = (self.ctx.clone(), index.clone(), did.to_string());
            tokio::spawn(async move {
                let _ = index_op(&ctx, &index, &did, &op, None).await;
            });
        }
    }

    /// Takes in a firehose or Jetstream event: refreshes the mirrors of an account whose
    /// repositories moved, indexes what it wrote, and hides or drops an
    /// account its host says is no longer active or deleted ([`accounts`]).
    pub fn observe(self: &Arc<App>, mut event: firehose::Event) {
        let Some(did) = event.did else {
            return;
        };
        if let Some(status) = &event.account {
            accounts::changed(self, &did, status);
            // Nothing of one deleted is synced again.
            if status == "deleted" {
                return;
            }
        }
        // A handle changed: the one kept is looked up again, as the event
        // only says what it is. Kept only for accounts that were looked up.
        if event.identity
            && let Some(index) = self.index.clone()
            && index.handle(&did).ok().flatten().is_some()
        {
            tokio::spawn(web::rename(self.ctx.clone(), index, did.clone()));
        }
        // Profiles change all over the network: an edit to one is taken in
        // only for an account whose profile is kept, and is no sign that
        // its repositories moved. A knotless profile is always taken in,
        // for whether it says the account is a bot.
        let profile = |op: &firehose::Op| {
            let bluesky = op.collection != knotless::records::nsid::PROFILE;
            bluesky && PROFILES.contains(&op.collection.as_str())
        };
        let kept = |index: &Arc<index::Index>| index.profile_kept(&did).unwrap_or(false);
        let unkept = event.ops.iter().any(profile) && !self.index.as_ref().is_some_and(kept);
        if unkept {
            event.ops.retain(|op| !profile(op));
        }
        if unkept && event.ops.is_empty() {
            return;
        }
        let repository = event
            .ops
            .iter()
            .any(|op| knotless::records::nsid::ALL.contains(&op.collection.as_str()));
        // Jetstream sends each record a commit wrote as an event of its own,
        // and all of them are there once the first is: a push's pack, claim
        // and ref are one sync, not a sync and then another for the rest.
        // A record of one repository makes only its mirrors stale; anything
        // else, all of the account's.
        let moved = event.ops.is_empty() || repository;
        let keys = scopes(&event.ops);
        for key in keys.iter().filter(|_| moved) {
            let seen = format!("{did} {}", key.as_deref().unwrap_or_default());
            if first_of(&self.revs, &seen, event.rev.as_deref()) {
                self.heard_of(&did, key.as_deref(), false);
            }
        }
        if let Some(index) = self.index.clone()
            && !event.ops.is_empty()
        {
            // The keys of the repositories whose records it deletes, read
            // before they go, for what is kept of them to go after.
            let repo = knotless::records::nsid::REPO;
            let deleted: Vec<String> = event
                .ops
                .iter()
                .filter(|op| op.action == "delete" && op.collection == repo)
                .filter_map(|op| {
                    let uri = format!("at://{did}/{repo}/{}", op.rkey);
                    let found = index.repo(&uri).ok().flatten();
                    found.and_then(|r| r["key"].as_str().map(str::to_string))
                })
                .collect();
            // What the event carries in full is indexed now, in order, so a
            // later write cannot land before an earlier one.
            if event
                .ops
                .iter()
                .all(|op| op.action == "delete" || op.record.is_some())
            {
                for op in &event.ops {
                    let rev = event.rev.as_deref();
                    telemetry::indexed(&did, index_inline(&index, &did, op, rev));
                }
                accounts::repos_deleted(self, &did, &deleted);
                self.warm_checks(&did, &event.ops);
                return;
            }
            let (app, ctx) = (self.clone(), self.ctx.clone());
            tokio::spawn(async move {
                let rev = event.rev.as_deref();
                for op in &event.ops {
                    telemetry::indexed(&did, index_op(&ctx, &index, &did, op, rev).await);
                }
                accounts::repos_deleted(&app, &did, &deleted);
                app.warm_checks(&did, &event.ops);
            });
        }
    }

    /// Reads ahead the logs of the checks `ops` filed for `did`, once they
    /// are indexed ([`web::warm_check_log`]).
    fn warm_checks(self: &Arc<App>, did: &str, ops: &[firehose::Op]) {
        let check = knotless::records::nsid::CHECK;
        let filed = ops.iter().filter(|op| op.collection == check);
        for op in filed.filter(|op| op.action != "delete") {
            let (app, did, rkey) = (self.clone(), did.to_string(), op.rkey.clone());
            tokio::spawn(web::warm_check_log(app, did, rkey));
        }
    }

    /// Whether the account `did` is taken down here, or by a labeler this
    /// server follows: never without an index, and not when the index
    /// cannot answer.
    pub fn taken_down(&self, did: &str) -> bool {
        let index = self.index.as_ref();
        index.is_some_and(|i| i.down(&[did]).unwrap_or(false))
    }

    /// Firehose events seen for `did`.
    pub fn events_for(&self, did: &str) -> u64 {
        self.events.lock().unwrap().get(did).copied().unwrap_or(0)
    }

    /// What a mirror of `did`'s repository with `key` goes stale with: the
    /// events that could change any of the account's repositories, and
    /// those of that one. One of a key not yet known goes stale with every
    /// event of the account.
    pub(crate) fn stamp(&self, did: &str, key: Option<&str>) -> (u64, u64) {
        let Some(key) = key else {
            return (self.events_for(did), u64::MAX);
        };
        let account = self.account_events.lock().unwrap().get(did).copied();
        let repos = self.repo_events.lock().unwrap();
        let repo = repos.get(&(did.to_string(), key.to_string())).copied();
        (account.unwrap_or(0), repo.unwrap_or(0))
    }

    /// What `entry`, synced now, is fresh as of: `stamp`, taken before its
    /// sync for the key it had then; for a key the sync first found, as
    /// of now, but only if no event of the account came meanwhile, as
    /// `any`, its events then, says. Otherwise none, to sync again.
    pub(crate) fn settled(
        &self,
        did: &str,
        entry_key: Option<&str>,
        stamp: (u64, u64),
        any: u64,
        found: Option<&str>,
    ) -> (u64, u64) {
        if found == entry_key {
            return stamp;
        }
        let quiet = self.events_for(did) == any;
        let stale = (u64::MAX, u64::MAX);
        [stale, self.stamp(did, found)][usize::from(quiet)]
    }

    /// Syncs of the space `uri` that found something written there.
    pub fn moves_of(&self, uri: &str) -> u64 {
        self.moves.lock().unwrap().get(uri).copied().unwrap_or(0)
    }

    /// Notes that the space `uri` was synced just now.
    pub(crate) fn synced_now(&self, uri: &str) {
        let mut last = self.last_synced.lock().unwrap();
        last.insert(uri.to_string(), Instant::now());
    }

    /// Whether the space `uri` was not synced within [`spaces::STALE`],
    /// or not since this server started.
    pub(crate) fn sync_stale(&self, uri: &str) -> bool {
        let last = self.last_synced.lock().unwrap();
        last.get(uri).is_none_or(|at| at.elapsed() >= spaces::STALE)
    }

    /// Records that the space `uri` was written to, or is no longer synced,
    /// so that mirrors of it sync again as they are next read.
    pub fn moved(&self, uri: &str) {
        *self
            .moves
            .lock()
            .unwrap()
            .entry(uri.to_string())
            .or_default() += 1;
    }

    /// Records that `did` wrote to a repository collection, and refreshes its
    /// mirrors in the background.
    pub fn touch(self: &Arc<App>, did: &str) {
        self.heard_of(did, None, true);
    }

    /// The writes from the pages recorded for `did` ([`App::touch`]).
    fn writes_for(&self, did: &str) -> u64 {
        self.writes.lock().unwrap().get(did).copied().unwrap_or(0)
    }

    /// The mirror of `did`'s repository `name` as last synced, if a read
    /// may have it now ([`serves_last`]): never unless the firehose is
    /// watching, which is what makes the sync of a stale one certain.
    fn served(&self, did: &str, name: &str, wrote: u64) -> Option<Entry> {
        if !self.watching.load(Ordering::Relaxed) {
            return None;
        }
        let last = self.last.lock().unwrap();
        let entry = last.get(&(did.to_string(), name.to_string()))?;
        let stamp = self.stamp(did, entry.key(did).as_deref());
        serves(entry, stamp, wrote).then(|| entry.clone())
    }

    /// Keeps `entry`, just synced, as what [`App::served`] gives, or none
    /// if the repository is not there.
    fn remember(&self, did: &str, name: &str, entry: &Entry) {
        let mut last = self.last.lock().unwrap();
        let at = (did.to_string(), name.to_string());
        match entry.git {
            Some(_) => last.insert(at, entry.clone()),
            None => last.remove(&at),
        };
    }

    /// Records that `did` wrote to its repository with `key`, or with none
    /// to what could change any of them, and refreshes its mirrors in the
    /// background: those it did not write to are found fresh, and left.
    fn heard_of(self: &Arc<App>, did: &str, key: Option<&str>, own: bool) {
        *self
            .events
            .lock()
            .unwrap()
            .entry(did.to_string())
            .or_default() += 1;
        let now = Instant::now();
        for (_, entry) in self
            .last
            .lock()
            .unwrap()
            .iter_mut()
            .filter(|(at, _)| at.0 == did)
        {
            entry.stale_since.get_or_insert(now);
        }
        if own {
            *self.writes.lock().unwrap().entry(did.into()).or_default() += 1;
        }
        match key {
            Some(key) => {
                let mut repos = self.repo_events.lock().unwrap();
                *repos.entry((did.into(), key.into())).or_default() += 1;
            }
            None => {
                *self
                    .account_events
                    .lock()
                    .unwrap()
                    .entry(did.into())
                    .or_default() += 1
            }
        }
        let names: Vec<String> = self
            .repos
            .lock()
            .unwrap()
            .keys()
            .filter(|(d, _)| d == did)
            .map(|(_, n)| n.clone())
            .collect();
        for name in names {
            let app = self.clone();
            let did = did.to_string();
            tokio::spawn(async move {
                let _ = app.located_with(&did, &name, true).await;
            });
        }
    }

    /// The mirror of `authority/name`, synced if it may be stale.
    pub async fn mirror(&self, authority: &str, name: &str) -> Result<Git, AppError> {
        Ok(self.located(authority, name).await?.git)
    }

    /// The mirror of `authority/name`, with the repository's key and the
    /// name it has now: a public repository, or a private one the viewer
    /// of the request being answered may read ([`private`]).
    pub async fn located(&self, authority: &str, name: &str) -> Result<Located, AppError> {
        self.located_with(authority, name, true).await
    }

    /// [`App::located`] for a read of the repository that answers a request:
    /// a mirror a change has made stale may be served while it syncs
    /// ([`App::served`]). Nothing that derives and keeps something from the
    /// mirror may read it so.
    pub async fn located_for_reader(
        &self,
        authority: &str,
        name: &str,
    ) -> Result<Located, AppError> {
        self.located_with(authority, name, false).await
    }

    /// [`App::located`], waiting for the sync of a stale mirror when `wait`,
    /// as the refresh a change starts does: it is the sync.
    async fn located_with(
        &self,
        authority: &str,
        name: &str,
        wait: bool,
    ) -> Result<Located, AppError> {
        let url = RepoUrl::parse(&format!("knotless://{authority}/{name}"))
            .map_err(|e| AppError(StatusCode::BAD_REQUEST, e.to_string()))?;
        let remote = Remote::connect(&self.ctx, &url).await;
        let remote = remote.map_err(AppError::upstream)?;
        let did = remote.did().to_string();
        let gateway = self.gateway.is_some();
        // A delegation token for this one's space proves its asker a member,
        // with no session of theirs here.
        let delegated = private::delegated(&knotless::space::uri(&did, &url.name));
        let delegated = delegated.filter(|_| gateway);
        let viewer = private::viewer().filter(|_| gateway);
        let viewer = viewer.or_else(|| delegated.as_ref().map(|d| d.did.clone()));
        // One the viewer was shown as private is read as theirs again. Not
        // found so, it is looked for as anyone looks: shown here may be only
        // a look under way, such as the new repository form's check of a
        // name, which a repository made public meanwhile is not in.
        let shown = viewer
            .as_ref()
            .filter(|v| self.shown_private(v, &did, &url.name));
        if let Some(viewer) = shown {
            let read = self.as_viewer(viewer, &did, &url, None, delegated.as_ref());
            let read = read.await;
            if !read.as_ref().is_err_and(|e| e.0 == StatusCode::NOT_FOUND) {
                return read;
            }
        }
        let owner = wait || viewer.as_deref() == Some(did.as_str());
        let missing = match self.public(remote, &url, owner).await {
            Err(e) if e.0 == StatusCode::NOT_FOUND => e,
            found => return found,
        };
        match &viewer {
            Some(viewer) => {
                let read = self.as_viewer(viewer, &did, &url, Some(missing), delegated.as_ref());
                read.await
            }
            None => Err(missing),
        }
    }

    /// [`App::located`] for a repository anyone may read.
    async fn public(
        &self,
        remote: Remote,
        url: &RepoUrl,
        owner: bool,
    ) -> Result<Located, AppError> {
        let did = remote.did().to_string();
        // Watched, a mirror events have made stale is still served for a
        // moment, while the sync they started finishes: a read does not
        // wait on the lock that sync holds, unless its owner made it, who
        // may have just written with a tool of their own.
        let wrote = self.writes_for(&did);
        let view = self.served(&did, &url.name, wrote).filter(|_| !owner);
        if let Some(view) = view {
            return self.admit(url, self.found(&view, &did, self.ctx.clone()));
        }
        let entry = self
            .repos
            .lock()
            .unwrap()
            .entry((did.clone(), url.name.clone()))
            .or_default()
            .clone();
        let mut entry = entry.lock().await;
        let key = entry.key(&did);
        let any = self.events_for(&did);
        let events = self.stamp(&did, key.as_deref());
        let fresh = match (&entry.git, entry.synced) {
            (Some(_), Some(_)) if self.watching.load(Ordering::Relaxed) => entry.events == events,
            // A write from the pages touches it either way. Not there, it is
            // asked after again on the same terms as one that is.
            (_, Some(at)) => at.elapsed() < self.ttl && entry.events == events,
            _ => false,
        };
        if fresh && entry.git.is_none() {
            return Err(AppError(
                StatusCode::NOT_FOUND,
                format!("{url}: no such repository"),
            ));
        }
        if !fresh {
            let synced = self.sync(&self.ctx, remote, url, &mut entry).await;
            // Found, or found not there, as of these events.
            if synced.is_ok() || entry.git.is_none() {
                let found = entry.key(&did);
                entry.events = self.settled(&did, key.as_deref(), events, any, found.as_deref());
            }
            if synced.is_ok() {
                entry.writes = wrote;
            }
            self.remember(&did, &url.name, &entry);
            synced?;
        }
        self.admit(url, self.found(&entry, &did, self.ctx.clone()))
    }

    /// `found`, unless it is taken down.
    fn admit(&self, url: &RepoUrl, found: Located) -> Result<Located, AppError> {
        let uri = knotless::forge::repo_uri(&found.did, &found.name);
        if let Some(index) = &self.index
            && index
                .down(&[&found.did, &uri])
                .map_err(anyhow::Error::from)
                .map_err(AppError::internal)?
        {
            return Err(AppError(
                StatusCode::NOT_FOUND,
                format!("{url}: taken down"),
            ));
        }
        Ok(found)
    }

    /// Reads `remote`'s records afresh with `ctx`, and syncs `entry`'s
    /// mirror, kept under `ctx`'s cache, to them.
    async fn sync(
        &self,
        ctx: &Ctx,
        remote: Remote,
        url: &RepoUrl,
        entry: &mut Entry,
    ) -> Result<(), AppError> {
        let started = Instant::now();
        // Its record and refs, not its packs: those are read only when the
        // mirror lacks an object they hold.
        let snap = remote.snapshot(ctx).await.map_err(AppError::upstream)?;
        let (remote, snap) = remote.follow(ctx, snap).await.map_err(AppError::upstream)?;
        let records = started.elapsed();
        let Some(repo) = &snap.repo else {
            // Not there, as of now.
            entry.git = None;
            entry.synced = Some(Instant::now());
            return Err(AppError(
                StatusCode::NOT_FOUND,
                format!("{url}: no such repository"),
            ));
        };
        let key = repo.value.key.as_str();
        let mirror = Mirror::open(ctx, remote.did(), key)
            .await
            .map_err(AppError::internal)?;
        mirror
            .refresh(&remote, &snap)
            .await
            .map_err(AppError::upstream)?;
        aside(&mirror);
        entry.found = Some((remote.did().to_string(), key.to_string(), repo.rkey.clone()));
        entry.record = repo.value.clone();
        entry.space = remote.space.as_ref().map(|s| s.uri.clone());
        entry.git = Some(mirror.git);
        entry.synced = Some(Instant::now());
        // Every read of the repository waits on its sync: how long each
        // takes, and how much of that is its records, at warn when slow.
        let took = started.elapsed();
        let f = Fields {
            repo: Some(&url.to_string()),
            duration_ms: Some(millis(took)),
            records_ms: Some(millis(records)),
            ..Fields::default()
        };
        let level = [Level::INFO, Level::WARN][usize::from(slow(took))];
        note(level, "mirror sync", &f);
        Ok(())
    }

    /// What a synced `entry` holds, asked for under `did`, read with
    /// `reader`.
    fn found(&self, entry: &Entry, did: &str, reader: Ctx) -> Located {
        let (now, key, name) = entry.found.clone().expect("a synced entry has a mirror");
        let record = self.with_members(&now, entry.space.as_deref(), entry.record.clone());
        Located {
            git: entry.git.clone().expect("a synced entry has a mirror"),
            moved: now != did,
            did: now,
            key,
            name,
            record,
            space: entry.space.clone(),
            reader,
        }
    }
}

/// Tidies `mirror` in a task of its own, out of the way of the request
/// that synced it and of every other waiting on that sync.
pub(crate) fn aside(mirror: &Mirror) {
    let mirror = Mirror {
        git: mirror.git.clone(),
    };
    tokio::spawn(async move { mirror.tidy().await.ok() });
}

/// The repositories `ops` write to, by key, each once: a record of one
/// repository, kept under its key, names it; any other, or none at all,
/// is `None`, for all of them.
fn scopes(ops: &[firehose::Op]) -> Vec<Option<String>> {
    use knotless::records::nsid;
    let one = [nsid::REF, nsid::CLAIM, nsid::PACK];
    let mut keys: Vec<Option<String>> = ops
        .iter()
        .map(|op| {
            let key = op.rkey.split_once('.').map(|(key, _)| key.to_string());
            key.filter(|_| one.contains(&op.collection.as_str()))
        })
        .collect();
    if keys.is_empty() {
        keys.push(None);
    }
    keys.sort();
    keys.dedup();
    keys
}

impl Entry {
    /// The key of the repository it last found, if it is still `did`'s.
    fn key(&self, did: &str) -> Option<String> {
        let found = self.found.as_ref().filter(|(now, _, _)| now == did);
        found.map(|(_, key, _)| key.clone())
    }
}

/// Whether `rev` is a commit of `did`'s that `seen` has not had yet, and
/// from now on has. An event naming no commit always is one.
fn first_of(seen: &Mutex<HashMap<String, String>>, did: &str, rev: Option<&str>) -> bool {
    rev.is_none_or(|rev| {
        let last = seen.lock().unwrap().insert(did.into(), rev.into());
        last.as_deref() != Some(rev)
    })
}

/// Indexes a write the event carried in full, made in the commit at `rev`
/// when it says: a deletion, or a record delivered inline. `false` when
/// the record has to be fetched.
fn index_inline(
    index: &index::Index,
    did: &str,
    op: &firehose::Op,
    rev: Option<&str>,
) -> rusqlite::Result<bool> {
    let (collection, rkey) = (op.collection.as_str(), op.rkey.as_str());
    match (&op.record, op.action.as_str()) {
        (_, "delete") => index.delete_as_of(did, collection, rkey, rev)?,
        (Some(record), _) => index.put_as_of(did, collection, rkey, record, rev)?,
        (None, _) => return Ok(false),
    }
    Ok(true)
}

/// Indexes one write, made in the commit at `rev` when it is known: the
/// record the event carried, else the record as its PDS now has it, which
/// is as of that commit or later.
pub async fn index_op(
    ctx: &Ctx,
    index: &index::Index,
    did: &str,
    op: &firehose::Op,
    rev: Option<&str>,
) -> Result<()> {
    if index_inline(index, did, op, rev)? {
        return Ok(());
    }
    let id = knotless::identity::resolve(ctx, did).await?;
    let out: serde_json::Value = knotless::xrpc::Xrpc::new(ctx.http.clone(), &id.pds)
        .query(
            "com.atproto.repo.getRecord",
            &[
                ("repo", did),
                ("collection", &op.collection),
                ("rkey", &op.rkey),
            ],
            None,
        )
        .await?;
    index.put_as_of(did, &op.collection, &op.rkey, &out["value"], rev)?;
    Ok(())
}

/// Backfills from `service`, saying how it went.
pub async fn backfilled(
    ctx: &Ctx,
    index: &index::Index,
    service: &str,
    sources: &Sources,
) -> String {
    match backfill(ctx, index, service, sources).await {
        Ok(f) => format!(
            "backfilled {} records of {} accounts from {service}, {} unchanged, {} failed",
            f.records, f.read, f.unchanged, f.failed
        ),
        Err(err) => format!("backfilling from {service}: {err:#}"),
    }
}

/// The collections a backfill reads.
fn indexed_collections() -> impl Iterator<Item = &'static str> {
    [
        knotless::records::nsid::REPO,
        knotless::records::nsid::REF,
        knotless::records::nsid::CLAIM,
        knotless::records::nsid::REDIRECT,
        knotless::records::nsid::CHECK,
        knotless::records::nsid::MINUTES,
        knotless::sealed::KEY_NSID,
        knotless::records::nsid::STAR,
        knotless::records::nsid::FOLLOW,
        knotless::records::nsid::REACTION,
        knotless::records::nsid::RELEASE,
        knotless::records::nsid::RELEASE_YANK,
        knotless::records::nsid::SITE,
        knotless::records::nsid::THEME,
        knotless::records::nsid::ORG,
        knotless::records::nsid::ORG_MEMBER,
        knotless::records::nsid::ORG_ACCEPTANCE,
        knotless::records::nsid::PROFILE,
    ]
    .into_iter()
    .chain(knotless::records::nsid::FORGE)
}

/// The accounts on `xrpc`'s service to index: from a relay, those it lists
/// as holding an indexed collection, a few among millions; from a PDS,
/// which lists accounts only all at once, every one.
///
/// Each of `directories` is asked first, in turn, and the first to answer
/// for every collection is believed.
async fn accounts(ctx: &Ctx, service: &str, directories: &[String]) -> Result<Vec<String>> {
    for directory in directories {
        let xrpc = knotless::xrpc::Xrpc::new(ctx.http.clone(), directory);
        if let Ok(dids) = holders(&xrpc).await {
            return Ok(dids);
        }
    }
    let xrpc = knotless::xrpc::Xrpc::new(ctx.http.clone(), service);
    match holders(&xrpc).await {
        Ok(dids) => Ok(dids),
        Err(_) => pages(&xrpc, "com.atproto.sync.listRepos", None).await,
    }
}

/// The accounts `xrpc`'s service lists as holding an indexed collection.
async fn holders(xrpc: &knotless::xrpc::Xrpc) -> Result<Vec<String>> {
    let mut dids = std::collections::BTreeSet::new();
    for collection in indexed_collections() {
        let by = "com.atproto.sync.listReposByCollection";
        dids.extend(pages(xrpc, by, Some(collection)).await?);
    }
    Ok(dids.into_iter().collect())
}

/// The DIDs every page of `method` lists, with `collection` if given.
async fn pages(
    xrpc: &knotless::xrpc::Xrpc,
    method: &str,
    collection: Option<&str>,
) -> Result<Vec<String>> {
    #[derive(serde::Deserialize)]
    struct Repos {
        repos: Vec<Account>,
        cursor: Option<String>,
    }
    #[derive(serde::Deserialize)]
    struct Account {
        did: String,
    }
    let mut dids = Vec::new();
    let mut cursor: Option<String> = None;
    loop {
        let mut params = vec![("limit", "1000")];
        params.extend(collection.map(|c| ("collection", c)));
        params.extend(cursor.as_deref().map(|c| ("cursor", c)));
        let page: Repos = xrpc.query(method, &params, None).await?;
        let done = page.repos.is_empty();
        dids.extend(page.repos.into_iter().map(|r| r.did));
        match page.cursor {
            Some(next) if !done => cursor = Some(next),
            _ => break,
        }
    }
    Ok(dids)
}

/// Where a backfill finds the accounts to read and reads them, besides the
/// PDS or relay it is from and each account's PDS.
#[derive(Clone, Debug, Default, PartialEq)]
pub struct Sources {
    /// Directories asked, in turn, which accounts hold knotless records
    /// (`com.atproto.sync.listReposByCollection`), before the relay is.
    pub directories: Vec<String>,
    /// A mirror each account's repository is read from first, as
    /// STAR-lite or a CAR ([`archive::fetch`]).
    pub mirror: Option<String>,
}

/// What a backfill did: the records it indexed, from the accounts it
/// read; the accounts it found unchanged since it last read them, or put
/// off as one failed not long ago; and those it failed to read.
#[derive(Debug, Default)]
pub struct Filled {
    pub records: usize,
    pub read: usize,
    pub unchanged: usize,
    pub failed: usize,
}

/// How many accounts a backfill reads at once.
const READERS: usize = 4;

/// Indexes every account on `service`, a PDS or relay, that holds knotless
/// records: each account's repositories, checks and forge records, read
/// whole from its PDS and checked ([`archive`]).
///
/// An account whose latest commit is the one last read is passed over. A
/// record the feed indexed from a later commit than the one read stands
/// ([`index::Index::put_as_of`]). An account that fails is tried again
/// by a later backfill, a minute later for each time it failed, up to an
/// hour, and does not stop the rest.
pub async fn backfill(
    ctx: &Ctx,
    index: &index::Index,
    service: &str,
    sources: &Sources,
) -> Result<Filled> {
    use futures::StreamExt;
    let found = accounts(ctx, service, &sources.directories).await?;
    let mirror = sources.mirror.as_deref();
    let reads = futures::stream::iter(found)
        .map(|did| async move {
            let read = read_account(ctx, index, &did, mirror).await;
            (did, read)
        })
        .buffer_unordered(READERS);
    let reads: Vec<(String, Result<Option<usize>>)> = reads.collect().await;
    let mut filled = Filled::default();
    for (did, read) in reads {
        match read {
            Ok(Some(records)) => {
                (filled.records, filled.read) = (filled.records + records, filled.read + 1)
            }
            Ok(None) => filled.unchanged += 1,
            Err(err) => {
                filled.failed += 1;
                telemetry::backfill_failed(&did, &err);
                let tries = u64::from(index.backfilled(&did)?.tries);
                index.backfill_failed(&did, telemetry::now_secs() + 60 * (tries + 1).min(60))?;
            }
        }
    }
    Ok(filled)
}

/// Reads `did` into the index, from `mirror` when it has it, and how many
/// records it held; `None` when it is as last read, or a failure put it off
/// until later.
async fn read_account(
    ctx: &Ctx,
    index: &index::Index,
    did: &str,
    mirror: Option<&str>,
) -> Result<Option<usize>> {
    #[derive(serde::Deserialize)]
    struct Latest {
        rev: String,
    }
    let was = index.backfilled(did)?;
    if was.retry_at > telemetry::now_secs() {
        return Ok(None);
    }
    let id = knotless::identity::resolve(ctx, did).await?;
    let xrpc = knotless::xrpc::Xrpc::new(ctx.http.clone(), &id.pds);
    let asked = [("did", did)];
    let latest = xrpc.query::<Latest>("com.atproto.sync.getLatestCommit", &asked, None);
    if was.rev.as_deref() == Some(latest.await?.rev.as_str()) {
        return Ok(None);
    }
    let kept = |c: &str| indexed_collections().any(|i| i == c);
    let read = archive::fetch(ctx, did, &id.pds, mirror, archive::MAX, &kept).await?;
    for r in &read.records {
        index.put_as_of(did, &r.collection, &r.rkey, &r.value, Some(&read.rev))?;
    }
    index.read_through(did, &read.rev)?;
    index.backfill_read(did, &read.rev)?;
    Ok(Some(read.records.len()))
}

/// How often the index is filled again from the relay.
const SWEEP: Duration = Duration::from_secs(600);

/// How often the spaces this server keeps an index of are asked what was
/// written there, besides as their authorities tell of writes: for those
/// whose notifications went missing.
const SPACES: Duration = Duration::from_secs(30);
/// How often devices are told of what is new.
const PUSH_ROUND: Duration = Duration::from_secs(30);

/// How often what was counted is summed up in the log.
const SUMMARY: Duration = Duration::from_secs(60);

/// Backfills from `relay` every `every`, until the process ends. A record
/// whose event a feed never delivered, one Jetstream dropped or a commit
/// the relay resynced over, is otherwise missing from the index for good.
/// And puts back, each time, what labels that lapsed took down.
pub async fn sweep(
    ctx: Ctx,
    index: Arc<index::Index>,
    relay: String,
    sources: Sources,
    every: Duration,
) {
    loop {
        tokio::time::sleep(every).await;
        let _ = index.lapse();
        telemetry::sweep(&ctx, &index, &relay, &sources).await;
    }
}

/// `err` as the pages and the app say it: what to do about a refusal in
/// their own controls, never a command ([`knotless::fix`]).
pub(crate) fn shown(err: &anyhow::Error) -> String {
    knotless::fix::said(err, knotless::fix::Words::Pages)
}

/// A status and a message, as the response.
#[derive(Debug)]
pub struct AppError(pub StatusCode, pub String);

impl AppError {
    /// 502 for what another server failed at; 404 for a name that is no
    /// one's handle, as it cannot be one or a resolver found no one.
    pub(crate) fn upstream(err: anyhow::Error) -> AppError {
        let status = match err.downcast_ref::<knotless::identity::NoSuchHandle>() {
            Some(_) => StatusCode::NOT_FOUND,
            None => StatusCode::BAD_GATEWAY,
        };
        AppError(status, shown(&err))
    }
    /// 400 for input the request got wrong, else 500.
    pub(crate) fn input(err: anyhow::Error) -> AppError {
        let status = match err.downcast_ref::<knotless::read::BadInput>() {
            Some(_) => StatusCode::BAD_REQUEST,
            None => StatusCode::INTERNAL_SERVER_ERROR,
        };
        AppError(status, shown(&err))
    }

    pub(crate) fn internal(err: anyhow::Error) -> AppError {
        AppError(StatusCode::INTERNAL_SERVER_ERROR, shown(&err))
    }
}

impl IntoResponse for AppError {
    fn into_response(self) -> Response {
        telemetry::failed(self.0, self.1)
    }
}

pub fn router(app: Arc<App>) -> Router {
    // The queries, which know who asks when a service token says.
    let queries = Router::new()
        .route("/xrpc/dev.knotless.getForge", get(appview::get_forge))
        .route("/xrpc/dev.knotless.listRepos", get(appview::list_repos))
        .route(
            "/xrpc/dev.knotless.searchOwners",
            get(appview::search_owners),
        )
        .route(
            "/xrpc/dev.knotless.describeServer",
            get(appview::describe_server),
        )
        .route("/xrpc/dev.knotless.listChecks", get(appview::list_checks))
        .route("/xrpc/dev.knotless.getFeed", get(web::get_feed))
        .route("/xrpc/dev.knotless.getProfile", get(web::get_profile))
        .route(
            "/xrpc/dev.knotless.getActorRepos",
            get(web::get_actor_repos),
        )
        .route(
            "/xrpc/dev.knotless.getActorActivity",
            get(web::get_actor_activity),
        )
        .merge(web::queries())
        .route_layer(axum::middleware::from_fn_with_state(
            app.clone(),
            viewer::viewer,
        ))
        .layer(viewer::cors());
    Router::new()
        .route("/healthz", get(|| async { "ok" }))
        .route("/_status", get(status::status))
        .route("/_hit", post(usage::hit))
        .route("/xrpc/dev.knotless.getUsage", get(usage::report))
        .route("/xrpc/dev.knotless.getStats", get(usage::stats))
        .route("/xrpc/dev.knotless.getLogs", get(logs::get_logs))
        .route("/{authority}/{repo}/info/refs", get(smart::info_refs))
        .route(
            "/{authority}/{repo}/git-upload-pack",
            post(smart::upload_pack),
        )
        .route("/api/{authority}/{repo}/refs", get(api::refs))
        .route("/api/{authority}/{repo}/tree", get(api::tree))
        .route("/api/{authority}/{repo}/blob", get(api::blob))
        .route("/api/{authority}/{repo}/log", get(api::log))
        .route("/api/{authority}/{repo}/diff", get(api::diff))
        .route("/api/{authority}/{repo}/object/{sha}", get(api::object))
        // Public, and how a page elsewhere finds where the queries are.
        .route(
            "/.well-known/did.json",
            get(gateway::did_document).layer(viewer::cors()),
        )
        .route("/xrpc/dev.knotless.putDelegation", post(gateway::delegate))
        .route(
            "/xrpc/dev.knotless.getDelegation",
            get(gateway::await_delegation),
        )
        .route(gateway::METADATA, get(gateway::client_metadata))
        .route("/oauth/callback", get(gateway::callback))
        .route(gateway::BOT_METADATA, get(gateway::bot_metadata))
        .route("/oauth/bot", get(gateway::bot_login))
        .route("/oauth/bot-callback", get(gateway::bot_callback))
        .route(login::METADATA, get(login::client_metadata))
        .route("/oauth/login", post(login::login))
        .route("/oauth/app-login", get(login::app_login))
        .route("/oauth/app-token", post(login::app_token))
        .route(login::APP_RETURN, get(login::app_return))
        .route(
            "/.well-known/apple-app-site-association",
            get(login::app_association),
        )
        .route("/oauth/web-callback", get(login::callback))
        .route("/oauth/switch", post(login::switch))
        .route("/oauth/logout", post(login::logout))
        .route(
            "/xrpc/dev.knotless.pushRefs",
            post(gateway::push).layer(axum::extract::DefaultBodyLimit::max(PUSH_LIMIT)),
        )
        .route("/xrpc/dev.knotless.createCommit", post(gateway::commit))
        .route(
            "/xrpc/dev.knotless.putOrgMember",
            post(gateway::put_org_member),
        )
        .route(
            "/xrpc/dev.knotless.removeOrgMember",
            post(gateway::remove_org_member),
        )
        .route("/xrpc/dev.knotless.joinOrg", post(gateway::join_org))
        .route(
            "/xrpc/dev.knotless.createOrgRepo",
            post(gateway::create_org_repo),
        )
        .route("/xrpc/dev.knotless.leaveOrg", post(gateway::leave_org))
        .route(
            "/xrpc/com.atproto.moderation.createReport",
            post(moderation::create_report),
        )
        .route(
            "/xrpc/com.atproto.admin.updateSubjectStatus",
            post(moderation::update_subject_status),
        )
        .route(
            "/xrpc/com.atproto.admin.getSubjectStatus",
            get(moderation::get_subject_status),
        )
        .route(
            "/xrpc/dev.knotless.listReports",
            get(moderation::list_reports),
        )
        .route("/xrpc/dev.knotless.inviteRunner", post(invites::invite))
        .route("/xrpc/com.atproto.space.notifyWrite", post(spaces::written))
        .route(
            "/xrpc/com.atproto.space.notifySpaceDeleted",
            post(spaces::deleted),
        )
        .route("/xrpc/dev.knotless.listRunnerInvites", get(invites::list))
        .route("/xrpc/dev.knotless.putSecret", post(secrets::put))
        .route("/xrpc/dev.knotless.deleteSecret", post(secrets::delete))
        .route("/xrpc/dev.knotless.listSecrets", get(secrets::list))
        .route(
            "/xrpc/dev.knotless.putRunOutput",
            post(live::put_run_output).layer(axum::extract::DefaultBodyLimit::max(live::BODY)),
        )
        .route(
            "/xrpc/dev.knotless.awaitSpaceWrites",
            get(live::await_space_writes),
        )
        .route(
            "/xrpc/com.atproto.label.queryLabels",
            get(moderation::query_labels),
        )
        .route(
            "/xrpc/com.atproto.label.subscribeLabels",
            get(moderation::subscribe_labels),
        )
        .merge(queries)
        .merge(web::routes())
        .layer(axum::middleware::from_fn(web::xrpc_failures))
        .layer(axum::middleware::from_fn(web::origin))
        .with_state(app)
}

/// How long a check may run before the index's word for it is read
/// again from its PDS, and how often again after.
const RECHECK: Duration = Duration::from_secs(60);

/// The largest push a gateway takes.
const PUSH_LIMIT: usize = 512 << 20;

/// `name` with a trailing `.git` dropped, as clone URLs often carry one.
fn repo_name(name: &str) -> &str {
    name.strip_suffix(".git").unwrap_or(name)
}

/// The arguments `knotless-serve` takes.
#[derive(Debug, PartialEq)]
pub struct Options {
    pub listen: String,
    pub ttl: Duration,
    pub firehose: Option<String>,
    /// A Jetstream v2 host, followed like the firehose.
    pub jetstream: Option<String>,
    /// The URL this server is reached at, which makes it a gateway.
    pub public_url: Option<String>,
    /// Where to keep the appview index, which makes it an appview.
    pub index: Option<std::path::PathBuf>,
    /// A PDS or relay to index every account of at startup.
    pub backfill: Option<String>,
    /// A relay to index the accounts holding knotless records from when
    /// the index starts empty, so a new version of its tables loses
    /// nothing older than what Jetstream replays.
    pub relay: Option<String>,
    /// Where a backfill finds accounts and reads them first: directories,
    /// and a mirror.
    pub sources: Sources,
    /// Labelers, by DID, whose takedowns hold here too.
    pub labelers: Vec<String>,
    /// The runner, by DID, a new public repository names unless its
    /// owner names others: one, as every runner named runs each check.
    pub default_runner: Option<String>,
    /// The oldest client version to ask for, when one harms what it
    /// touches, and why: clients older refuse to run until upgraded.
    pub client_minimum: Option<String>,
    pub client_minimum_reason: Option<String>,
    /// The oldest build of the iOS app to let run, once a newer one is in
    /// people's reach, and why: builds older show only a way to update.
    pub ios_minimum_build: Option<u64>,
    pub ios_minimum_reason: Option<String>,
    /// The SvelteKit app that draws the web pages, which this server hands
    /// every request it does not answer itself.
    pub web: Option<String>,
    /// Where the client's builds are, a directory or a repository's
    /// releases, handed out under `/releases` with `/install.sh` to fetch
    /// one.
    pub releases: Option<releases::Source>,
    /// How long a feed may stay silent before `/_status` calls it dead.
    pub stale: Option<Duration>,
    /// Where to keep the counts of how this server is used.
    pub usage: Option<std::path::PathBuf>,
    /// Where to keep the lines it logs, and how many at info and below.
    pub logs: Option<std::path::PathBuf>,
    pub logs_keep: u64,
    /// Only check that the index at this path is of this build's tables,
    /// so that starting would keep it rather than rebuild it.
    pub check_index: Option<std::path::PathBuf>,
    /// Where `/metrics` and `/log` are answered, apart from the rest.
    pub metrics: Option<String>,
    /// Where a gateway keeps the sessions its bots log in to.
    pub bots: Option<std::path::PathBuf>,
    /// The PDS where the pages send someone making an account:
    /// [`gateway::SIGN_UP_PDS`] unless `--sign-up-pds` names another.
    pub sign_up_pds: String,
    /// Where the PDS sends a gateway's OAuth logins back to, when not
    /// `--public-url`: an https front for it, as the PDS asks of a client.
    pub oauth_url: Option<String>,
    /// Where code search keeps its Zoekt shards, if it is on.
    pub code_index: Option<std::path::PathBuf>,
    /// The domain repositories' sites are answered on, each owner at a
    /// label of it, as `knotless.page`: none answers none.
    pub pages_domain: Option<String>,
    /// The key Apple gave for its push service, a `.p8` file, which with
    /// its id and team tells devices of their accounts' notifications.
    pub apns_key: Option<std::path::PathBuf>,
    pub apns_key_id: Option<String>,
    pub apns_team: Option<String>,
}

/// What `--private`, which is gone, says instead.
pub const PRIVATE_GONE: &str = "--private is gone: knotless-serve shows a private repository to each member who logs in to its pages (--public-url), read as them, so run it without --private and log in, or browse https://knotless.dev logged in";

pub fn parse_args(args: &[String]) -> Result<Options, String> {
    let mut opts = Options {
        listen: "127.0.0.1:2590".into(),
        ttl: Duration::from_secs(5),
        firehose: None,
        jetstream: None,
        public_url: None,
        index: None,
        backfill: None,
        relay: None,
        sources: Sources::default(),
        labelers: Vec::new(),
        default_runner: None,
        client_minimum: None,
        client_minimum_reason: None,
        ios_minimum_build: None,
        ios_minimum_reason: None,
        web: None,
        releases: None,
        stale: None,
        usage: None,
        logs: None,
        logs_keep: logs::KEEP,
        check_index: None,
        metrics: None,
        bots: None,
        sign_up_pds: gateway::SIGN_UP_PDS.into(),
        oauth_url: None,
        code_index: None,
        pages_domain: None,
        apns_key: None,
        apns_key_id: None,
        apns_team: None,
    };
    let mut it = args.iter().skip(1);
    while let Some(flag) = it.next() {
        let mut value = || it.next().cloned().ok_or(format!("{flag} takes a value"));
        match flag.as_str() {
            "--listen" => opts.listen = value()?,
            "--ttl" => {
                opts.ttl = Duration::from_secs(
                    value()?
                        .parse()
                        .map_err(|_| "--ttl takes seconds".to_string())?,
                )
            }
            // An empty stream or backfill is none, so a setting can turn
            // one off: `--firehose "$KNOTLESS_FIREHOSE"`, left unset.
            "--firehose" => opts.firehose = Some(value()?).filter(|v| !v.is_empty()),
            "--jetstream" => opts.jetstream = Some(value()?).filter(|v| !v.is_empty()),
            "--public-url" => opts.public_url = Some(value()?),
            "--pages-domain" => opts.pages_domain = Some(value()?).filter(|v| !v.is_empty()),
            "--apns-key" => {
                opts.apns_key = Some(value()?).filter(|v| !v.is_empty()).map(Into::into)
            }
            "--apns-key-id" => opts.apns_key_id = Some(value()?).filter(|v| !v.is_empty()),
            "--apns-team" => opts.apns_team = Some(value()?).filter(|v| !v.is_empty()),
            "--index" => opts.index = Some(value()?.into()),
            "--backfill" => opts.backfill = Some(value()?).filter(|v| !v.is_empty()),
            "--relay" => opts.relay = Some(value()?),
            "--directory" => opts
                .sources
                .directories
                .extend(Some(value()?).filter(|v| !v.is_empty())),
            "--mirror" => opts.sources.mirror = Some(value()?).filter(|v| !v.is_empty()),
            "--labeler" => opts.labelers.push(value()?),
            "--default-runner" => {
                let did = value()?;
                if !did.is_empty() && !did.starts_with("did:") {
                    return Err(format!("--default-runner takes a DID, not {did}"));
                }
                if opts.default_runner.is_some() {
                    return Err("--default-runner names one runner".into());
                }
                opts.default_runner = Some(did).filter(|d| !d.is_empty());
            }
            "--client-minimum" => {
                let version = value()?;
                if !version.is_empty() {
                    knotless::release::version(&version).map_err(|e| e.to_string())?;
                }
                opts.client_minimum = Some(version).filter(|v| !v.is_empty());
            }
            "--client-minimum-reason" => {
                opts.client_minimum_reason = Some(value()?).filter(|v| !v.is_empty())
            }
            "--ios-minimum-build" => {
                let build = value()?;
                let number = build.parse::<u64>();
                let number = number.map_err(|_| "--ios-minimum-build takes a build number");
                opts.ios_minimum_build = match build.is_empty() {
                    true => None,
                    false => Some(number?),
                };
            }
            "--ios-minimum-reason" => {
                opts.ios_minimum_reason = Some(value()?).filter(|v| !v.is_empty())
            }
            // Gone: the pages read private repositories as whoever logs
            // in to them.
            "--private" => return Err(PRIVATE_GONE.into()),
            "--web" => opts.web = Some(value()?),
            "--releases" => opts.releases = Some(releases::Source::parse(&value()?)?),
            "--stale" => {
                let secs = value()?.parse();
                let secs = secs.map_err(|_| "--stale takes seconds".to_string())?;
                opts.stale = Some(Duration::from_secs(secs));
            }
            "--usage" => opts.usage = Some(value()?.into()),
            "--logs" => opts.logs = Some(value()?.into()),
            "--logs-keep" => {
                let n = value()?.parse();
                opts.logs_keep = n.map_err(|_| "--logs-keep takes a count of lines".to_string())?;
            }
            "--check-index" => opts.check_index = Some(value()?.into()),
            "--metrics" => opts.metrics = Some(value()?),
            "--bots" => opts.bots = Some(value()?.into()),
            "--sign-up-pds" => opts.sign_up_pds = value()?,
            "--oauth-url" => opts.oauth_url = Some(value()?),
            "--code-index" => opts.code_index = Some(value()?.into()),
            other => {
                return Err(format!(
                    "unexpected {other}\nusage: knotless-serve [--listen <addr>] [--ttl <secs>] [--firehose <wss://relay or pds>] [--jetstream <wss://host>] [--public-url <https://host>] [--oauth-url <https://host>] [--index <file>] [--backfill <https://pds or relay>] [--relay <https://relay>] [--labeler <did>]... [--default-runner <did>] [--client-minimum <version> [--client-minimum-reason <text>]] [--ios-minimum-build <build> [--ios-minimum-reason <text>]] [--web <http://app>] [--releases <dir or knotless://repo>] [--stale <secs>] [--usage <file>] [--logs <file> [--logs-keep <lines>]] [--check-index <file>] [--metrics <addr>] [--bots <dir>] [--sign-up-pds <https://pds>] [--code-index <dir>] [--apns-key <file> --apns-key-id <id> --apns-team <id>] [--pages-domain <domain>]"
                ));
            }
        }
    }
    Ok(opts)
}

/// Whether the index at `path` is missing or of this build's tables: an
/// error when starting with it would rebuild it.
fn check_index(path: &std::path::Path) -> Result<()> {
    let found = index::version_at(path)?;
    if let Some(v) = found.filter(|v| *v != index::VERSION) {
        anyhow::bail!(
            "{}: its tables are version {v}, this build's {}: starting would rebuild it",
            path.display(),
            index::VERSION
        );
    }
    println!("{}: this build keeps it", path.display());
    Ok(())
}

/// Serves until SIGTERM or SIGINT. Prints the bound address first.
pub async fn run(opts: Options, ctx: Ctx) -> Result<()> {
    // Asked by deploy/deploy.sh before starting this build beside a running
    // one: both have the index open while one hands over to the other,
    // which is safe only if this one would not rebuild it.
    if let Some(path) = opts.check_index {
        return check_index(&path);
    }
    if opts.public_url.is_none() && opts.bots.is_some() {
        anyhow::bail!("--bots takes --public-url, where bots log in");
    }
    if opts.public_url.is_none() && opts.pages_domain.is_some() {
        anyhow::bail!(
            "--pages-domain takes --public-url, where a site's owner is sent from its root"
        );
    }
    let apns = match &opts.apns_key {
        Some(path) => {
            let (key_id, team) = (opts.apns_key_id.as_deref(), opts.apns_team.as_deref());
            Some(web::push::Apns::open(path, key_id, team)?)
        }
        None => None,
    };
    let gateway = match &opts.public_url {
        Some(url) => {
            let mut gw = gateway::Gateway::new(url).map_err(anyhow::Error::msg)?;
            gw.bots = opts.bots.clone();
            gw.sign_up_pds = opts.sign_up_pds.clone();
            if let Some(url) = &opts.oauth_url {
                gw.oauth_url = url.trim_end_matches('/').to_string();
            }
            Some(gw)
        }
        None => None,
    };
    let opening = Instant::now();
    let index = match &opts.index {
        Some(path) => Some(index::Index::open(path)?),
        None => None,
    };
    telemetry::opened(index.as_ref(), opening.elapsed());
    let usage = match &opts.usage {
        Some(path) => Some(Arc::new(usage::Usage::open(path)?)),
        None => None,
    };
    let logs = match &opts.logs {
        Some(path) => Some(Arc::new(logs::Logs::open(path, opts.logs_keep)?)),
        None => None,
    };
    // Every line logged from here on, kept to read back.
    if let Some(logs) = logs.clone() {
        knotless::telemetry::keep_lines(move |line| logs.keep(line));
    }
    // The index's slow statements are counted in it.
    usage.iter().for_each(usage::sink);
    // MCP over HTTP, for any account not taken down, made once
    // the address it reads this server's index at is known.
    let mcp_ctx = ctx.clone();
    let code = match &opts.code_index {
        Some(dir) => Some(Arc::new(code::Code::start(dir).await?)),
        None => None,
    };
    let mut app = App::build(ctx, opts.ttl, gateway, index);
    let reason = opts.client_minimum_reason.clone();
    let client_minimum = opts.client_minimum.clone().map(|m| (m, reason));
    let reason = opts.ios_minimum_reason.clone();
    let ios_minimum = opts.ios_minimum_build.map(|b| (b, reason));
    if let Some(app) = Arc::get_mut(&mut app) {
        app.stale = opts.stale;
        app.disk = Some(app.ctx.env.cache_dir.clone());
        app.usage = usage.clone();
        app.logs = logs.clone();
        app.default_runner = opts.default_runner.clone();
        app.client_minimum = client_minimum;
        app.ios_minimum = ios_minimum;
        app.code = code;
    }
    // An index built anew is filled from the relay, unless asked otherwise.
    let rebuilt = app.index.as_ref().is_some_and(|i| i.rebuilt);
    let backfill = opts.backfill.or(opts.relay.clone().filter(|_| rebuilt));
    if let (Some(service), Some(index)) = (backfill, app.index.clone()) {
        let ctx = app.ctx.clone();
        let sources = opts.sources.clone();
        tokio::spawn(async move { telemetry::sweep(&ctx, &index, &service, &sources).await });
    }
    // And again now and then, for what the feeds missed.
    if let (Some(relay), Some(index)) = (opts.relay, app.index.clone()) {
        tokio::spawn(sweep(app.ctx.clone(), index, relay, opts.sources, SWEEP));
    }
    for did in opts.labelers {
        tokio::spawn(moderation::follow(app.clone(), did));
    }
    if let Some(url) = opts.firehose {
        app.watching.store(true, Ordering::Relaxed);
        tokio::spawn(firehose::watch(app.clone(), url));
    }
    if let Some(url) = opts.jetstream {
        app.watching.store(true, Ordering::Relaxed);
        tokio::spawn(jetstream::watch(app.clone(), url));
    }
    if let (Some(apns), Some(index)) = (apns, app.index.clone()) {
        tokio::spawn(web::push::follow(app.clone(), index, apns, PUSH_ROUND));
    }
    tokio::spawn(spaces::follow(app.clone(), SPACES));
    tokio::spawn(languages::follow(app.clone()));
    if let Some(follow) = code::follow(app.clone()) {
        tokio::spawn(follow);
    }
    tokio::spawn(shared::follow(app.clone()));
    tokio::spawn(telemetry::summaries(app.clone(), SUMMARY));
    let listener = tokio::net::TcpListener::bind(&opts.listen).await?;
    let addr = listener.local_addr()?;
    let listening = format!("http://{addr}");
    println!("listening on {listening}");
    telemetry::said("listening", Some(&listening));
    // The numbers and the log's level, on a port of their own, kept off
    // the proxy: printed second, after where the pages are answered.
    telemetry::serve_admin(opts.metrics.as_deref(), app.clone()).await?;
    // With no pages to hand on to, what this server does not answer is not found.
    let pages = opts.web.map(web::pages).unwrap_or_default();
    let index = app.index.clone();
    let mcp = opts
        .public_url
        .as_deref()
        .map(|url| mcp_routes(mcp_ctx, url, index, addr, usage.clone()));
    let mcp = mcp.transpose()?.unwrap_or_default();
    let registry = oci::routes(app.ctx.clone(), opts.public_url.clone());
    let home = opts.public_url.clone().unwrap_or_default();
    let releases = releases::routes(opts.releases, opts.public_url, app.ctx.clone());
    let stopping = app.live.stopping.clone();
    let sites = opts
        .pages_domain
        .map(|d| sites::Sites::new(app.clone(), d, home));
    let mut routes = router(app)
        .merge(releases)
        .merge(registry)
        .merge(mcp)
        .merge(pages);
    if let Some(sites) = sites {
        routes = sites::hosted(routes, sites);
    }
    if let Some(usage) = usage.clone() {
        tokio::spawn(usage::flushing(usage.clone(), usage::FLUSH));
        routes = usage::counted(routes, usage);
    }
    if let Some(logs) = logs.clone() {
        tokio::spawn(logs::flushing(logs, logs::FLUSH));
    }
    axum::serve(listener, telemetry::observed(routes))
        .with_graceful_shutdown(async move {
            let mut term =
                tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
                    .expect("installing a SIGTERM handler");
            tokio::select! {
                _ = term.recv() => {}
                _ = tokio::signal::ctrl_c() => {}
            }
            telemetry::said("stopping: finishing the requests it has", None);
            // What the pages follow ends, for them to follow the next build.
            stopping.cancel();
        })
        .await?;
    // What was counted since the last write.
    usage.as_deref().map_or(Ok(()), usage::Usage::flush)?;
    telemetry::said("stopped", None);
    // The lines logged since the last write, that one with them.
    logs.as_deref().map_or(Ok(()), logs::Logs::flush)?;
    Ok(())
}

/// MCP over HTTP at `url`, for any account `index` has not taken down.
fn mcp_routes(
    mut ctx: Ctx,
    url: &str,
    index: Option<Arc<index::Index>>,
    addr: SocketAddr,
    usage: Option<Arc<usage::Usage>>,
) -> Result<Router> {
    ctx.env.appview = mcp_appview(ctx.env.appview, index.is_some(), addr);
    let limits = knotless_mcp::http::Limits::default();
    let mcp = knotless_mcp::http::Mcp::new(&ctx, url, limits)?;
    if let Some(usage) = usage {
        mcp.observe(usage::mcp_events(usage));
    }
    if let Some(index) = index {
        mcp.refuse(taken_down(index));
    }
    Ok(knotless_mcp::http::router(mcp))
}

/// Whether an account is taken down here, or by a labeler this server
/// follows: one the index cannot answer for is let in.
fn taken_down(index: Arc<index::Index>) -> knotless_mcp::http::Refused {
    Arc::new(move |did| index.down(&[did]).unwrap_or(false))
}

/// Where MCP over HTTP finds pulls and issues on public repositories: the
/// appview named, or else this server itself when it keeps the index, at
/// `addr` on this machine. An index of its own is read before the default
/// appview, knotless.dev's, which only an unset KNOTLESS_APPVIEW gives.
fn mcp_appview(named: Option<String>, indexed: bool, addr: SocketAddr) -> Option<String> {
    let default = named.as_deref() == Some(knotless::env::DEFAULT_APPVIEW);
    let named = named.filter(|_| !(indexed && default));
    named.or_else(|| {
        indexed.then(|| {
            let ip = match addr.ip() {
                IpAddr::V4(ip) if ip.is_unspecified() => IpAddr::V4(Ipv4Addr::LOCALHOST),
                IpAddr::V6(ip) if ip.is_unspecified() => IpAddr::V6(Ipv6Addr::LOCALHOST),
                ip => ip,
            };
            format!("http://{}", SocketAddr::new(ip, addr.port()))
        })
    })
}

#[cfg(test)]
mod tests {

    #[test]
    fn a_record_of_one_repository_is_its_alone() {
        let op = |collection: &str, rkey: &str| firehose::Op {
            action: "create".into(),
            collection: collection.into(),
            rkey: rkey.into(),
            record: None,
        };
        use knotless::records::nsid;
        // A push's pack, claim and ref name one repository, once.
        let push = [
            op(nsid::PACK, "k.3kaaa"),
            op(nsid::CLAIM, "k.m.1"),
            op(nsid::REF, "k.m"),
        ];
        assert_eq!(scopes(&push), [Some("k".to_string())]);
        // A repository's own record, or one under no key, is of them all.
        let both = [op(nsid::REF, "j.m"), op(nsid::REPO, "name")];
        assert_eq!(scopes(&both), [None, Some("j".to_string())]);
        assert_eq!(scopes(&[op(nsid::REF, "bare")]), [None]);
        assert_eq!(scopes(&[]), [None]);
    }

    #[test]
    fn a_stale_mirror_is_served_while_it_syncs() {
        let long_ago = Instant::now().checked_sub(STALE_FOR * 2).unwrap();
        let mut entry = Entry {
            events: (1, 1),
            ..Entry::default()
        };
        // Quiet, it is served however long ago it synced.
        entry.synced = Some(long_ago);
        assert!(serves(&entry, (1, 1), 0));
        // Stale with no event noted, or the first long ago, it is not.
        assert!(!serves(&entry, (2, 1), 0));
        entry.stale_since = Some(long_ago);
        assert!(!serves(&entry, (2, 1), 0));
        // The first event only now, it is, however long ago it synced.
        entry.stale_since = Some(Instant::now());
        assert!(serves(&entry, (2, 1), 0));
        // A write from the pages it may not have seen waits for the sync.
        assert!(!serves(&entry, (2, 1), 1));
        assert!(!serves(&entry, (1, 1), 1));
    }

    #[test]
    fn one_commit_is_one_touch() {
        let seen = Mutex::default();
        assert!(first_of(&seen, "did:a", Some("r1")));
        // The rest of the commit's records, each an event of its own.
        assert!(!first_of(&seen, "did:a", Some("r1")));
        assert!(!first_of(&seen, "did:a", Some("r1")));
        assert!(first_of(&seen, "did:b", Some("r1")));
        assert!(first_of(&seen, "did:a", Some("r2")));
        assert!(first_of(&seen, "did:a", None));
        assert!(first_of(&seen, "did:a", None));
    }
    use super::*;

    #[test]
    fn mcp_reads_its_own_index() {
        let at = |s: &str| s.parse::<SocketAddr>().unwrap();
        let named = Some("https://appview.example".to_string());
        assert_eq!(mcp_appview(named.clone(), true, at("0.0.0.0:2590")), named);
        assert_eq!(mcp_appview(None, false, at("0.0.0.0:2590")), None);
        // The default appview gives way to an index of its own.
        let default = Some(knotless::env::DEFAULT_APPVIEW.to_string());
        let unindexed = mcp_appview(default.clone(), false, at("0.0.0.0:2590"));
        assert_eq!(unindexed, default);
        let indexed = mcp_appview(default, true, at("10.0.0.2:2590"));
        assert_eq!(indexed.as_deref(), Some("http://10.0.0.2:2590"));
        let own = |addr| mcp_appview(None, true, at(addr)).unwrap();
        assert_eq!(own("0.0.0.0:2590"), "http://127.0.0.1:2590");
        assert_eq!(own("[::]:2590"), "http://[::1]:2590");
        assert_eq!(own("10.0.0.2:2590"), "http://10.0.0.2:2590");
    }

    #[test]
    fn taken_down_accounts_are_refused() {
        let dir = tempfile::tempdir().unwrap();
        let index = Arc::new(index::Index::open(&dir.path().join("index.db")).unwrap());
        let refused = taken_down(index.clone());
        assert!(!refused("did:plc:a"));
        let subject = serde_json::json!({"did": "did:plc:a"});
        index
            .take_down("did:web:me", "did:plc:a", &subject, true)
            .unwrap();
        assert!(refused("did:plc:a"));
        assert!(!refused("did:plc:b"));
    }

    #[test]
    fn input_errors() {
        let bad = AppError::input(knotless::read::BadInput("no".into()).into());
        assert_eq!(bad.0, StatusCode::BAD_REQUEST);
        let broken = AppError::input(anyhow::anyhow!("disk on fire"));
        assert_eq!(broken.0, StatusCode::INTERNAL_SERVER_ERROR);
    }

    #[test]
    fn upstream_errors() {
        let nobody = knotless::identity::NoSuchHandle("alice.test".into());
        assert_eq!(AppError::upstream(nobody.into()).0, StatusCode::NOT_FOUND);
        let down = AppError::upstream(anyhow::anyhow!("resolver down"));
        assert_eq!(down.0, StatusCode::BAD_GATEWAY);
    }

    // A collection a backfill reads and the firehose passes by is indexed
    // once and never again: each list names it, or the index goes stale.
    // A rebuilt index must still know what was yanked, or each yanked
    // version reads as live again.
    #[test]
    fn a_backfill_reads_yanks() {
        let yanks = knotless::records::nsid::RELEASE_YANK;
        assert!(indexed_collections().any(|c| c == yanks));
    }

    #[test]
    fn the_firehose_keeps_every_indexed_collection() {
        for c in indexed_collections() {
            let kept = knotless::jetstream::knotless(c);
            assert!(kept, "{c} is backfilled but not kept live");
        }
    }
}