didbot_serve/routes/

repo.rs

use super::*;

use crate::write_meter::{Bucket, WriteMeter};
use tracing::warn;

/// `POST /xrpc/com.atproto.repo.createRecord`
///
/// 404 `RepoNotFound` when the repository is unknown, 400 when the collection
/// is not one this deployment defines, the record's `$type` disagrees with it,
/// or the requested `rkey` is one the collection's key strategy will not admit
///; 400 `InvalidSwap` when the `swapCommit` named is not the commit the
/// repository is at. The key in the returned `uri` is the one the lexicon
/// decided, which is not always the one that was asked for. The response
/// carries the `uri` and the `cid` the lexicon requires; see
/// [`CreateRecordResponse`].
///
/// # A key already in use is refused
///
/// This is `createRecord`, and in atproto `createRecord` creates: a key
/// already in use is a 400 and `putRecord` is the upsert. This server could not
/// tell the two apart while there was no CID to condition a refusal on. There
/// is one now, and compare-and-swap has landed. The write carries
/// [`Precondition::Absent`], which is the same check every other
/// precondition goes through and is evaluated in the same place.
///
/// The refusal is `InvalidRequest` rather than the `InvalidSwap` the lexicon
/// declares, and for a reason about names: `InvalidSwap` tells a
/// client to re-read and retry, and this retry can never succeed. A caller
/// that meant to replace the record wants `putRecord`.
pub(crate) async fn create_record(
    State(state): State<AppState>,
    headers: HeaderMap,
    XrpcBody(body): XrpcBody,
) -> Response {
    // Stages 1 and 2 of `docs/write-pipeline.md`, before this route reads
    // the body or spends a credential: a halted server and a server not
    // accepting writes are both public facts -- every `503` body
    // names the second -- and the same answer is owed to a caller with a bad token, a
    // malformed body or neither. Deciding either one needs no body, so a
    // caller that trips one hears about it rather than about its JSON.
    if let Err(err) = check_write_not_halted(&state) {
        return err.into_response();
    }
    let writer = match resource_credential(
        &state,
        &headers,
        "POST",
        "/xrpc/com.atproto.repo.createRecord",
    ) {
        Ok(writer) => writer,
        Err(err) => return err.into_response(),
    };
    let request: CreateRecordRequest = match parse_body(&body) {
        Ok(request) => request,
        Err(err) => return err.into_response(),
    };
    let did = match require_writable_repo(&state, writer.did(), &request.repo) {
        Ok(did) => did,
        Err(err) => return err.into_response(),
    };
    let mut swap = match request.swap.resolve_commit_only("createRecord") {
        Ok(swap) => swap,
        Err(err) => return err.into_response(),
    };
    swap.record = Precondition::Absent;
    let authorized =
        match authorize_write(&state, &writer, &[(&request.collection, Action::Create)]) {
            Ok(authorized) => authorized,
            Err(err) => return err.into_response(),
        };
    if let Some(err) = refuse_skipped_validation(request.validate, [request.collection.as_str()]) {
        return err.into_response();
    }
    if let Some(err) = require_known_lexicon(&request.collection, &request.record, request.validate)
    {
        return err.into_response();
    }
    write_record(
        &state,
        &did,
        authorized.client_id,
        &request.collection,
        request.rkey.as_deref(),
        request.record,
        &swap,
        Occupied::Refuse,
        body.len(),
    )
    .await
}

/// `POST /xrpc/com.atproto.repo.putRecord`
///
/// The upsert, which is what it is in atproto: a key already in use is
/// replaced rather than refused, and that is now the difference between this
/// and [`create_record`]. Both `swapRecord` and `swapCommit` are honoured,
/// and `swapRecord: null` asserts the record does not exist yet.
pub(crate) async fn put_record(
    State(state): State<AppState>,
    headers: HeaderMap,
    XrpcBody(body): XrpcBody,
) -> Response {
    // Stages 1 and 2 of `docs/write-pipeline.md`, before this route reads
    // the body or spends a credential: a halted server and a server not
    // accepting writes are both public facts -- every `503` body
    // names the second -- and the same answer is owed to a caller with a bad token, a
    // malformed body or neither. Deciding either one needs no body, so a
    // caller that trips one hears about it rather than about its JSON.
    if let Err(err) = check_write_not_halted(&state) {
        return err.into_response();
    }
    let writer =
        match resource_credential(&state, &headers, "POST", "/xrpc/com.atproto.repo.putRecord") {
            Ok(writer) => writer,
            Err(err) => return err.into_response(),
        };
    let request: PutRecordRequest = match parse_body(&body) {
        Ok(request) => request,
        Err(err) => return err.into_response(),
    };
    let did = match require_writable_repo(&state, writer.did(), &request.repo) {
        Ok(did) => did,
        Err(err) => return err.into_response(),
    };
    let swap = match request.swap.resolve() {
        Ok(swap) => swap,
        Err(err) => return err.into_response(),
    };
    // An upsert may do either, and which one is not known until the store
    // is read, so the token must cover both.
    let wanted = [
        (request.collection.as_str(), Action::Create),
        (request.collection.as_str(), Action::Update),
    ];
    let authorized = match authorize_write(&state, &writer, &wanted) {
        Ok(authorized) => authorized,
        Err(err) => return err.into_response(),
    };
    if let Some(err) = refuse_skipped_validation(request.validate, [request.collection.as_str()]) {
        return err.into_response();
    }
    if let Some(err) = require_known_lexicon(&request.collection, &request.record, request.validate)
    {
        return err.into_response();
    }
    write_record(
        &state,
        &did,
        authorized.client_id,
        &request.collection,
        Some(&request.rkey),
        request.record,
        &swap,
        Occupied::Replace,
        body.len(),
    )
    .await
}

/// What a route does about a record key that is already taken.
///
/// The one difference between `createRecord` and `putRecord`, named so the
/// refusal can be reported as the thing it is rather than as a swap the
/// caller never asked for.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum Occupied {
    /// `createRecord`: a key in use is a 400 naming it.
    Refuse,
    /// `putRecord`: a key in use is replaced.
    Replace,
}

/// The write both `createRecord` and `putRecord` perform.
///
/// One function because they are one operation apart from `swap` and
/// `occupied`, and writing the response assembly twice is how the two would
/// start disagreeing about the shape of a `uri`.
#[allow(clippy::too_many_arguments)]
pub(crate) async fn write_record(
    state: &AppState,
    did: &str,
    client_id: Option<&didbot_pds::policy::ClientId>,
    collection: &str,
    rkey: Option<&str>,
    record: serde_json::Value,
    swap: &didbot_pds::Swap,
    occupied: Occupied,
    bytes: usize,
) -> Response {
    if let Err(err) = check_write_not_halted(state) {
        return err.into_response();
    }
    // `docs/write-pipeline.md`'s "where the records a policy declares are
    // read": whatever the applicable policies declared they read about the
    // accounts this record names is read here, while there is still a
    // runtime to await on. The gate itself is judged inside the store's
    // synchronous section, where nothing can.
    state.lookups.prefetch(collection, Some(&record)).await;
    // Off the async worker: every write commits through the WAL, which
    // holds its store mutex across `fsync`, and `createRecord`/`putRecord`
    // are the highest-traffic writes this server answers.
    let registry = state.registry.clone();
    let did_owned = did.to_owned();
    let collection_owned = collection.to_owned();
    let rkey_owned = rkey.map(str::to_owned);
    let swap_owned = swap.clone();
    let client_owned = client_id.cloned();
    let meter = state.write_meter.clone();
    let outcome = tokio::task::spawn_blocking(move || {
        let charge = || meter_write(&meter, &did_owned, 1, bytes);
        registry.put_record_from(
            &did_owned,
            &collection_owned,
            rkey_owned.as_deref(),
            record,
            &swap_owned,
            client_owned.as_ref(),
            Some(&charge),
        )
    })
    .await;
    let client = client_id.map(didbot_pds::policy::ClientId::as_str);
    match outcome {
        Err(join_err) => blocking_task_failed("a record write", join_err).into_response(),
        Ok(Ok(written)) => {
            info!(
                repo = %did,
                client,
                collection,
                rkey = %written.rkey,
                cid = %written.cid,
                "wrote record"
            );
            Json(CreateRecordResponse::written(did, collection, &written)).into_response()
        }
        Ok(Err(err)) => {
            // The record itself is never logged, not even truncated: this
            // line exists to explain a refusal, and the body that caused it
            // is the caller's content rather than the server's business.
            info!(repo = %did, client, collection, error = %err, "record write refused");
            // `createRecord`'s implied precondition is reported as what the
            // caller actually did — named a key that is taken — rather than
            // as a swap it never sent.
            if occupied == Occupied::Refuse && is_swap_failure(&err) {
                // The key is not always the one the caller named: a
                // `literal:self` collection decides it, so the message says
                // where the write would have landed in the caller's own
                // terms.
                let occupied = match rkey {
                    Some(rkey) => format!("`{}`", record_uri(did, collection, rkey)),
                    None => format!("the key `{collection}` chose for this write"),
                };
                return ApiError::bad_request(format!(
                    "`createRecord` will not replace a record, and one is already at \
                     {occupied}; use `putRecord` to replace it"
                ))
                .into_response();
            }
            refused_write(&err, did, client).into_response()
        }
    }
}

/// A store's refusal of a write, as the caller hears it.
///
/// A write past the record ceiling also goes to the refusal journal, naming
/// the account: it is the operator's ceiling, and only the operator can
/// raise it.
pub(crate) fn refused_write(
    err: &didbot_pds::ProvisionError,
    did: &str,
    client_id: Option<&str>,
) -> ApiError {
    let refused = crate::error::repo_surface(err);
    if matches!(err, didbot_pds::ProvisionError::TooManyRecords { .. }) {
        crate::refusal::refused(crate::refusal::Refusal {
            endpoint: "xrpc",
            client_id,
            account: Some(did),
            error: refused.error(),
            reason: refused.message(),
            severity: crate::refusal::Severity::Warn,
        });
    }
    refused
}

/// Whether a refusal was a record-level compare-and-swap failing.
pub(crate) fn is_swap_failure(err: &didbot_pds::ProvisionError) -> bool {
    matches!(
        err,
        didbot_pds::ProvisionError::Record(didbot_pds::RecordError::SwapFailed { .. })
    )
}

/// The one place this server's own lifecycle refuses a request.
///
/// `capability` selects a field of [`ServerPolicy`]; `route` names what was
/// asked for, so the refusal says which surface deferred rather than only
/// that something did. Every caller is one line
/// (`require_lifecycle(&state, |p| p.serves_reads, "…")?`), which is the
/// point: a surface that consults the lifecycle by matching on the state
/// itself would be a second, drifting copy of
/// `didbot_pds::server_state`'s policy table.
///
/// # Where the read gate is applied, and where it is not
///
/// `ServerPolicy::serves_reads` is false only in `booting` and
/// `provisioning`, which last a store read and a keypair mint respectively
/// and during which no agent account can exist. So it is applied where an
/// absent or empty answer would be *misread* — `listAccounts` answering `[]`,
/// `subscribeRepos` opening an empty stream, `did.json` answering `404` —
/// and not on the per-repository reads, where the natural `RepoNotFound` is
/// already the true answer and a `503` would say less.
///
/// The refusal is [`ApiError::not_ready`] — a `503` carrying a
/// `Retry-After`, never a `404` and never an empty answer. See that
/// constructor and `didbot_pds::RETRY_AFTER_SECONDS` for how a relay is
/// meant to tell "try later" from "broken" from "nothing here."
pub(crate) fn require_lifecycle(
    state: &AppState,
    capability: impl FnOnce(&ServerPolicy) -> bool,
    route: &str,
) -> Result<(), ApiError> {
    let snapshot = state.lifecycle.snapshot();
    if capability(&snapshot.policy) {
        return Ok(());
    }
    Err(ApiError::not_ready(route, &snapshot))
}

/// Refuses a new account once this deployment already holds
/// [`AppState::max_accounts`] of them; see `crate::tree::create_account`.
///
/// Counted from [`Registry::account_count`] rather than a purpose-built
/// counter, so this and `crate::health`'s periodic line — which reads
/// [`Registry::stats`], the pricier sibling this call deliberately avoids —
/// can never disagree about what "how many accounts" means. This route runs
/// on every provisioning request on a two-worker runtime, so unlike the
/// health tick it cannot afford `stats`'s per-account clone just to read one
/// field off it; bounded regardless by [`AppState::max_accounts`] itself,
/// since nothing can be counted here that was not admitted past this same
/// check.
pub(crate) fn require_account_capacity(state: &AppState) -> Result<(), ApiError> {
    let live = state.registry.account_count() as u64;
    if live < state.max_accounts {
        return Ok(());
    }
    Err(ApiError::account_cap_reached(state.max_accounts))
}

/// The shared gate on every `com.atproto.repo.*` write.
///
/// Two independent refusals, in a fixed order. The e-stop goes first because
/// `Halted` is the answer callers already switch on and it names whose stop
/// it is; the lifecycle follows, and `ServerNotReady` means something only
/// this server knows — that nobody is currently answerable for it, so it
/// accepts no new data. A lapsed operator claim trips both, and the caller
/// hears `Halted` naming the missing record.
///
/// `ServerPolicy::accepts_record_writes` is the axis the whole lifecycle
/// turns on. An unclaimed server keeps every read this one serves and
/// accumulates nothing new; see `didbot_pds::server_state`.
pub(crate) fn check_write_not_halted(state: &AppState) -> Result<(), ApiError> {
    state.estop.check_use(EstopRefusal::Operation)?;
    require_record_writes(state)?;
    Ok(())
}

/// `docs/write-pipeline.md`'s stage 3, run at the earliest moment a route
/// can run it: the instant the request envelope names a repository.
///
/// Resolves the repository, refuses a caller writing into one that is not
/// its own, and then refuses a repository whose [`didbot_pds::HostedAccount`]
/// policy does not accept writes. The store performs the last check again on the
/// way past; this one exists so that the answer a frozen account gets is
/// `AccountNotWritable` rather than whatever stage 4 would have said about
/// its record. "Account frozen" says nothing you write will work until an
/// operator acts and "your record is malformed" says fix this write, and an
/// agent that hears the second when the first is true retries forever.
///
/// The envelope has to be parsed to know which repository is meant, so this
/// cannot precede parsing outright the way stages 1 and 2 do — it precedes
/// every judgment of the *record*, which is what stage 4 in the diagram is.
pub(crate) fn require_writable_repo(
    state: &AppState,
    authenticated: &str,
    repo: &str,
) -> Result<String, ApiError> {
    let did = resolve_repo(state, repo)?;
    if did != authenticated {
        return Err(ApiError::repo_mismatch(authenticated, &did));
    }
    state
        .registry
        .check_writable(&did)
        .map_err(|err| crate::error::repo_surface(&err))?;
    Ok(did)
}

/// Spends `ops` write operations and `bytes` request bytes of `did`'s share
/// of [`AppState::write_meter`], or refuses with `429 RateLimitExceeded`.
///
/// `uploadBlob` runs it once the upload has been judged. Record writes hand
/// [`meter_write`] to the store, which runs it once policy has allowed the
/// write. Either way, a write policy refuses spends nothing of the
/// account's share.
pub(crate) fn charge_write(
    state: &AppState,
    did: &str,
    ops: usize,
    bytes: usize,
) -> Result<(), ApiError> {
    meter_write(&state.write_meter, did, ops, bytes).map_err(|err| ApiError::from(&err))
}

/// [`charge_write`], against `meter`, refusing as the store refuses: the
/// admission charge a record write carries into it.
pub(crate) fn meter_write(
    meter: &WriteMeter,
    did: &str,
    ops: usize,
    bytes: usize,
) -> Result<(), didbot_pds::ProvisionError> {
    let ops = u32::try_from(ops).unwrap_or(u32::MAX);
    let Err(refusal) = meter.charge(did, ops, bytes as u64) else {
        return Ok(());
    };
    let limits = meter.limits();
    let (bucket, rate, burst) = match refusal.bucket {
        Bucket::Ops => (
            "operations",
            u64::from(limits.ops_per_hour),
            u64::from(limits.ops_burst),
        ),
        Bucket::Bytes => ("bytes", limits.bytes_per_hour, limits.bytes_burst),
    };
    let wanted = refusal.wanted;
    let retry_after_secs = refusal
        .retry_after
        .map(|wait| (wait.as_secs() + u64::from(wait.subsec_nanos() > 0)).max(1));
    warn!(
        repo = %did,
        bucket,
        wanted,
        rate_per_hour = rate,
        burst,
        retry_after_secs,
        "write refused: the account is over its write rate"
    );
    let message = match retry_after_secs {
        Some(secs) => format!(
            "this write needs {wanted} {bucket}, and `{did}` may write {rate} {bucket} an hour \
             with at most {burst} at once; retry after {secs}s"
        ),
        None => format!(
            "this write needs {wanted} {bucket}, and `{did}` may write at most {burst} {bucket} \
             at once"
        ),
    };
    Err(didbot_pds::ProvisionError::OverWriteBudget {
        message,
        retry_after_secs,
    })
}

/// The lifecycle's half of [`check_write_not_halted`], alone.
///
/// `crate::blobs` runs its own e-stop check for its own refusal reason and
/// needs this half separately; nothing else should call it directly.
pub(crate) fn require_record_writes(state: &AppState) -> Result<(), ApiError> {
    require_lifecycle(
        state,
        |p| p.accepts_record_writes,
        "com.atproto.repo writes",
    )
}

/// `POST /xrpc/com.atproto.repo.deleteRecord`
///
/// Idempotent, as the lexicon's own description asks for: "delete a
/// repository record, or ensure it doesn't exist". A record that was not
/// there is a 200 and an empty body, exactly like one that was. An unknown
/// *repository* is still a 404 — that is a caller pointing at nothing, not a
/// caller making sure of something.
///
/// The response omits `commit`, which the lexicon marks optional. There is no
/// commit here to report.
pub(crate) async fn delete_record(
    State(state): State<AppState>,
    headers: HeaderMap,
    XrpcBody(body): XrpcBody,
) -> Response {
    // Stages 1 and 2 of `docs/write-pipeline.md`, before this route reads
    // the body or spends a credential: a halted server and a server not
    // accepting writes are both public facts -- every `503` body
    // names the second -- and the same answer is owed to a caller with a bad token, a
    // malformed body or neither. Deciding either one needs no body, so a
    // caller that trips one hears about it rather than about its JSON.
    if let Err(err) = check_write_not_halted(&state) {
        return err.into_response();
    }
    let writer = match resource_credential(
        &state,
        &headers,
        "POST",
        "/xrpc/com.atproto.repo.deleteRecord",
    ) {
        Ok(writer) => writer,
        Err(err) => return err.into_response(),
    };
    let request: DeleteRecordRequest = match parse_body(&body) {
        Ok(request) => request,
        Err(err) => return err.into_response(),
    };
    let did = match require_writable_repo(&state, writer.did(), &request.repo) {
        Ok(did) => did,
        Err(err) => return err.into_response(),
    };
    let swap = match request.swap.resolve() {
        Ok(swap) => swap,
        Err(err) => return err.into_response(),
    };
    let client_id = match authorize_write(&state, &writer, &[(&request.collection, Action::Delete)])
    {
        Ok(authorized) => authorized.client_id,
        Err(err) => return err.into_response(),
    };
    if let Err(err) = check_write_not_halted(&state) {
        return err.into_response();
    }
    // Off the async worker for the reason `write_record` gives.
    let registry = state.registry.clone();
    let did_owned = did.clone();
    let collection = request.collection.clone();
    let rkey = request.rkey.clone();
    let client_owned = client_id.cloned();
    let meter = state.write_meter.clone();
    let bytes = body.len();
    let outcome = tokio::task::spawn_blocking(move || {
        let charge = || meter_write(&meter, &did_owned, 1, bytes);
        registry.delete_record_from(
            &did_owned,
            &collection,
            &rkey,
            &swap,
            client_owned.as_ref(),
            Some(&charge),
        )
    })
    .await;
    match outcome {
        Err(join_err) => blocking_task_failed("a record deletion", join_err).into_response(),
        Ok(Ok(removed)) => {
            info!(
                repo = %did,
                client = client_id.map(didbot_pds::policy::ClientId::as_str),
                collection = %request.collection,
                rkey = %request.rkey,
                removed,
                "record deletion"
            );
            Json(json!({})).into_response()
        }
        Ok(Err(err)) => crate::error::repo_surface(&err).into_response(),
    }
}

/// `POST /xrpc/com.atproto.repo.applyWrites`
///
/// One commit, over every operation in the batch, so at most
/// [`crate::wire::MAX_APPLY_WRITES`] of them. Every write is checked —
/// against its collection's lexicon, its record key strategy and the atproto
/// data model — before any of them touches the store, so a batch that was
/// going to be refused is refused whole rather than applied as a prefix; see
/// [`didbot_pds::Registry::apply_writes`]. `swapCommit` is a genuine
/// compare-and-swap against the commit this batch is about to replace, not a
/// look before it: it is checked under the same lock the batch's one commit
/// is built under.
pub(crate) async fn apply_writes(
    State(state): State<AppState>,
    headers: HeaderMap,
    XrpcBody(body): XrpcBody,
) -> Response {
    // Stages 1 and 2 of `docs/write-pipeline.md`, before this route reads
    // the body or spends a credential: a halted server and a server not
    // accepting writes are both public facts -- every `503` body
    // names the second -- and the same answer is owed to a caller with a bad token, a
    // malformed body or neither. Deciding either one needs no body, so a
    // caller that trips one hears about it rather than about its JSON.
    if let Err(err) = check_write_not_halted(&state) {
        return err.into_response();
    }
    let writer = match resource_credential(
        &state,
        &headers,
        "POST",
        "/xrpc/com.atproto.repo.applyWrites",
    ) {
        Ok(writer) => writer,
        Err(err) => return err.into_response(),
    };
    let request: ApplyWritesRequest = match parse_body(&body) {
        Ok(request) => request,
        Err(err) => return err.into_response(),
    };
    if request.writes.len() > crate::wire::MAX_APPLY_WRITES {
        return ApiError::bad_request(format!(
            "`applyWrites` takes at most {} writes, and this batch has {}",
            crate::wire::MAX_APPLY_WRITES,
            request.writes.len()
        ))
        .into_response();
    }
    let did = match require_writable_repo(&state, writer.did(), &request.repo) {
        Ok(did) => did,
        Err(err) => return err.into_response(),
    };
    let swap = match request.swap.resolve_commit_only("applyWrites") {
        Ok(swap) => swap,
        Err(err) => return err.into_response(),
    };
    let written = request.writes.iter().filter_map(|write| match write {
        RepoWrite::Create { collection, .. } | RepoWrite::Update { collection, .. } => {
            Some(collection.as_str())
        }
        RepoWrite::Delete { .. } => None,
    });
    if let Some(err) = refuse_skipped_validation(request.validate, written) {
        return err.into_response();
    }
    if let Err(err) = check_write_not_halted(&state) {
        return err.into_response();
    }

    let writes: Vec<didbot_pds::BatchOp> = request.writes.into_iter().map(to_batch_op).collect();
    // The batch store writes a named key whether or not it is taken, so a
    // create with a key and an update are both upserts, as in `put_record`.
    let wanted: Vec<(&str, Action)> = writes
        .iter()
        .flat_map(|op| {
            let actions: &[Action] = match op {
                didbot_pds::BatchOp::Create { rkey: None, .. } => &[Action::Create],
                didbot_pds::BatchOp::Create { rkey: Some(_), .. }
                | didbot_pds::BatchOp::Update { .. } => &[Action::Create, Action::Update],
                didbot_pds::BatchOp::Delete { .. } => &[Action::Delete],
            };
            actions.iter().map(move |action| (op.collection(), *action))
        })
        .collect();
    let client_id = match authorize_write(&state, &writer, &wanted) {
        Ok(authorized) => authorized.client_id,
        Err(err) => return err.into_response(),
    };
    let stance = didbot_pds::Stance::from_validate(request.validate);
    // Kept alongside the writes handed to the registry, so a result can be
    // told apart as a create or an update and named with the collection it
    // went into: [`didbot_pds::BatchOutcome`] carries neither, because the
    // registry already had them from the operation that produced it.
    let ops = writes.clone();
    for op in &writes {
        let after = match op {
            didbot_pds::BatchOp::Create { record, .. }
            | didbot_pds::BatchOp::Update { record, .. } => Some(record),
            didbot_pds::BatchOp::Delete { .. } => None,
        };
        state.lookups.prefetch(op.collection(), after).await;
    }
    let client = client_id.map(didbot_pds::policy::ClientId::as_str);
    // Off the async worker for the reason `write_record` gives.
    let registry = state.registry.clone();
    let did_owned = did.clone();
    let client_owned = client_id.cloned();
    let meter = state.write_meter.clone();
    // The whole batch, at one operation per write: a batch is the
    // cheapest way to put many writes into the log.
    let (charged, bytes) = (writes.len().max(1), body.len());
    let outcome = tokio::task::spawn_blocking(move || {
        let charge = || meter_write(&meter, &did_owned, charged, bytes);
        registry.apply_writes_from(
            &did_owned,
            writes,
            stance,
            swap.commit.as_ref(),
            client_owned.as_ref(),
            Some(&charge),
        )
    })
    .await;
    match outcome {
        Err(join_err) => blocking_task_failed("a batch write", join_err).into_response(),
        Ok(Ok(outcomes)) => {
            let results: Vec<WriteResult> = ops
                .iter()
                .zip(outcomes)
                .map(|(op, outcome)| write_result(&did, op, outcome))
                .collect();
            info!(repo = %did, client, writes = results.len(), "applied a batch as one commit");
            Json(json!({ "results": results })).into_response()
        }
        Ok(Err(err)) => {
            info!(repo = %did, client, error = %err, "batch refused; nothing was written");
            refused_write(&err, &did, client).into_response()
        }
    }
}

/// Converts one wire-level write into the shape [`Registry::apply_writes`]
/// takes, dropping nothing: the collection, key and value every variant
/// carries reach the registry exactly as the caller sent them.
pub(crate) fn to_batch_op(write: RepoWrite) -> didbot_pds::BatchOp {
    match write {
        RepoWrite::Create {
            collection,
            rkey,
            value,
        } => didbot_pds::BatchOp::Create {
            collection,
            rkey,
            record: value,
        },
        RepoWrite::Update {
            collection,
            rkey,
            value,
        } => didbot_pds::BatchOp::Update {
            collection,
            rkey,
            record: value,
        },
        RepoWrite::Delete { collection, rkey } => didbot_pds::BatchOp::Delete { collection, rkey },
    }
}

/// One batch outcome, on the wire — told apart as a create or an update by
/// the operation that produced it, since [`didbot_pds::BatchOutcome`] itself
/// does not carry that distinction.
pub(crate) fn write_result(
    did: &str,
    op: &didbot_pds::BatchOp,
    outcome: didbot_pds::BatchOutcome,
) -> WriteResult {
    match (op, outcome) {
        (
            didbot_pds::BatchOp::Create { collection, .. },
            didbot_pds::BatchOutcome::Written(written),
        ) => WriteResult::Create {
            uri: record_uri(did, collection, &written.rkey),
            cid: written.cid.to_string(),
            validation_status: didbot_pds::validation_status(
                collection,
                didbot_pds::Stance::Optimistic,
            ),
        },
        (
            didbot_pds::BatchOp::Update { collection, .. },
            didbot_pds::BatchOutcome::Written(written),
        ) => WriteResult::Update {
            uri: record_uri(did, collection, &written.rkey),
            cid: written.cid.to_string(),
            validation_status: didbot_pds::validation_status(
                collection,
                didbot_pds::Stance::Optimistic,
            ),
        },
        (didbot_pds::BatchOp::Delete { .. }, didbot_pds::BatchOutcome::Deleted) => {
            WriteResult::Delete {}
        }
        // `apply_writes` promises `outcomes` answers `writes` one-for-one, in
        // the same order and the same kind; anything else is this server's
        // own bug rather than a caller's.
        (op, outcome) => unreachable!("a {op:?} produced a {outcome:?}"),
    }
}

/// `GET /xrpc/com.atproto.repo.getRecord`
///
/// 404 `RecordNotFound` when the record is not there, which is the name the
/// lexicon declares and the only one it declares.
pub(crate) async fn get_record(
    State(state): State<AppState>,
    peer: crate::rate_limit::Peer,
    headers: axum::http::HeaderMap,
    query: Result<Query<GetRecordQuery>, QueryRejection>,
) -> Response {
    if let Err(refusal) = require_sync_budget(&state, &headers, peer) {
        return refusal.into_response();
    }
    let Query(query) = match query {
        Ok(query) => query,
        Err(rejection) => return ApiError::bad_request(rejection.body_text()).into_response(),
    };
    let did = match resolve_repo(&state, &query.repo) {
        Ok(did) => did,
        Err(err) => return err.into_response(),
    };
    let uri = record_uri(&did, &query.collection, &query.rkey);
    match state
        .registry
        .get_record(&did, &query.collection, &query.rkey)
    {
        Ok(Some(value)) => {
            let cid = match didbot_pds::content_id(&value) {
                Ok(cid) => cid.to_string(),
                Err(err) => return unnameable_record(&uri, &err).into_response(),
            };
            // A caller that named a version is answered from the one version
            // there is. The lexicon says an absent `cid` returns the most
            // recent, and this repository keeps no older ones, so a `cid`
            // that is not the current one names a record this server does not
            // hold — which is `RecordNotFound`, and is what the reference
            // implementation answers for the same request.
            if query.cid.is_some_and(|asked| asked != cid) {
                return ApiError::record_not_found(&uri).into_response();
            }
            Json(GetRecordResponse { uri, cid, value }).into_response()
        }
        Ok(None) => ApiError::record_not_found(&uri).into_response(),
        Err(err) => crate::error::repo_surface(&err).into_response(),
    }
}

/// The 500 for a stored record the data model cannot name.
///
/// Unreachable for anything written through this server: every store computes
/// the CID before it stores the record and refuses one it cannot. It exists
/// for a record replayed out of a write-ahead log written before that was
/// true, which is a server-side fact and not a request a caller can rewrite.
pub(crate) fn unnameable_record(subject: &str, error: &didbot_pds::RecordError) -> ApiError {
    ApiError::new(
        StatusCode::INTERNAL_SERVER_ERROR,
        "InternalServerError",
        format!("`{subject}` is stored and cannot be named: {error}"),
    )
}

/// `GET /xrpc/com.atproto.repo.describeRepo`
///
/// The whole account as a stranger would see it: its handle, its DID, its DID
/// document, the collections it has written anything in, and whether the
/// handle resolves back to the DID.
///
/// # What `collections` now means
///
/// It was always derived from what the repository holds rather than from
/// `didbot_lexicon::nsid::ALL`, so nothing here changed when the
/// collection set opened — but what a reader may conclude from it did.
/// The list is a report of what this account has written and never a menu of
/// what it may write, because there is no longer any such menu. A collection
/// nobody has heard of appearing in it is a client that wrote one, which is
/// also where a typo now shows up: `com.example.scrobbel` is a second
/// collection in this list rather than a refusal on the write.
pub(crate) async fn describe_repo(
    State(state): State<AppState>,
    query: Result<Query<DescribeRepoQuery>, QueryRejection>,
) -> Response {
    let Query(query) = match query {
        Ok(query) => query,
        Err(rejection) => return ApiError::bad_request(rejection.body_text()).into_response(),
    };
    let did = match resolve_repo(&state, &query.repo) {
        Ok(did) => did,
        Err(err) => return err.into_response(),
    };
    let Some(account) = state.registry.account(&did) else {
        return ApiError::repo_not_found(&query.repo).into_response();
    };
    // The document is looked up by the account's own hostname, which is what
    // `did:web` means: the DID *is* the host it is served from.
    let Some(document) = state.registry.did_document(account.did.host()) else {
        return ApiError::new(
            StatusCode::INTERNAL_SERVER_ERROR,
            "DidDocumentUnavailable",
            format!("`{did}` has an account and no DID document"),
        )
        .into_response();
    };
    // Read out of the document rather than off the account, so this reports
    // the handle a resolver would actually find.
    let handle = document.handle().unwrap_or_default().to_owned();
    let handle_is_correct = state
        .registry
        .handle_did(&handle)
        .is_some_and(|resolved| resolved == did);
    let collections = state.registry.collections(&did).unwrap_or_default();
    let did_doc = match serde_json::to_value(&document) {
        Ok(value) => value,
        Err(error) => {
            return ApiError::new(
                StatusCode::INTERNAL_SERVER_ERROR,
                "DidDocumentUnavailable",
                error.to_string(),
            )
            .into_response()
        }
    };
    Json(DescribeRepoResponse {
        handle,
        did,
        did_doc,
        collections,
        handle_is_correct,
    })
    .into_response()
}

/// `GET /xrpc/com.atproto.repo.listRecords`
///
/// Newest first unless `reverse` says otherwise, capped at `limit` and
/// defaulting to [`DEFAULT_LIST_LIMIT`](didbot_pds::DEFAULT_LIST_LIMIT).
/// `cursor` resumes strictly after the record key a previous page ended on,
/// which is what makes a repository readable past the lexicon's ceiling of a
/// hundred.
pub(crate) async fn list_records(
    State(state): State<AppState>,
    query: Result<Query<ListRecordsQuery>, QueryRejection>,
) -> Response {
    // The query is extracted as a `Result` rather than directly, because
    // axum's own rejection is a plain-text body and every failure this server
    // returns is `{"error":..,"message":..}`.
    let Query(query) = match query {
        Ok(query) => query,
        Err(rejection) => return ApiError::bad_request(rejection.body_text()).into_response(),
    };

    let did = match resolve_repo(&state, &query.repo) {
        Ok(did) => did,
        Err(err) => return err.into_response(),
    };
    let page = query.page();
    match state.registry.list_records(&did, &query.collection, &page) {
        Ok(records) => {
            // The cursor is taken before the rows are turned into views,
            // because it is the last row's *record key* and a view holds an
            // `at://` URI. A client hands it straight back.
            let cursor = page.next_cursor(&records);
            // Naming a row can fail now that every row carries a CID, and the
            // whole listing fails rather than the row being dropped: a page
            // that silently omitted a record this repository holds would be a
            // wrong answer that reads like a right one.
            let records: Vec<RecordView> = match records
                .into_iter()
                .map(|(rkey, value)| RecordView::new(&did, &query.collection, &rkey, value))
                .collect::<Result<_, _>>()
            {
                Ok(records) => records,
                Err(err) => {
                    let subject = format!("at://{did}/{}", query.collection);
                    return unnameable_record(&subject, &err).into_response();
                }
            };
            // The field is omitted rather than sent empty when the page did
            // not fill: the lexicon marks it optional so a server can say
            // there is nothing after this page by leaving it out.
            let mut body = json!({ "records": records });
            if let Some(cursor) = cursor {
                body["cursor"] = json!(cursor);
            }
            Json(body).into_response()
        }
        Err(err) => crate::error::repo_surface(&err).into_response(),
    }
}