didbot_serve/routes/

mod.rs

//! The router: every route the development server answers.

use std::sync::Arc;

use axum::body::Bytes;
use axum::extract::rejection::QueryRejection;
use axum::extract::ws::{close_code, CloseFrame, Message, WebSocket, WebSocketUpgrade};
use axum::extract::{Query, RawQuery, State};
use axum::http::{header, HeaderMap, HeaderName, HeaderValue, Method, StatusCode};
use axum::response::{IntoResponse, Response};
use axum::routing::{get, post};
use axum::{Json, Router};
use didbot_identity::{canonical_did, ATPROTO_DID_PATH};
use didbot_pds::{
    Actor, Delta, Estop, EstopRefusal, Lock, Precondition, ProvisionError, Registry,
    ServerLifecycle, ServerPolicy,
};
use serde::de::DeserializeOwned;
use serde_json::json;
use tower_http::cors::{Any, CorsLayer};
use tower_http::timeout::TimeoutLayer;
use tower_http::trace::{DefaultOnRequest, DefaultOnResponse, MakeSpan, TraceLayer};
use tower_http::LatencyUnit;
use tracing::{debug, info, warn, Level};

use crate::auth;
use crate::error::ApiError;
use crate::health::HealthState;
use crate::oauth::scope::Action;
use crate::rate_limit::{caller_key, Peer, RateLimiter};
use crate::subscribe::Repos;
use crate::wire::{
    record_uri, refuse_skipped_validation, repeated, require_known_lexicon, AccountSummary,
    ApplyWritesRequest, CreateRecordRequest, CreateRecordResponse, DeleteRecordRequest,
    DescribeRepoQuery, DescribeRepoResponse, DescribeServerResponse, DidRequest, GetRecordQuery,
    GetRecordResponse, GetRepoQuery, LedgerQuery, ListRecordsQuery, ListReposQuery,
    PutRecordRequest, RecordView, RepoWrite, StatsResponse, SyncGetRecordQuery, WriteResult,
};

mod accounts;
mod identity;
mod oauth;
mod preferences;
mod repo;
mod session;
mod sync;

pub(crate) use accounts::*;
pub(crate) use identity::*;
pub(crate) use oauth::*;
pub(crate) use preferences::*;
pub(crate) use repo::*;
pub(crate) use session::*;
pub(crate) use sync::*;

/// Session storage, the e-stop latch and the operator credential, bundled so
/// a caller building [`app_with_auth`] hands them over together.
///
/// Separate from [`Registry`] on purpose: neither a session nor an e-stop
/// engagement is provisioning, and bundling them into the same object-safe
/// trait every backend must implement would grow that trait for every
/// deployment whether or not it uses a local stop. The
/// default is what every other builder here gets: fresh, empty, in-memory
/// state and no operator credential configured, so `app` keeps behaving
/// exactly as before this existed.
#[derive(Clone)]
pub struct AuthState {
    /// The local emergency stop.
    pub estop: Arc<Estop>,
    /// Signing in as this deployment's operator, when it has one. `None`
    /// when no `--operator` was given: there is no DID a sign-in could match,
    /// so the operator routes answer that they are not configured rather
    /// than offering a sign-in that can only fail. See
    /// [`crate::operator::OperatorAuth`].
    pub operator: Option<Arc<crate::operator::OperatorAuth>>,
    /// The relay an operator may announce this server to. `None` when no
    /// relay was named, and under a `.localhost` zone regardless; see
    /// [`crate::relay`].
    pub relay: Option<crate::relay::RelayTarget>,
    /// Bounds `GET /dashboard/login` per caller address.
    ///
    /// That route takes no credential — signing in is how a caller gets one
    /// — and answering it costs an identity resolution and a pushed
    /// authorization request against a host the *caller* named. That is the
    /// same shape as `POST /oauth/par`, which is already bounded this way,
    /// and rather more outbound work.
    pub operator_login_limiter: Arc<RateLimiter>,
    /// Which `Credential::Disclosure` routes this deployment has closed.
    /// Defaults to every route open; see `auth::Disclosure`.
    pub disclosure: auth::Disclosure,
    /// PAR, `authorize`, `token` and discovery; see [`crate::oauth::OAuthState`].
    /// Defaults to in-memory stores and every policy hook refusing — see
    /// that type's own doc.
    pub oauth: crate::oauth::OAuthState,
    /// Bounds `bot.did.createAccount` and `bot.did.createSession` calls per
    /// caller address, the two together; see this module's
    /// `PROVISION_RATE_LIMIT` and `crate::rate_limit`'s own doc for where
    /// that address comes from and [`AuthState::trust_forwarded_headers`] for
    /// when a caller-supplied header may stand in for it.
    ///
    /// `authorize` is keyed this way already;
    /// `createAccount` mints an account — a signing key, a hostname, a blob
    /// quota, a permanent ledger and firehose entry — and the claim
    /// `Credential::Attested` checks is verified only after this budget, so
    /// it needs the same address-keyed budget at least as much as either.
    /// Both routes read a proof, which can fetch an issuer's keys or a DID
    /// document before any signature is checked.
    pub provision_rate_limiter: Arc<RateLimiter>,
    /// Whether `bot.did.createAccount` with no credential creates beneath
    /// the deployment's own account; see `tree::prove`.
    ///
    /// `false` unless a deployment turns it on: a public deployment creates
    /// only for a creator that proved itself. `didbot-pds` turns it on
    /// under `.localhost`, where nothing holds a credential yet, and
    /// nowhere else.
    pub unauthenticated_creates: bool,
    /// How many accounts this deployment may hold before
    /// `bot.did.createAccount` refuses to mint another; see
    /// this module's `DEFAULT_MAX_ACCOUNTS` and `create_account`.
    ///
    /// Checked against [`didbot_pds::RegistryStats::accounts`] — the same
    /// count `crate::health`'s periodic line already reports — rather than a
    /// second counter kept here, so a cap and a dashboard can never disagree
    /// about what "how many accounts" means.
    pub max_accounts: u64,
    /// Whether this deployment answers only callers on this host.
    ///
    /// A `.localhost` zone is a development run: `createAccount` mints with
    /// no attestation claim there, and without a `--data` directory every
    /// blob is held in memory — so a run reachable from the network hands
    /// both to anybody sharing it. The listener binds every address because
    /// that is the only single bind answering both `127.0.0.1` and `::1`
    /// (see `didbot-pds`'s own comment), so the restriction is here: a
    /// request whose peer address is not a loopback address is refused
    /// before it reaches a route.
    ///
    /// `false` unless a deployment turns it on, which is the posture a real
    /// zone wants: a deployment nobody can reach serves no agents.
    pub loopback_callers_only: bool,
    /// The `Strict-Transport-Security` header every response carries, if
    /// any; see [`StrictTransportSecurity`].
    ///
    /// `None` unless a deployment turns it on. `didbot-pds` turns it on only
    /// when it serves a production ACME certificate.
    pub strict_transport_security: Option<StrictTransportSecurity>,
    /// Bounds how many never-ending streams — `com.atproto.sync.subscribeRepos`
    /// — this server holds open at once,
    /// in total and per caller address; see
    /// [`crate::subscribers::SubscriberLimit`].
    ///
    /// [`crate::MAX_CONCURRENT_REQUESTS`] does not cover these: its permit
    /// is released when the response head is sent, which for a stream is
    /// before the stream has carried a single frame.
    pub subscriber_limit: Arc<crate::subscribers::SubscriberLimit>,
    /// Bounds `bot.did.parkKey` and `describeParkedKey` per caller address;
    /// see `tree::PARK_RATE_LIMIT`.
    pub park_rate_limiter: Arc<RateLimiter>,
    /// Bounds the `com.atproto.sync.*` reads per caller address; see
    /// `routes::SYNC_READ_LIMIT`. Those reads are unauthenticated and each
    /// can rebuild and sign a repository.
    pub sync_rate_limiter: Arc<RateLimiter>,
    /// Bounds each account's record writes and blob uploads, by operation
    /// and by request byte; see [`crate::write_meter`].
    pub write_meter: Arc<crate::write_meter::WriteMeter>,
    /// Bounds each account's `atproto-proxy` calls, and caches where their
    /// targets live; see [`crate::ProxyLimits`].
    pub proxy: Arc<crate::ProxyLimits>,
    /// The shortest `idleFor` a `neverWrote` account delete takes; see
    /// [`crate::NEVER_WROTE_MIN_IDLE`].
    pub never_wrote_min_idle: std::time::Duration,
    /// Parked keys, spent proofs and issuers' keys; see [`crate::tree`].
    pub tree: Arc<crate::tree::TreeState>,
    /// Whether `X-Forwarded-For`/`X-Real-IP` may be trusted as the caller's
    /// address for every per-address budget this router keeps —
    /// [`AuthState::provision_rate_limiter`],
    /// [`crate::oauth::OAuthState::par_rate_limiter`] and
    /// [`crate::oauth::OAuthState::authorize_rate_limiter`] — instead of the
    /// TCP peer address `crate::rate_limit::caller_key` gets by default.
    ///
    /// `false` unless a deployment turns it on — see `crate::rate_limit`'s
    /// own doc for why trusting a caller-supplied header by default would
    /// make every per-address budget bypassable rather than merely
    /// degraded. Turn this on only once a real reverse proxy sits in front
    /// of this server and sets whichever header it reads.
    pub trust_forwarded_headers: bool,
    /// Where this deployment is in its own lifecycle; see
    /// [`didbot_pds::ServerLifecycle`] and this module's own
    /// `require_lifecycle`, which is the single point every capability
    /// question goes through.
    ///
    /// Defaults to [`ServerLifecycle::claimed`] — a server that exists and
    /// works — because that is what every builder in this crate other than
    /// `didbot-pds`'s own path is: a test, or a `.localhost` run. A real
    /// deployment hands in one built by
    /// [`ServerLifecycle::booting`](didbot_pds::ServerLifecycle::booting) and
    /// driven by a store read and an operator poll.
    pub lifecycle: Arc<ServerLifecycle>,
    /// Where `bot.did.pollOperatorClaim` delivers its "look now" request;
    /// see [`crate::PollNudge`] and this module's own `poll_operator_claim`.
    ///
    /// Defaults to a channel with no poll task attached, which absorbs
    /// nudges and does nothing — the correct behaviour for a build with no
    /// operator repository to read.
    pub nudge: Arc<crate::PollNudge>,
    /// What the operator poll's last listing of the operator's repository
    /// read, for the dashboard's provenance panel. Shared with the poll,
    /// which is the only thing that writes to it; see
    /// [`crate::OperatorRecords`].
    pub operator_records: Arc<crate::OperatorRecords>,
    /// The operator's handle as the operator poll last verified it, for the
    /// homepage. Shared with the poll, which is the only thing that writes
    /// to it; see [`crate::OperatorHandle`].
    pub operator_handle: Arc<crate::OperatorHandle>,
    /// Where `com.atproto.identity.resolveHandle` looks up a handle no
    /// account here holds; see [`crate::handles`]. Defaults to the network,
    /// through Cloudflare's resolver.
    pub handles: Arc<dyn crate::handles::HandleDid>,
    /// Bounds those lookups per caller address; see
    /// `routes::HANDLE_LOOKUP_LIMIT`.
    pub handle_lookup_limiter: Arc<RateLimiter>,
    /// Bounds `com.atproto.identity.resolveDid`'s lookups of DIDs hosted
    /// elsewhere per caller address; see `routes::DID_LOOKUP_LIMIT`.
    pub did_lookup_limiter: Arc<RateLimiter>,
    /// The lookups this deployment's policies declare, shared with the
    /// policy poll, which is the only thing that writes to it; see
    /// [`crate::DeclaredLookups`].
    pub declared_lookups: Arc<crate::DeclaredLookups>,
}

impl Default for AuthState {
    fn default() -> Self {
        Self {
            estop: Arc::new(Estop::default()),
            operator: None,
            relay: None,
            operator_login_limiter: Arc::new(RateLimiter::new(
                OPERATOR_LOGIN_LIMIT,
                OPERATOR_LOGIN_WINDOW,
            )),
            disclosure: auth::Disclosure::default(),
            oauth: crate::oauth::OAuthState::default(),
            provision_rate_limiter: Arc::new(RateLimiter::new(
                PROVISION_RATE_LIMIT,
                PROVISION_RATE_WINDOW,
            )),
            max_accounts: DEFAULT_MAX_ACCOUNTS,
            subscriber_limit: Arc::new(crate::subscribers::SubscriberLimit::default()),
            park_rate_limiter: Arc::new(RateLimiter::new(
                crate::tree::PARK_RATE_LIMIT,
                crate::tree::PARK_RATE_WINDOW,
            )),
            sync_rate_limiter: Arc::new(RateLimiter::new(SYNC_READ_LIMIT, SYNC_READ_WINDOW)),
            write_meter: Arc::default(),
            proxy: Arc::new(crate::ProxyLimits::default()),
            never_wrote_min_idle: crate::NEVER_WROTE_MIN_IDLE,
            tree: Arc::new(crate::tree::TreeState::default()),
            unauthenticated_creates: false,
            loopback_callers_only: false,
            strict_transport_security: None,
            trust_forwarded_headers: false,
            lifecycle: Arc::new(ServerLifecycle::claimed()),
            nudge: Arc::new(crate::PollNudge::new()),
            operator_records: Arc::new(crate::OperatorRecords::new()),
            operator_handle: Arc::default(),
            handles: Arc::new(crate::handles::Network::default()),
            handle_lookup_limiter: Arc::new(RateLimiter::new(
                HANDLE_LOOKUP_LIMIT,
                HANDLE_LOOKUP_WINDOW,
            )),
            did_lookup_limiter: Arc::new(RateLimiter::new(DID_LOOKUP_LIMIT, DID_LOOKUP_WINDOW)),
            declared_lookups: Arc::new(crate::DeclaredLookups::new()),
        }
    }
}

/// Everything a handler needs.
///
/// `Registry` is deliberately an `Arc<dyn ..>`: the HTTP layer must not be
/// generic over the provisioner's three type parameters, or every handler
/// signature grows them.
#[derive(Clone)]
pub(crate) struct AppState {
    pub(crate) registry: Arc<dyn Registry>,
    repos: Repos,
    pub(crate) estop: Arc<Estop>,
    /// See [`AuthState::operator`].
    pub(crate) operator: Option<Arc<crate::operator::OperatorAuth>>,
    /// See [`AuthState::relay`].
    pub(crate) relay: Option<crate::relay::RelayTarget>,
    /// See [`AuthState::operator_login_limiter`].
    pub(crate) operator_login_limiter: Arc<RateLimiter>,
    /// Which `Credential::Disclosure` routes are closed; see
    /// `auth::Disclosure`.
    pub(crate) disclosure: auth::Disclosure,
    pub(crate) oauth: crate::oauth::OAuthState,
    /// See [`AuthState::provision_rate_limiter`].
    pub(crate) provision_rate_limiter: Arc<RateLimiter>,
    /// See [`AuthState::max_accounts`].
    pub(crate) max_accounts: u64,
    /// See [`AuthState::unauthenticated_creates`].
    pub(crate) unauthenticated_creates: bool,
    /// See [`AuthState::park_rate_limiter`].
    pub(crate) park_rate_limiter: Arc<RateLimiter>,
    /// See [`AuthState::sync_rate_limiter`].
    pub(crate) sync_rate_limiter: Arc<RateLimiter>,
    /// See [`AuthState::write_meter`].
    pub(crate) write_meter: Arc<crate::write_meter::WriteMeter>,
    /// See [`AuthState::proxy`].
    pub(crate) proxy: Arc<crate::ProxyLimits>,
    /// See [`AuthState::never_wrote_min_idle`].
    pub(crate) never_wrote_min_idle: std::time::Duration,
    /// See [`AuthState::tree`].
    pub(crate) tree: Arc<crate::tree::TreeState>,
    /// See [`AuthState::subscriber_limit`].
    pub(crate) subscriber_limit: Arc<crate::subscribers::SubscriberLimit>,
    /// See [`AuthState::trust_forwarded_headers`].
    pub(crate) trust_forwarded_headers: bool,
    pub(crate) health: Arc<HealthState>,
    /// See [`AuthState::lifecycle`] and [`require_lifecycle`].
    pub(crate) lifecycle: Arc<ServerLifecycle>,
    /// See [`AuthState::nudge`].
    pub(crate) nudge: Arc<crate::PollNudge>,
    /// See [`AuthState::operator_records`].
    pub(crate) operator_records: Arc<crate::OperatorRecords>,
    /// See [`AuthState::operator_handle`].
    pub(crate) operator_handle: Arc<crate::OperatorHandle>,
    /// See [`AuthState::handles`].
    pub(crate) handles: Arc<dyn crate::handles::HandleDid>,
    /// See [`AuthState::handle_lookup_limiter`].
    pub(crate) handle_lookup_limiter: Arc<RateLimiter>,
    /// See [`AuthState::did_lookup_limiter`].
    pub(crate) did_lookup_limiter: Arc<RateLimiter>,
    /// The one reader for records in repositories this deployment does not
    /// hold; see `crate::foreign`. Built here rather than carried on
    /// [`AuthState`] because the two facts it needs — which repositories
    /// are ours and which names we answer to — are the registry's, and the
    /// registry arrives with the router rather than with the auth state.
    pub(crate) foreign: Arc<didbot_lookup::ForeignRecords>,
    /// The two-phase read of the records a write's policies declared they
    /// read: this half prefetches, before the write reaches the store's
    /// synchronous section. See `crate::lookups`.
    pub(crate) lookups: Arc<crate::LookupReader>,
}

/// How many `bot.did.createAccount` and `bot.did.createSession` calls one
/// caller address may make per [`PROVISION_RATE_WINDOW`], the two together.
///
/// Sized against a legitimate caller rather than against a read route:
/// `crates/didbot-swarm`, the local agent simulator, defaults to 8 events per
/// second (`didbot-swarm --rate`, `scripts/dev-swarm.sh`) and spends most of
/// them on `Spawn` while its population is below target, all from the one
/// address a local run shares.
/// Sustained, that is up to 480 provisioning calls a minute from a single
/// caller with nothing wrong — the case `plan/pds-xrpc.md`'s "what \[rate
/// limits\] do to a swarm sharing one address" names. This limit clears that
/// with better than double the headroom, while still bounding what a caller
/// this route does not authenticate can spend of this deployment's signing
/// keys, Route53 record-set quota and blob storage per minute.
pub const PROVISION_RATE_LIMIT: u32 = 1000;

/// The window [`PROVISION_RATE_LIMIT`] is measured over.
pub const PROVISION_RATE_WINDOW: std::time::Duration = std::time::Duration::from_secs(60);

/// How many `GET /dashboard/login` calls one caller address may make per
/// [`OPERATOR_LOGIN_WINDOW`]; see [`AuthState::operator_login_limiter`].
///
/// A person signing in makes one, retries a handful of times at worst, and
/// is done. The number is small because this route is unauthenticated and
/// each call costs an outbound identity resolution plus a pushed
/// authorization request to a host the caller named.
pub const OPERATOR_LOGIN_LIMIT: u32 = 20;

/// The window [`OPERATOR_LOGIN_LIMIT`] is measured over.
pub const OPERATOR_LOGIN_WINDOW: std::time::Duration = std::time::Duration::from_secs(60);

/// How many `com.atproto.sync.*` reads one caller address may make per
/// [`SYNC_READ_WINDOW`]; see [`AuthState::sync_rate_limiter`].
///
/// **Reasonable range: 60 to 60000.** Each of these reads can rebuild and
/// sign a repository on the blocking pool, and none of them is
/// authenticated. Six hundred a minute is ten a second sustained from one
/// address, which clears a relay backfilling this deployment's repositories
/// and a `didbot-swarm` run sharing one address, and still bounds what a
/// stranger can spend of the pool.
pub const SYNC_READ_LIMIT: u32 = 600;

/// The window [`SYNC_READ_LIMIT`] is measured over.
pub const SYNC_READ_WINDOW: std::time::Duration = std::time::Duration::from_secs(60);

/// How many handles no account here holds one caller address may ask
/// `com.atproto.identity.resolveHandle` to look up per
/// [`HANDLE_LOOKUP_WINDOW`]; see [`AuthState::handle_lookup_limiter`].
///
/// **Reasonable range: 10 to 600.** Each one costs a DNS-over-HTTPS query
/// and can cost an HTTPS fetch from a host the caller named, and none is
/// authenticated. A browser client resolves one handle per sign-in. A
/// handle hosted here costs no lookup and spends nothing.
pub const HANDLE_LOOKUP_LIMIT: u32 = 60;

/// The window [`HANDLE_LOOKUP_LIMIT`] is measured over.
pub const HANDLE_LOOKUP_WINDOW: std::time::Duration = std::time::Duration::from_secs(60);

/// How many DIDs hosted elsewhere one caller address may ask
/// `com.atproto.identity.resolveDid` to fetch per [`DID_LOOKUP_WINDOW`]; see
/// [`AuthState::did_lookup_limiter`].
///
/// **Reasonable range: 10 to 600.** Each one costs an HTTPS fetch from a
/// host the caller named or from the PLC directory, and none is
/// authenticated. A client resolves a few DIDs as it opens a view. A DID
/// hosted here costs no lookup and spends nothing.
pub const DID_LOOKUP_LIMIT: u32 = 60;

/// The window [`DID_LOOKUP_LIMIT`] is measured over.
pub const DID_LOOKUP_WINDOW: std::time::Duration = std::time::Duration::from_secs(60);

/// How many `com.atproto.sync.*` reads may hold a blocking-pool thread at
/// once, across every caller.
///
/// The same bound `crate::blobs`'s `BLOB_READ_CONCURRENCY` puts on
/// `getBlob`, and for the same reason: these reads run on the pool session
/// creation, provisioning, deletion and firehose publishing also draw from,
/// and each can rebuild the Merkle search tree and sign a commit. Without a
/// bound, [`crate::MAX_CONCURRENT_REQUESTS`] is the only one, and the
/// deployment runs two async workers.
///
/// A caller that finds every permit taken waits for one, and
/// [`REQUEST_TIMEOUT`]'s layer answers `503` if that wait outlasts the
/// request.
static SYNC_READ_CONCURRENCY: tokio::sync::Semaphore = tokio::sync::Semaphore::const_new(32);

/// How many accounts a deployment may hold at once, absent `--max-accounts`
/// or a `[capacity]` section; see [`AuthState::max_accounts`].
///
/// `plan/capacity.md` computes the raw ceiling this has to stay under: two
/// Route53 record sets per account against the default 10,000-record quota
/// is five thousand accounts before AWS's own quota — not this server's —
/// refuses a write mid-provisioning. A thousand is well under half of that,
/// leaving headroom for the zone's fixed `_acme-challenge` overhead and for
/// the time a quota increase takes to land, and it is already generous for
/// the fleet this project's own README describes: an agent operator's own
/// instance, not a multi-tenant population.
pub const DEFAULT_MAX_ACCOUNTS: u64 = 1000;
/// How long any one request may take before this router gives up on it and
/// answers `503` with an empty body.
///
/// Applies to every route, not only the anonymous ones the rate limiters
/// single out: a slow backend or a stalled write should not be able to hold
/// a connection (and the concurrency slot it occupies) open indefinitely.
/// The body is empty rather than this crate's usual `{"error":..,"message":..}`
/// shape — `tower_http::timeout::TimeoutLayer` answers with a bare status and
/// a default body directly, without going through this router's handlers or
/// its error type at all, which is what makes it a backstop that keeps
/// working even if a handler itself is the thing hanging.
pub const REQUEST_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);

/// The XRPC methods this server serves: for each one, the handler behind it
/// **and the credential it accepts**.
///
/// One list producing three things — the routes themselves, the names of
/// them, and the credential table `crate::auth` resolves against. That is
/// the whole point of the macro: there is no way to write a `.route()` call
/// for an XRPC method except by adding an entry here, and an entry with no
/// `as <Credential>` is a syntax error rather than a route that quietly
/// defaults to open. The three lists cannot drift because there is only one
/// list.
///
/// Expands to `$methods` (the NSIDs, for the conformance harness),
/// `$credentials` (the NSID → `auth::Credential` table, which
/// `auth::ROUTE_CREDENTIALS` is assembled from) and `$router` (the
/// routes those names stand for).
macro_rules! xrpc_methods {
    (
        $(#[$methods_meta:meta])*
        $methods:ident,
        $(#[$credentials_meta:meta])*
        $credentials:ident,
        $router:ident,
        { $($nsid:literal => $verb:ident($handler:path) as $credential:ident),* $(,)? }
    ) => {
        $(#[$methods_meta])*
        pub const $methods: &[&str] = &[$($nsid),*];

        $(#[$credentials_meta])*
        pub const $credentials: &[(&str, auth::Credential)] =
            &[$(($nsid, auth::Credential::$credential)),*];

        /// The routes those names stand for.
        fn $router() -> Router<AppState> {
            Router::new()$(.route(concat!("/xrpc/", $nsid), $verb($handler)))*
        }
    };
}

xrpc_methods! {
    /// Every `com.atproto.*` method this router answers, in NSID order.
    ///
    /// Public because it is what
    /// `crates/didbot/tests/conformance/wire.rs` reads: every name
    /// here must be a method in the vendored upstream documents, so a
    /// route this server invents fails a test rather than shipping.
    ATPROTO_METHODS,
    /// The credential each of those methods accepts. See
    /// `auth::Credential` for what each kind means, and that module's
    /// "Why every public route is public" for why the read surface is
    /// `Public` on purpose: protocol-required for `com.atproto.sync.*` and
    /// `com.atproto.repo.*`'s reads, structurally required for
    /// `describeServer` and `resolveHandle`. The write half takes the agent
    /// token `createAccount` hands back, and `routes::write_record` /
    /// `routes::apply_writes` are where the authenticated DID is checked
    /// against the repository a write names. `getSession` takes the same
    /// credentials, and names the account they sign in as.
    ATPROTO_CREDENTIALS,
    atproto_routes,
    {
        "com.atproto.identity.resolveDid" => get(resolve_did) as Public,
        "com.atproto.identity.resolveHandle" => get(resolve_handle) as Public,
        "com.atproto.repo.applyWrites" => post(apply_writes) as AccountToken,
        "com.atproto.repo.createRecord" => post(create_record) as AccountToken,
        "com.atproto.repo.deleteRecord" => post(delete_record) as AccountToken,
        "com.atproto.repo.describeRepo" => get(describe_repo) as Public,
        "com.atproto.repo.getRecord" => get(get_record) as Public,
        "com.atproto.repo.listRecords" => get(list_records) as Public,
        "com.atproto.repo.putRecord" => post(put_record) as AccountToken,
        "com.atproto.repo.uploadBlob" => post(crate::blobs::upload_blob) as AccountToken,
        "com.atproto.server.describeServer" => get(describe_server) as Public,
        "com.atproto.server.getServiceAuth" => get(crate::service_auth::get_service_auth) as AccountToken,
        "com.atproto.server.getSession" => get(get_session) as AccountToken,
        "com.atproto.sync.getBlob" => get(crate::blobs::get_blob) as Public,
        "com.atproto.sync.getBlocks" => get(get_blocks) as Public,
        "com.atproto.sync.getLatestCommit" => get(get_latest_commit) as Public,
        "com.atproto.sync.getRecord" => get(sync_get_record) as Public,
        "com.atproto.sync.getRepo" => get(get_repo) as Public,
        "com.atproto.sync.getRepoStatus" => get(get_repo_status) as Public,
        "com.atproto.sync.listBlobs" => get(crate::blobs::list_blobs) as Public,
        "com.atproto.sync.listRepos" => get(list_repos) as Public,
        "com.atproto.sync.subscribeRepos" => get(subscribe_repos) as Public,
    }
}

xrpc_methods! {
    /// The `app.bsky.*` methods this server serves itself rather than
    /// proxies, as the reference PDS does: the account's preferences.
    ///
    /// Its own invocation for the reason [`BOT_DID_METHODS`] has one: the
    /// vendored documents are `com.atproto.*` alone.
    APP_BSKY_METHODS,
    /// The credential each of those methods accepts: the ones `getSession`
    /// takes.
    APP_BSKY_CREDENTIALS,
    app_bsky_routes,
    {
        "app.bsky.actor.getPreferences" => get(get_preferences) as AccountToken,
        "app.bsky.actor.putPreferences" => post(put_preferences) as AccountToken,
    }
}

xrpc_methods! {
    /// The `bot.did.*` methods this server serves — its own account admin
    /// surface, alongside the `com.atproto.*` one [`ATPROTO_METHODS`] names.
    ///
    /// Its own invocation rather than folded into the one above: these are
    /// not vendored anywhere, so `crates/didbot/tests/conformance/wire.rs`'s
    /// "every route is a vendored method" check must not see them.
    ///
    /// `bot.did.hardDeleteAccount` is absent on purpose. It frees a burned
    /// name, so self-service is exactly wrong for it. The operator reaches
    /// `Registry::hard_delete` through the dashboard's
    /// `POST /dashboard/api/accounts/{did}/delete` instead.
    BOT_DID_METHODS,
    /// The credential each of those methods accepts.
    ///
    /// `createAccount` is `Credential::Creator`: the credential names the
    /// account creating, in any of the three shapes `tree::prove` resolves,
    /// and `createSession` is `Credential::Proof`, the two of those that
    /// are not a session. `updateCredentials` is
    /// `Credential::KeyProofOrOperator`: a JWT one of the account's keys
    /// signed, or the operator session. Every
    /// account-lifecycle mutation names an account that already exists and
    /// so takes `Credential::AccountSelf` — an agent acting on
    /// *itself*, the only caller this server can authenticate for them. The
    /// four authorization-decision routes take it for the same reason from
    /// the other side: they name no account at all, and the account they act
    /// on is the one the presented token authenticates as, so a decision can
    /// only ever be read, approved or declined by the agent it was addressed
    /// to. `createBrowserSignIn` takes it the same way, and signs a browser in
    /// as that account alone.
    /// The four reads are `Credential::Disclosure`: public by
    /// default, narrowable per route, never widenable. `pollOperatorClaim`
    /// is `Public` because it confers nothing — see the `auth` module's doc.
    /// `parkKey` and `describeParkedKey` are `Public` for the same reason:
    /// a park mints nothing until the operator admits it from their own
    /// side, and both are bounded per address rather than by a credential.
    BOT_DID_CREDENTIALS,
    bot_did_routes,
    {
        "bot.did.activateAccount" => post(activate_account) as AccountSelf,
        "bot.did.approveAuthorization" => post(approve_authorization) as AccountSelf,
        "bot.did.deactivateAccount" => post(deactivate_account) as AccountSelf,
        "bot.did.declineAuthorization" => post(decline_authorization) as AccountSelf,
        "bot.did.deleteAccount" => post(delete_account) as AccountSelf,
        "bot.did.freezeAccount" => post(freeze_account) as AccountSelf,
        "bot.did.getAccountLedger" => get(get_account_ledger) as Disclosure,
        "bot.did.getAuthorization" => get(get_authorization) as AccountSelf,
        "bot.did.listAccountLedgers" => get(list_ledgers) as Disclosure,
        "bot.did.listAccounts" => get(list_accounts) as Disclosure,
        "bot.did.listPendingAuthorizations" => get(list_pending_authorizations) as AccountSelf,
        "bot.did.pollOperatorClaim" => post(poll_operator_claim) as Public,
        "bot.did.createAccount" => post(crate::tree::create_account) as Creator,
        "bot.did.createBrowserSignIn" => post(create_browser_sign_in) as AccountSelf,
        "bot.did.createSession" => post(crate::tree::create_session) as Proof,
        "bot.did.describeParkedKey" => get(crate::tree::describe_parked_key) as Public,
        "bot.did.parkKey" => post(crate::tree::park_key) as Public,
        "bot.did.stats" => get(stats) as Disclosure,
        "bot.did.unfreezeAccount" => post(unfreeze_account) as AccountSelf,
        "bot.did.updateCredentials" => post(crate::tree::update_credentials) as KeyProofOrOperator,
    }
}

/// The header a client may name the turn a request belongs to in.
///
/// A record crosses three processes — the hook that stamps it, the record
/// host that writes it, and this server that stores it — and each logs what it
/// did. Until this existed nothing joined those three lines together at a
/// finer grain than the agent, and an agent that has been working for an hour
/// has hundreds of lines under one DID.
///
/// It is logged and nothing else. Nothing here parses it, no behaviour depends
/// on it, and it never reaches a record: it is a harness-internal string, and
/// writing it into a repository anybody can read would publish an operational
/// detail permanently for the benefit of nobody reading it there.
///
/// The producing half is whatever client sends it, which is why the name and
/// the bound below are stated here rather than inferred from a caller.
pub const TURN_HEADER: &str = "quernstone-turn";

/// The longest turn identifier this server will put in a log line.
///
/// A sanity bound rather than an agreement: anything longer is dropped, so a
/// client cannot make an arbitrarily long log line out of a header.
pub const TURN_HEADER_MAX_LENGTH: usize = 256;

/// The request span, carrying the turn when the client named one.
///
/// A [`MakeSpan`] rather than a separate middleware because a span is already
/// being made here, and putting the field on it means every line emitted while
/// the request is being served carries it — including the refusals, which are
/// the lines somebody correlating is most often looking for.
#[derive(Debug, Clone, Copy, Default)]
struct TurnSpan;

impl<B> MakeSpan<B> for TurnSpan {
    fn make_span(&mut self, request: &axum::http::Request<B>) -> tracing::Span {
        let turn = request
            .headers()
            .get(TURN_HEADER)
            .and_then(|value| value.to_str().ok())
            .filter(|value| !value.is_empty() && value.chars().count() <= TURN_HEADER_MAX_LENGTH)
            .unwrap_or("-");
        // The same three fields `DefaultMakeSpan` puts on a request span, plus
        // the turn. Named "request" rather than "http.request" for the same
        // reason the rest of this log is short: it is read by a person
        // watching a terminal.
        //
        // The path, not the URI: `/oauth/authorize` and
        // `bot.did.getAuthorization` carry a pushed-request reference in their
        // query, and that reference redeems a pending sign-in. Logs are shipped
        // and copied like operational data, so nothing that spends a credential
        // goes in one.
        tracing::info_span!(
            "request",
            method = %request.method(),
            uri = %request.uri().path(),
            version = ?request.version(),
            turn = %turn,
        )
    }
}

/// Builds the router.
pub fn app(registry: Arc<dyn Registry>) -> Router {
    app_with_health(registry, HealthState::new())
}

/// Builds the router with the health pulse wired to an existing producer.
///
/// The e-stop and the operator credential stay the
/// [`AuthState::default`] every other builder but [`app_with_auth`] uses;
/// call that one directly to wire real ones alongside a real `health`.
pub fn app_with_health(registry: Arc<dyn Registry>, health: Arc<HealthState>) -> Router {
    app_with_auth(registry, AuthState::default(), health)
}

/// Builds the router with every collaborator supplied explicitly.
///
/// The full builder every other one in this module calls. A deployment that
/// wants a durable session store, a configured operator credential, or a
/// file-backed e-stop latch calls this one directly; everything else keeps
/// working exactly as it did before [`AuthState`] existed.
///
/// `health` should be [`HealthTick::state`](crate::HealthTick::state) from
/// the tick a caller has spawned for this registry. A router built any other
/// way here answers `/health`'s `progress` from a state nothing ever ticks.
pub fn app_with_auth(
    registry: Arc<dyn Registry>,
    auth: AuthState,
    health_state: Arc<HealthState>,
) -> Router {
    app_with_repos(registry, Repos::default(), auth, health_state)
}

/// Builds the router with `com.atproto.sync.subscribeRepos` attached too.
///
/// The full builder. `repos` must be the same [`Repos`] handed to
/// [`Provisioner::with_repo_sink`](didbot_pds::Provisioner::with_repo_sink),
/// or the method answers with a stream nothing publishes into — and, unlike
/// the other two streams, one whose sequence numbers do not survive a restart
/// either. [`app_with_auth`] builds one with an in-memory sequence, which is
/// the right thing for a test and the wrong thing for a deployment; see
/// [`Durable::sequence`](didbot_pds::Durable::sequence).
pub fn app_with_repos(
    registry: Arc<dyn Registry>,
    repos: Repos,
    auth: AuthState,
    health_state: Arc<HealthState>,
) -> Router {
    // Wired here rather than left to whoever built `AuthState`: a Revoke
    // must tear down outstanding credentials or releasing it resurrects them,
    // and every router this crate builds needs that true regardless of who
    // constructed its `Estop` and `OAuthState`. The durable issuer writes
    // each revocation to the log before this returns, so a Revoke outlives
    // the restart that follows it. See `Estop::set_revoke_hook`.
    let auth_loopback_only = auth.loopback_callers_only;
    let strict_transport_security = auth
        .strict_transport_security
        .map(|hsts| hsts.header_value());
    let nonces = auth.oauth.dpop.clone();
    let revoke_grants = auth.oauth.tokens.clone();
    let revoke_browsers = auth.oauth.browser_sessions.clone();
    let revoke_remembered = auth.oauth.decision_store.clone();
    auth.estop.set_revoke_hook(Arc::new(move || {
        revoke_grants.revoke_all();
        revoke_browsers.revoke_all();
        revoke_remembered.forget_all();
    }));
    // The other credential an account answers with: the agent token every
    // createAccount and createSession hands out. Ended when the latch
    // leaves Revoke rather than when it is thrown, because
    // `bot.did.freezeAccount` and `bot.did.deactivateAccount` are reached
    // with that token and are exempt from the stop — a stop must never
    // stand between a caller and restraining itself. The latch refuses the
    // token for everything else while it stands, and this is what keeps it
    // refused afterwards.
    let revoke_tokens = registry.clone();
    auth.estop.set_release_hook(Arc::new(move || {
        revoke_tokens.revoke_all_agent_tokens();
    }));
    // And the same wiring for a single account, at the other end of an
    // account's life. The registry holds the agent token store and revokes
    // through it directly; every other credential bound to a DID is here,
    // beside the router, so a deletion tells this closure and this closure
    // ends them. Wired here for the reason above: every router this crate
    // builds needs it true, and the stores it names are the ones `AuthState`
    // carries. See `Registry::set_account_revoke_hook`.
    //
    // A credential is anything that answers for the account: its OAuth
    // grants, the authorization codes one hop short of a grant, and the
    // browsers signed in as it. Its remembered approvals go with them. The
    // codes, the browsers and the approvals go whatever the grant store
    // answers.
    let deleted_tokens = auth.oauth.tokens.clone();
    let deleted_codes = auth.oauth.code_store.clone();
    let deleted_browsers = auth.oauth.browser_sessions.clone();
    let deleted_remembered = auth.oauth.decision_store.clone();
    registry.set_account_revoke_hook(Arc::new(move |did: &str| {
        let ended = deleted_tokens.revoke_account(did);
        deleted_codes.revoke_account(did);
        deleted_browsers.revoke_account(did);
        deleted_remembered.forget_account(did);
        ended
    }));
    let registry_for_foreign = registry.clone();
    let private_addresses = auth.tree.private_addresses();
    let foreign = crate::foreign::reader(registry_for_foreign, private_addresses);
    // The reader needs the registry to tell a repository this deployment
    // holds from a stranger's, so it cannot be built until the registry
    // is; the registry is told about it here, once, and both halves of the
    // read -- the route's prefetch and the write path's own -- share it.
    let lookups = Arc::new(crate::LookupReader::new(
        auth.declared_lookups.clone(),
        foreign.clone(),
        registry.clone(),
    ));
    registry.set_lookups(lookups.clone());
    let state = AppState {
        registry,
        repos,
        estop: auth.estop,
        operator: auth.operator,
        relay: auth.relay,
        operator_login_limiter: auth.operator_login_limiter,
        disclosure: auth.disclosure,
        oauth: auth.oauth,
        provision_rate_limiter: auth.provision_rate_limiter,
        max_accounts: auth.max_accounts,
        unauthenticated_creates: auth.unauthenticated_creates,
        park_rate_limiter: auth.park_rate_limiter,
        sync_rate_limiter: auth.sync_rate_limiter,
        write_meter: auth.write_meter,
        proxy: auth.proxy,
        never_wrote_min_idle: auth.never_wrote_min_idle,
        tree: auth.tree,
        subscriber_limit: auth.subscriber_limit,
        trust_forwarded_headers: auth.trust_forwarded_headers,
        health: health_state,
        lifecycle: auth.lifecycle,
        nudge: auth.nudge,
        operator_records: auth.operator_records,
        operator_handle: auth.operator_handle,
        handles: auth.handles,
        handle_lookup_limiter: auth.handle_lookup_limiter,
        did_lookup_limiter: auth.did_lookup_limiter,
        foreign,
        lookups,
    };

    // The routes a stranger's browser tab may call from its own origin: the
    // protocol surface, its discovery documents, and the two OAuth endpoints
    // a web client fetches rather than navigates to. `/oauth/authorize` and
    // `/oauth/browser` are pages a browser is sent to and the dashboard holds
    // a person's session — none answers cross-origin, and widening
    // this set is a security decision, not a convenience (see
    // `plan/web-launch.md`).
    let cross_origin = Router::new()
        .route("/xrpc/_health", get(xrpc_health))
        .route("/.well-known/did.json", get(did_json))
        .route(ATPROTO_DID_PATH, get(atproto_did))
        .route(
            "/.well-known/oauth-authorization-server",
            get(oauth_server_metadata),
        )
        .route(
            "/.well-known/oauth-protected-resource",
            get(oauth_protected_resource_metadata),
        )
        .route(
            "/oauth/par",
            post(oauth_par).layer(axum::middleware::from_fn(never_cached)),
        )
        .route(
            "/oauth/token",
            post(oauth_token).layer(axum::middleware::from_fn(never_cached)),
        )
        .merge(bot_did_routes())
        .merge(atproto_routes())
        .merge(app_bsky_routes())
        // Set here, before the layer, and not only on the outer router
        // below: axum layers a route's method-not-allowed slot along with
        // its handlers, but the outer `method_not_allowed_fallback` replaces
        // every slot still holding the default — layered or not — and a
        // preflight is an `OPTIONS` on a route that serves no `OPTIONS`, so
        // it is exactly that slot the layer has to be wrapped around.
        .method_not_allowed_fallback(wrong_method)
        .layer(cross_origin_layer());

    let loopback_callers_only = auth_loopback_only;
    Router::new()
        .route("/health", get(health))
        .route("/oauth/authorize", get(oauth_authorize))
        .route(crate::oauth::browser::PATH, get(oauth_browser))
        .merge(crate::home::routes())
        .merge(crate::dashboard::routes())
        .merge(crate::analytics::routes())
        .merge(cross_origin)
        // A path nothing serves, and a method nothing serves on a path that
        // exists, are both failures a client has to be able to read. axum's
        // own answers are an empty 404 and an empty 405, which are not the
        // shape every other failure here takes. The fallback is also where
        // an XRPC call this server does not serve is proxied; see
        // `crate::proxy`.
        .fallback(crate::proxy::fallback)
        .with_state(state)
        .method_not_allowed_fallback(wrong_method)
        // Requests are logged at INFO because this server exists to be watched.
        // The default for `TraceLayer` is DEBUG, which would mean turning on
        // DEBUG for everything else to see them.
        .layer(
            TraceLayer::new_for_http()
                .make_span_with(TurnSpan)
                .on_request(DefaultOnRequest::new().level(Level::DEBUG))
                .on_response(
                    DefaultOnResponse::new()
                        .level(Level::INFO)
                        .latency_unit(LatencyUnit::Millis),
                ),
        )
        // A concurrency bound is deliberately *not* layered on here. This
        // router is built and handed to `crate::serve_router` or
        // `crate::serve_tls`, which both apply that bound through
        // `crate::limited_make_service` instead — see
        // `crate::MAX_CONCURRENT_REQUESTS`'s own doc for why `Router::layer`
        // cannot be trusted with a stateful limiter like
        // `tower::limit::ConcurrencyLimitLayer`.
        .layer(TimeoutLayer::with_status_code(
            StatusCode::SERVICE_UNAVAILABLE,
            REQUEST_TIMEOUT,
        ))
        // The buffering bound, pinned rather than inherited; see
        // `MAX_REQUEST_BODY`. `uploadBlob` is unaffected because it takes the
        // body as a stream, which this layer does not reach.
        .layer(axum::extract::DefaultBodyLimit::max(MAX_REQUEST_BODY))
        // Outside the body limit and the timeout, so their refusals carry a
        // nonce too. See `hand_out_nonce`.
        .layer(axum::middleware::from_fn(move |request, next| {
            hand_out_nonce(nonces.clone(), request, next)
        }))
        // Outside the body limit, the request log and the timeout, so a
        // caller this deployment does not answer is refused before any of
        // them spends anything on it. `security_headers` below is the one
        // layer outside this one, so the refusal carries them too. See
        // `AuthState::loopback_callers_only`.
        .layer(axum::middleware::from_fn(move |request, next| {
            only_loopback_callers(loopback_callers_only, request, next)
        }))
        .layer(axum::middleware::from_fn(move |request, next| {
            security_headers(strict_transport_security.clone(), request, next)
        }))
}

/// A `Strict-Transport-Security` header; see `[tls] hsts_max_age_secs` and
/// `[tls] hsts_include_subdomains` in `didbot-config`.
///
/// `preload` is never sent. The preload list takes only registrable domains,
/// and getting off it takes months.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct StrictTransportSecurity {
    /// How long a browser keeps the header, in seconds.
    pub max_age_secs: u64,
    /// Whether the header covers every name below the one it was served on.
    pub include_subdomains: bool,
}

impl StrictTransportSecurity {
    /// Two years, the same as did.bot's own header.
    pub const DEFAULT_MAX_AGE_SECS: u64 = 63_072_000;

    /// The header's value.
    pub fn header_value(&self) -> HeaderValue {
        let value = if self.include_subdomains {
            format!("max-age={}; includeSubDomains", self.max_age_secs)
        } else {
            format!("max-age={}", self.max_age_secs)
        };
        HeaderValue::try_from(value).expect("digits and ASCII words are a valid header value")
    }
}

/// The headers every response this server sends carries.
///
/// This process terminates its own TLS and renders its own pages, so there is
/// nothing in front of it to add them: `/oauth/authorize` and `/dashboard`
/// are HTML a browser is handed directly. The set is the one a
/// layer can apply without knowing the route — a consent page that cannot be
/// framed, a content type that is believed rather than sniffed, and a request
/// that says nothing about where the reader came from.
///
/// Applied to every response rather than to the HTML routes alone: the JSON,
/// the dashboard's assets and a blob are each served with the content type
/// they actually are, and none of them is meant to be put in a frame either.
///
/// `frame-ancestors` is the whole of the content security policy. A
/// `script-src` here would have to admit the `new Function` and the
/// `WebAssembly` that `dashboard.js` loads its field animation through, and a
/// policy written to permit `eval` buys nothing; framing is what the consent
/// page needs refused. `X-Frame-Options` says the same thing again for a
/// browser that reads only that one.
///
/// A route that sets a policy of its own keeps it: `getBlob` sets a sandbox
/// (see [`crate::blobs`]), and `X-Frame-Options` still refuses framing there.
///
/// `no-referrer`, where the static site sends
/// `strict-origin-when-cross-origin`: `/oauth/authorize`'s own URL carries the
/// authorization request, and no page here has a reason to pass its address to
/// anything it links to.
///
/// `Strict-Transport-Security` is added when the deployment set one; see
/// [`AuthState::strict_transport_security`]. Every response goes out over
/// TLS, so it is never sent over plain HTTP.
async fn security_headers(
    strict_transport_security: Option<HeaderValue>,
    request: axum::extract::Request,
    next: axum::middleware::Next,
) -> Response {
    let mut response = next.run(request).await;
    let headers = response.headers_mut();
    if let Some(value) = strict_transport_security {
        headers.insert(header::STRICT_TRANSPORT_SECURITY, value);
    }
    headers.insert(header::X_FRAME_OPTIONS, HeaderValue::from_static("DENY"));
    if !headers.contains_key(header::CONTENT_SECURITY_POLICY) {
        headers.insert(
            header::CONTENT_SECURITY_POLICY,
            HeaderValue::from_static("frame-ancestors 'none'"),
        );
    }
    headers.insert(
        header::X_CONTENT_TYPE_OPTIONS,
        HeaderValue::from_static("nosniff"),
    );
    headers.insert(
        header::REFERRER_POLICY,
        HeaderValue::from_static("no-referrer"),
    );
    response
}

/// Refuses a caller that is not on this host, when this deployment answers
/// only callers that are; see [`AuthState::loopback_callers_only`].
///
/// A request that arrived with no peer address — a test driving the
/// [`Router`] directly, or a deployment that wired neither
/// [`crate::serve_router`] nor [`crate::serve_tls`] — is served, the same
/// posture `crate::rate_limit::caller_key` takes for the same absence: this
/// cannot tell such a caller apart from any other, and refusing every
/// request on a router nobody put a listener under would make the router
/// untestable rather than safe.
async fn only_loopback_callers(
    enabled: bool,
    request: axum::extract::Request,
    next: axum::middleware::Next,
) -> Response {
    if enabled {
        let peer = request
            .extensions()
            .get::<axum::extract::ConnectInfo<std::net::SocketAddr>>()
            .map(|axum::extract::ConnectInfo(addr)| addr.ip());
        if let Some(ip) = peer {
            if !ip.to_canonical().is_loopback() {
                info!(
                    %ip,
                    path = %request.uri().path(),
                    "refused: this deployment answers only callers on this host"
                );
                return ApiError::new(
                    StatusCode::FORBIDDEN,
                    "NotServedHere",
                    "this deployment answers only callers on the host it runs on",
                )
                .into_response();
            }
        }
    }
    next.run(request).await
}

/// Hands the client a DPoP nonce on the answer to any request that carried
/// a `DPoP` proof, success and refusal alike, unless the answer names one
/// already, as a nonce challenge does.
///
/// The atproto OAuth profile has a client refuse an answer to a DPoP request
/// that carries no `DPoP-Nonce`, and RFC 9449 §8.2 lets every answer carry
/// the nonce for the next request.
async fn hand_out_nonce(
    verifier: Arc<dyn crate::oauth::dpop_seam::DpopVerifier>,
    request: axum::extract::Request,
    next: axum::middleware::Next,
) -> Response {
    let carried = request.headers().contains_key(crate::oauth::dpop::DPOP);
    let mut response = next.run(request).await;
    if carried
        && !response
            .headers()
            .contains_key(crate::oauth::dpop::DPOP_NONCE)
    {
        if let Some(value) = verifier
            .nonce()
            .and_then(|nonce| HeaderValue::from_str(&nonce).ok())
        {
            response
                .headers_mut()
                .insert(crate::oauth::dpop::DPOP_NONCE, value);
        }
    }
    response
}

/// How long a browser may cache a preflight answer before asking again.
///
/// A day, which is what the reference PDS sets. The answer never varies by
/// caller, so there is nothing for a shorter cache to pick up sooner.
pub const CORS_MAX_AGE: std::time::Duration = std::time::Duration::from_secs(24 * 60 * 60);

/// The request headers an atproto client sends and a preflight must name.
///
/// `authorization` and `dpop` carry the two credentials, `content-type` is
/// on every procedure call, and `atproto-proxy` / `atproto-accept-labelers`
/// are what the official client adds to every request whether or not the
/// server it reaches acts on them. A header left off this list makes the
/// browser refuse the request before it is sent, so the list is the
/// client's, not this server's.
pub const CORS_ALLOWED_HEADERS: [HeaderName; 5] = [
    header::AUTHORIZATION,
    header::CONTENT_TYPE,
    crate::oauth::dpop::DPOP,
    HeaderName::from_static("atproto-accept-labelers"),
    HeaderName::from_static("atproto-proxy"),
];

/// The response headers a browser is allowed to show a script.
///
/// Anything not named here is stripped by the browser before the calling
/// page sees it, and both of these carry the next step of a flow: the
/// nonce a DPoP client must echo on its retry, and the challenge that says
/// a token was refused and why.
pub const CORS_EXPOSED_HEADERS: [HeaderName; 2] =
    [crate::oauth::dpop::DPOP_NONCE, header::WWW_AUTHENTICATE];

/// The cross-origin policy on the public protocol surface.
///
/// Any origin, without credentials: the browser then never attaches a
/// cookie, so a page on somebody else's origin can present only what its
/// own script holds — a bearer or DPoP token it was issued — which is how
/// the atproto OAuth flow from a web app and every browser-based checker
/// reaches a PDS. What it is layered on is decided in [`app_with_repos`].
pub fn cross_origin_layer() -> CorsLayer {
    CorsLayer::new()
        .allow_origin(Any)
        .allow_methods([Method::GET, Method::POST, Method::OPTIONS])
        .allow_headers(CORS_ALLOWED_HEADERS)
        .expose_headers(CORS_EXPOSED_HEADERS)
        .max_age(CORS_MAX_AGE)
}

/// Parses a JSON body, mapping any failure to a 400 in the atproto shape.
///
/// `axum::Json` is not used as an extractor here because its rejection is
/// axum's own plain-text body, not `{"error":..,"message":..}`.
pub(crate) fn parse_body<T: DeserializeOwned>(body: &Bytes) -> Result<T, ApiError> {
    serde_json::from_slice(body).map_err(|err| ApiError::bad_request(err.to_string()))
}

/// The 500 for a `spawn_blocking` task that panicked instead of answering.
///
/// A panic inside a blocking task never reaches the worker that awaited it —
/// it surfaces as a `JoinError` instead — so this is the one place that
/// becomes a response rather than a connection that hangs up with nothing
/// said.
///
/// A `JoinError`'s own text is the panic payload, which is whatever the line
/// that panicked passed to `panic!` or to `expect`: a file it was reading, a
/// row it was holding, an assertion written for whoever is reading the log.
/// `route` and that payload go to the log line, and the caller — who may be
/// anyone, since `com.atproto.sync.getBlob` reaches here without a
/// credential — is told that the request was not answered and nothing else.
pub(crate) fn blocking_task_failed(route: &str, error: tokio::task::JoinError) -> ApiError {
    tracing::error!(route, %error, "a blocking task panicked before answering");
    ApiError::new(
        StatusCode::INTERNAL_SERVER_ERROR,
        "InternalServerError",
        "the server did not answer this request".to_owned(),
    )
}

/// Runs a `com.atproto.sync` read on the blocking pool.
///
/// Every read this is used for can reach `Registry::repository`, which
/// rebuilds the Merkle search tree and signs a commit when the cached one no
/// longer names the head — and then serialises it. That is milliseconds to
/// seconds of CPU with no await in it, and the deployment in `infra/pds`
/// runs two async workers, so two such reads left inline are every other
/// route waiting. `spawn_blocking` is the same seam `getBlob` and
/// `createAccount` already cross.
///
/// `route` names the method in the log line a panicking task leaves; see
/// [`blocking_task_failed`].
pub(crate) async fn sync_read<T, F>(
    state: &AppState,
    route: &'static str,
    work: F,
) -> Result<T, Response>
where
    F: for<'a> FnOnce(&'a dyn Registry) -> Result<T, ProvisionError> + Send + 'static,
    T: Send + 'static,
{
    // Acquired before `spawn_blocking`, not inside it: what it bounds is
    // blocking-pool threads, and claiming one of those first would hold it
    // idle while this request waits its turn. Released when this function
    // returns, which is when the read is done.
    let _permit = SYNC_READ_CONCURRENCY
        .acquire()
        .await
        .expect("this semaphore is never closed");
    let registry = state.registry.clone();
    match tokio::task::spawn_blocking(move || work(registry.as_ref())).await {
        Ok(Ok(value)) => Ok(value),
        Ok(Err(err)) => Err(sync_error(state, &err).into_response()),
        Err(err) => Err(blocking_task_failed(route, err).into_response()),
    }
}

/// Spends one of the caller's [`SYNC_READ_LIMIT`] sync reads, or answers
/// the refusal.
///
/// These routes take no credential, so the address is the only caller there
/// is. Checked before the work rather than after, which is the whole point:
/// the cost this bounds is the rebuild, not the response.
pub(crate) fn require_sync_budget(
    state: &AppState,
    headers: &axum::http::HeaderMap,
    peer: crate::rate_limit::Peer,
) -> Result<(), ApiError> {
    let key = crate::rate_limit::caller_key(headers, peer.0, state.trust_forwarded_headers);
    if state.sync_rate_limiter.check(&key) {
        return Ok(());
    }
    info!(caller = %key, "sync read refused: too many from this address");
    Err(
        ApiError::rate_limited("too many sync reads from this address; slow down").with_header(
            axum::http::header::RETRY_AFTER,
            SYNC_READ_WINDOW.as_secs().to_string(),
        ),
    )
}

/// The largest request body this server will hold in memory.
///
/// Two mebibytes, which is what axum bounds a buffering extractor to when
/// nothing says otherwise. It is written down here rather than inherited
/// silently, because it is the only thing standing between an anonymous
/// caller and this process's memory, and a bound that arrives from a
/// dependency's default is one nothing in this repository would notice
/// changing.
///
/// It is not a bound on a *blob*: `com.atproto.repo.uploadBlob` takes the
/// body as a stream and never buffers it, and its size limit is the blob
/// store's own. This is the bound on the JSON every other procedure sends.
///
/// Enforcement is streaming rather than after the fact — the body is refused
/// as it arrives, at the first frame that carries the request past the bound,
/// so a caller announcing or sending a gigabyte never has a gigabyte of it
/// resident here.
pub const MAX_REQUEST_BODY: usize = 2 * 1024 * 1024;

/// A buffered request body, refused in the atproto shape.
///
/// `Bytes` on its own is an extractor whose rejection is axum's own
/// plain-text sentence with no `error` name at all — the same reason
/// `parse_body` exists rather than `axum::Json`, and the same reason every
/// query below is extracted as a `Result`. A body over
/// [`MAX_REQUEST_BODY`] is the one failure on this surface a caller cannot
/// avoid by writing a better request, so it is the one that most has to be
/// machine-readable.
pub struct XrpcBody(pub Bytes);

impl<S> axum::extract::FromRequest<S> for XrpcBody
where
    S: Send + Sync,
{
    type Rejection = ApiError;

    async fn from_request(
        request: axum::extract::Request,
        state: &S,
    ) -> Result<Self, Self::Rejection> {
        match Bytes::from_request(request, state).await {
            Ok(body) => Ok(Self(body)),
            // `RequestTooLarge` names no error the lexicons declare, the
            // same position `RateLimitExceeded` and `ServerNotReady` are in:
            // 413 is the standard status and every client already knows it,
            // and the name is there for the ones that switch on it. The
            // message is this crate's own rather than the rejection's, so
            // nothing about how the body was buffered reaches the caller.
            Err(rejection) if rejection.status() == StatusCode::PAYLOAD_TOO_LARGE => {
                Err(request_too_large())
            }
            Err(rejection) => Err(ApiError::bad_request(rejection.body_text())),
        }
    }
}

/// `RequestTooLarge`: a body over [`MAX_REQUEST_BODY`].
pub(crate) fn request_too_large() -> ApiError {
    ApiError::new(
        StatusCode::PAYLOAD_TOO_LARGE,
        "RequestTooLarge",
        format!("a request body may not exceed {MAX_REQUEST_BODY} bytes"),
    )
}

/// Resolves a repo write route's credential — an agent token or a DPoP-bound
/// OAuth access token. See `auth::require_agent_token_or_dpop`.
fn resource_credential(
    state: &AppState,
    headers: &HeaderMap,
    http_method: &str,
    path: &str,
) -> Result<auth::Writer, ApiError> {
    auth::require_agent_token_or_dpop(
        state.registry.as_ref(),
        &state.oauth,
        headers,
        http_method,
        path,
    )
}

/// Checks a write against the scope of the token that presented it — see
/// `auth::Writer::authorize` — and writes a refusal to the evaluation log
/// beside the policy's own, so what a token was refused reads in one place.
///
/// A collection that is not an NSID is refused first, as the store would
/// refuse it, so a caller cannot write arbitrary strings into that log.
fn authorize_write<'a>(
    state: &AppState,
    writer: &'a auth::Writer,
    writes: &[(&str, Action)],
) -> Result<auth::Authorized<'a>, ApiError> {
    for (collection, _) in writes {
        didbot_pds::validate_collection(collection)
            .map_err(|err| crate::error::repo_surface(&didbot_pds::ProvisionError::Record(err)))?;
    }
    writer.authorize(writes).map_err(|missing| {
        record_scope_refusal(state, writer.did(), &missing);
        auth::insufficient_scope(writer.did(), &missing)
    })
}

/// Checks one upload against the scope of the token that presented it, and
/// writes a refusal down the way [`authorize_write`] does.
pub(crate) fn authorize_upload<'a>(
    state: &AppState,
    writer: &'a auth::Writer,
    mime_type: &str,
) -> Result<auth::Authorized<'a>, ApiError> {
    writer.authorize_blob(mime_type).map_err(|missing| {
        record_scope_refusal(state, writer.did(), &missing);
        auth::insufficient_scope(writer.did(), &missing)
    })
}

/// The evaluation log's row for a write outside its token's scope, named
/// [`auth::TOKEN_SCOPE_RULE`] where a tree's row names a policy.
fn record_scope_refusal(state: &AppState, did: &str, missing: &auth::ScopeMissing) {
    let Some(account) = state.registry.account(did) else {
        return;
    };
    let pds_did = deployment_did(state);
    let subject = match &missing.wanted {
        auth::Wanted::Record { collection, action } => didbot_pds::policy::write_subject(
            account.did.as_str(),
            account.kind,
            account.handle.as_deref(),
            &pds_did,
            state.registry.zone(),
            None,
            Some(&missing.client_id),
            collection,
            match action {
                Action::Create => didbot_pds::policy::WriteAction::Create,
                Action::Update | Action::Manage => didbot_pds::policy::WriteAction::Update,
                Action::Delete => didbot_pds::policy::WriteAction::Delete,
            },
            &[],
            None,
            None,
            &[],
        ),
        auth::Wanted::Upload { mime_type } => didbot_pds::policy::blob_subject(
            account.did.as_str(),
            account.kind,
            account.handle.as_deref(),
            &pds_did,
            state.registry.zone(),
            Some(&missing.client_id),
            mime_type,
            None,
            None,
        ),
        // A call at another service is no write or upload a policy judges,
        // so it has no subject to log; `service_auth::authorize` logs it.
        auth::Wanted::Rpc { .. } => return,
    };
    state
        .oauth
        .evaluation_log
        .record_denial(didbot_pds::evaluation_log::Denial {
            account: account.did.as_str(),
            policies_fired: &[auth::TOKEN_SCOPE_RULE.to_owned()],
            verdict: didbot_pds::evaluation_log::Verdict::Reject,
            reason: &format!("the token's scope does not cover `{}`", missing.needed()),
            cut: &[],
            payload_hash: &state.oauth.evaluation_log.hash_subject(&subject),
            policy_version: &didbot_pds::policy::PolicyVersion(auth::TOKEN_SCOPE_RULE.to_owned()),
            now: subject.universal().now,
            subject: &subject,
        });
}

/// `POST /xrpc/bot.did.pollOperatorClaim`
///
/// Asks this server to look for its operator's claim sooner than its timer
/// would. **Urgency, never authority.**
///
/// # Why this exists, and why it cannot be more than a hint
///
/// A server reaches [`ServerState::Claimed`](didbot_pds::ServerState) only
/// by *observing* a standing `bot.did.operator` record through its own poll.
/// Nothing here changes that: this method takes no body, reads no
/// credential, names no operator, and its entire effect is to wake a task
/// that was going to run anyway. A server that took anybody's word for a
/// claim would be a server anybody could claim.
///
/// What it buys is the gap between an operator running `didbot operate` and
/// the next scheduled poll, which is otherwise minutes of staring at a page
/// that says `unclaimed` after the work is already done.
///
/// # Why it cannot be turned into a poll amplifier
///
/// The bound is on this server's *outbound* rate, not on inbound requests:
/// `crate::operator_poll::MIN_POLL_INTERVAL` is a floor between two polls,
/// and every nudge arriving inside one window coalesces into the same single
/// poll. So an unbounded flood of nudges costs the operator's PDS at most
/// one extra request per window — which is what the timer was already going
/// to spend — and costs this server a wake-up on a task that exists. A rate
/// limiter on this route would bound how many nudges are *accepted* while
/// still letting an accepted burst multiply outbound requests, which is the
/// wrong quantity to bound.
///
/// It answers `202` — accepted, nothing has happened yet, which is literally
/// true — with a `Retry-After` naming the floor, so a caller that wants to
/// nudge again knows when it could matter.
async fn poll_operator_claim(State(state): State<AppState>) -> Response {
    state.nudge.nudge();
    (
        StatusCode::ACCEPTED,
        [(
            header::RETRY_AFTER,
            crate::operator_poll::MIN_POLL_INTERVAL
                .as_secs()
                .to_string(),
        )],
        Json(json!({})),
    )
        .into_response()
}

/// `GET /health`
///
/// `status` is liveness: it is `"ok"` whenever this process can schedule a
/// task to answer at all, which is what a load balancer's health check
/// should gate traffic on — see `crates/didbot-serve/src/health.rs` for why
/// that is a different claim from `progress`, which reports whether the
/// deployment's own heartbeat tick is still advancing rather than merely
/// whether a socket accepted.
///
/// `certificates` is the second claim, and the one an alarm can be pointed
/// at: the deployment's health checks read it for certificate state,
/// and a renewal that keeps failing reaches expiry — a total outage for
/// every hostname in the zone — over weeks. Carried in this same response
/// rather than on a route of its own so the check already being made says
/// it; see `crates/didbot-serve/src/health.rs`.
///
/// `oauthPushes` is the third, and it is here because `POST /oauth/par`
/// cannot say it anywhere else: that endpoint is unauthenticated, so it
/// answers a flood aimed at one account exactly as it answers an ordinary
/// push, and the log line it writes is sampled. These are counts since this
/// process started, so an operator watches them move rather than reading a
/// level; see `crate::oauth::par::DroppedPushes`.
///
/// `foreignRecords` is the fourth, and the same shape: counts since the
/// process started, from the one reader policy evaluation reads other
/// people's repositories through. `refusals` is separated from `unknowns`
/// because a refusal is this deployment's own decision and an unknown is
/// somebody else's outage, and an operator watching either number climb
/// wants to know which it is looking at. See `didbot_lookup::Counts`.
///
/// `disk` is the fifth: the data volume's `freeBytes` and `totalBytes`,
/// read as this request is answered, and `alert`, the lowest free-space
/// watermark it is below. It is null for a run with no `--data`. See
/// `crate::health::HealthState::disk`.
///
/// `firehose` is the sixth: how many `subscribeRepos` consumers are
/// attached right now, and the newest sequence number assigned. A relay
/// that has stopped reading is `consumers` at zero while `newestSeq` moves.
///
/// `revision` is the commit the binary was built from, or null for a build
/// that was not told; see `crate::health::REVISION`. A deploy waits for it
/// to name the commit it shipped.
async fn health(State(state): State<AppState>) -> Response {
    let pushes = crate::oauth::par::dropped_pushes();
    let foreign = state.foreign.counts();
    Json(json!({
        "status": "ok",
        "revision": crate::health::REVISION,
        "zone": state.registry.zone(),
        "progress": state.health.snapshot(),
        "certificates": state.health.certificates().await,
        "oauthPushes": {
            "displaced": pushes.displaced,
            "unresolvedHint": pushes.unresolved_hint,
            "serverFull": pushes.server_full,
        },
        "foreignRecords": {
            "hits": foreign.hits,
            "misses": foreign.misses,
            "unknowns": foreign.unknowns,
            "refusals": foreign.refusals,
        },
        "disk": state.health.disk().await,
        "firehose": {
            "consumers": state.repos.subscribers(),
            "newestSeq": state.repos.newest(),
        },
    }))
    .into_response()
}

/// `GET /xrpc/_health`
///
/// The reference PDS's health route, which pdscheck and debug.hose.cam probe
/// and read a server's version from. The same answer that server gives
/// (`packages/pds/src/basic-routes.ts` in bluesky-social/atproto at
/// `388dac9f`): `{"version": ...}`, or a 503 adding
/// `"error": "Service Unavailable"` while this deployment serves no reads.
/// `version` is the `revision` [`health`] reports, and is left out, as the
/// reference leaves out an unset one, for a build that was not told.
async fn xrpc_health(State(state): State<AppState>) -> Response {
    let mut body = serde_json::Map::new();
    if let Some(revision) = crate::health::REVISION {
        body.insert("version".to_owned(), json!(revision));
    }
    if state.lifecycle.snapshot().policy.serves_reads {
        return Json(body).into_response();
    }
    body.insert("error".to_owned(), json!("Service Unavailable"));
    (
        StatusCode::SERVICE_UNAVAILABLE,
        [(
            axum::http::header::RETRY_AFTER,
            didbot_pds::RETRY_AFTER_SECONDS.to_string(),
        )],
        Json(body),
    )
        .into_response()
}

/// `GET /.well-known/did.json`
///
/// Every agent gets its own hostname, and locally they all land on one
/// listener, so the Host header is the only thing that says which account the
/// caller means.
async fn did_json(
    State(state): State<AppState>,
    uri: axum::http::Uri,
    headers: HeaderMap,
) -> Response {
    // A `404` here says "no such identity", which during `booting` or
    // `provisioning` is not yet true -- and for a `did:web` document the
    // difference matters, since a verifier treats an absent document as the
    // identity never having existed.
    if let Err(err) = require_lifecycle(&state, |p| p.serves_reads, "/.well-known/did.json") {
        return err.into_response();
    }
    let Some(host) = request_authority(&uri, &headers) else {
        return ApiError::bad_request("request names no host").into_response();
    };
    let host = host_without_port(host);

    match state.registry.did_document(host) {
        // `application/json`, the type the did:web method expects a
        // document to be served with. atproto's own resolver refuses any
        // type but that one and `application/did+ld+json`.
        Some(document) => Json(document).into_response(),
        None => ApiError::new(
            StatusCode::NOT_FOUND,
            "NotFound",
            format!("no DID document for host `{host}`"),
        )
        .into_response(),
    }
}

/// `GET /.well-known/atproto-did`
///
/// The other direction from `did.json`: that one says what a DID resolves to,
/// this says which DID answers to a name. A deployment that issues its agents
/// handles publishes a second hostname per account, and this is what makes it
/// a handle rather than a decoration — the plain-text DID, which is what the
/// atproto handle resolution rule asks for.
///
/// It answers exactly when the account's DID document claims the host as its
/// handle, because the registry derives both from one function. That is
/// atproto's bidirectional rule holding by construction rather than by
/// agreement: a resolver that follows this DID to its document and looks for
/// the handle in `alsoKnownAs` finds it, for every account, always. A name
/// nobody holds is a 404, which is the whole of what a resolver reads here.
async fn atproto_did(
    State(state): State<AppState>,
    uri: axum::http::Uri,
    headers: HeaderMap,
) -> Response {
    let Some(host) = request_authority(&uri, &headers) else {
        return ApiError::bad_request("request names no host").into_response();
    };
    let host = host_without_port(host);

    match state.registry.handle_did(host) {
        // Plain text, not JSON: the response body is the DID and nothing else.
        Some(did) => ([(header::CONTENT_TYPE, "text/plain")], did).into_response(),
        None => ApiError::new(
            StatusCode::NOT_FOUND,
            "NotFound",
            format!("no account answers to the handle `{host}`"),
        )
        .into_response(),
    }
}

/// The authority a request names, however its protocol version spells it.
///
/// HTTP/1.1 carries it in `Host`. HTTP/2 carries it as the `:authority`
/// pseudo-header, which arrives in the request URI and never as a header —
/// so a route that picks an account by hostname reads both, or answers only
/// the callers that spoke the older version. This server offers both over
/// TLS.
pub(crate) fn request_authority<'a>(
    uri: &'a axum::http::Uri,
    headers: &'a HeaderMap,
) -> Option<&'a str> {
    uri.authority()
        .map(axum::http::uri::Authority::as_str)
        .or_else(|| headers.get(header::HOST).and_then(|v| v.to_str().ok()))
}

/// Strips a port from a Host header value.
///
/// A development zone is served on an explicit port, so `Host` arrives as
/// `agent.agents.localhost:3000` while the DID is registered under the bare
/// hostname. IPv6 literals arrive bracketed, and the brackets are dropped with
/// the port.
pub(crate) fn host_without_port(host: &str) -> &str {
    if let Some(rest) = host.strip_prefix('[') {
        return rest.split(']').next().unwrap_or(rest);
    }
    match host.rsplit_once(':') {
        Some((name, port)) if !port.is_empty() && port.bytes().all(|b| b.is_ascii_digit()) => name,
        _ => host,
    }
}

/// The answer for a path nothing serves, and for an XRPC call nothing
/// proxies; see `crate::proxy::fallback`.
///
/// `MethodNotImplemented` under `/xrpc/`, which is the name atproto's XRPC
/// layer uses for a procedure a server does not implement, and a 501 to go
/// with it: a client that asked for a method this server does not serve should
/// learn that this server has not built it, not that the URL was a typo.
/// Anything outside `/xrpc/` is an ordinary 404.
pub(crate) fn unknown_route(uri: &axum::http::Uri) -> Response {
    if uri.path().starts_with("/xrpc/") {
        let nsid = uri.path().trim_start_matches("/xrpc/");
        return ApiError::new(
            StatusCode::NOT_IMPLEMENTED,
            "MethodNotImplemented",
            format!("this server does not implement `{nsid}`"),
        )
        .into_response();
    }
    ApiError::new(
        StatusCode::NOT_FOUND,
        "NotFound",
        format!("nothing is served at `{}`", uri.path()),
    )
    .into_response()
}

/// The fallback for the right path and the wrong method.
///
/// A query fetched with POST, or a procedure called with GET. It is a real
/// mistake a hand-written client makes and an empty 405 does not say so.
async fn wrong_method(uri: axum::http::Uri) -> Response {
    ApiError::new(
        StatusCode::METHOD_NOT_ALLOWED,
        "InvalidRequest",
        format!("`{}` is not served with that HTTP method", uri.path()),
    )
    .into_response()
}

/// `GET /xrpc/bot.did.stats`
///
/// `Credential::Disclosure`: open by default, this deployment's own
/// aggregate inventory — see `auth::Disclosure`. The health tick logs the
/// same figures on its own schedule by calling `Registry::stats`
/// in-process, so nothing internal reaches this route over HTTP.
async fn stats(State(state): State<AppState>) -> Response {
    if let Err(err) = require_lifecycle(&state, |p| p.serves_reads, "bot.did.stats") {
        return err.into_response();
    }
    if let Err(err) = auth::require_disclosure(state.disclosure.stats, "bot.did.stats") {
        return err.into_response();
    }
    let enforced = state.registry.enforced_policies();
    Json(StatsResponse::from_stats(&state.registry.stats()).with_policy(enforced.as_deref()))
        .into_response()
}

/// Resolves `identifier` — a handle or a DID — to an account DID this
/// deployment hosts.
///
/// What `POST /oauth/par` resolves a `login_hint` with: [`resolve_repo`]
/// under another name, kept separate because a sign-in failure and a
/// repository lookup failure want different error shapes, and folding them
/// into one function would make one of the two callers wrong.
fn resolve_identifier(state: &AppState, identifier: &str) -> Option<String> {
    if identifier.starts_with("did:") {
        state
            .registry
            .account(&canonical_did(identifier))
            .map(|a| a.did.as_str().to_owned())
    } else {
        state
            .registry
            .handle_did(identifier)
            .filter(|did| state.registry.account(did).is_some())
    }
}

pub(crate) fn too_many_provisioning_attempts() -> Response {
    ApiError::rate_limited("too many provisioning attempts; slow down").into_response()
}

/// `GET /xrpc/com.atproto.server.describeServer`
///
/// Discovery: what this server is, before a client has any account on it.
/// `availableUserDomains` is read out of the registry's real zone
/// configuration rather than stated as a fixed string, so a deployment with
/// several zones reports all of them — see
/// [`Registry::available_user_domains`]. Every other field the lexicon
/// declares optional is left off rather than answered with an invented
/// value; see [`DescribeServerResponse`]'s own doc for which and why.
async fn describe_server(State(state): State<AppState>) -> Response {
    // One of the two reads `didbot operate` makes before it can write a claim;
    // the other is the DID document above. Both must answer in `unclaimed`
    // or the handshake deadlocks, which `didbot_pds::server_state`'s
    // `the_handshake_cannot_deadlock` is what actually holds.
    if let Err(err) = require_lifecycle(
        &state,
        |p| p.serves_reads,
        "com.atproto.server.describeServer",
    ) {
        return err.into_response();
    }
    Json(DescribeServerResponse {
        did: state.registry.service_did(),
        available_user_domains: state.registry.available_user_domains(),
    })
    .into_response()
}