use std::future::Future;
use std::sync::Arc;

use bobbin_resolver::{
    NormalizeRepoRefs, decode_canon_or_upgrade, normalize_record_fields, scrub_record_bytes,
    upgrade_wire_bytes,
};

use axum::{
    Router,
    body::{Body, Bytes},
    extract::{FromRequestParts, Query, RawQuery, State, rejection::QueryRejection},
    http::{
        HeaderMap, HeaderName, HeaderValue, Method, StatusCode,
        header::{
            ACCEPT_RANGES, AUTHORIZATION, CACHE_CONTROL, CONTENT_DISPOSITION, CONTENT_ENCODING,
            CONTENT_LENGTH, CONTENT_RANGE, CONTENT_SECURITY_POLICY, CONTENT_TYPE, ETAG,
            IF_MODIFIED_SINCE, IF_NONE_MATCH, IF_RANGE, LAST_MODIFIED, RANGE,
            X_CONTENT_TYPE_OPTIONS,
        },
        request::Parts,
    },
    response::{IntoResponse, Json, Response},
    routing::get,
};
pub use bobbin_codesearch::{CodeSearch, CodeSearchError};
use bobbin_edge_index::{
    Coverage, CoverageWatch, CursorParseError, EdgeItem, EdgePage, EdgeStore, InviteIndex,
    InviteScope, IssueStateKind, Offer, PageCursor, PageLimit, PageOffset, PageStart, PageToken,
    ParsedCid, PullStatusKind, Rejection, Settlement, Settlements, SortDir, StateIndex, StateKind,
    StreamHealth, StreamHealthSnapshot, TotalCount, Watch,
};
use bobbin_knot_proxy::{
    KnotHost, KnotProxy, KnotProxyError, MirrorNsid, MirrorProxy, ProxyResponse, RepoSlug,
};
use bobbin_record_lru::RecordStore;
use bobbin_resolver::{IdentityResolveError, IdentityResolver, RepoIdResolver};
use bobbin_runtime::{ReqwestHttp, UnixMicros};
use bobbin_search::{
    ActorIndex, ActorIndexError, SearchCursor, SearchError, SearchFilters, SearchHit, SearchOffset,
    SearchReader,
};
use bobbin_slingshot_client::{SlingshotClient, SlingshotError};
use bobbin_types::edges::REPO_SOURCE_EDGE_KIND;
use bobbin_types::ids::{EdgeKey, SubjectRef, nsid_static, owner_did_from_aturi};
use bobbin_types::knot_acl::{KnotOwnedSource, decode_knot_owned_source};
use bobbin_types::org_tangled::feed::subscription::SubscriptionRecord;
use bobbin_types::record::RecordBody;
use bobbin_types::search::SearchableRecord;
use bobbin_types::sh_tangled::actor::profile::{Profile, ProfileGetRecordOutput, ProfileRecord};
use bobbin_types::sh_tangled::feed::comment::{
    Comment as FeedComment, CommentRecord as FeedCommentRecord,
};
use bobbin_types::sh_tangled::feed::reaction::{Reaction, ReactionRecord};
use bobbin_types::sh_tangled::feed::star::{Star, StarRecord};
use bobbin_types::sh_tangled::graph::follow::{Follow, FollowRecord};
use bobbin_types::sh_tangled::graph::vouch::{Vouch, VouchRecord};
use bobbin_types::sh_tangled::knot::{Knot, KnotRecord};
use bobbin_types::sh_tangled::label::definition::{
    Definition as LabelDefinition, DefinitionRecord as LabelDefinitionRecord,
};
use bobbin_types::sh_tangled::label::op::{Op as LabelOp, OpRecord as LabelOpRecord};
use bobbin_types::sh_tangled::pipeline::status::{
    Status as PipelineStatus, StatusRecord as PipelineStatusRecord,
};
use bobbin_types::sh_tangled::pipeline::{Pipeline, PipelineRecord};
use bobbin_types::sh_tangled::public_key::{PublicKey, PublicKeyGetRecordOutput, PublicKeyRecord};
use bobbin_types::sh_tangled::repo::artifact::{Artifact, ArtifactRecord};
use bobbin_types::sh_tangled::repo::collaborator::{Collaborator, CollaboratorRecord};
use bobbin_types::sh_tangled::repo::issue::state::{
    State as IssueState, StateRecord as IssueStateRecord,
};
use bobbin_types::sh_tangled::repo::issue::{Issue, IssueGetRecordOutput, IssueRecord};
use bobbin_types::sh_tangled::repo::pull::status::{
    Status as PullStatus, StatusRecord as PullStatusRecord,
};
use bobbin_types::sh_tangled::repo::pull::{Pull, PullGetRecordOutput, PullRecord};
use bobbin_types::sh_tangled::repo::{Repo, RepoGetRecordOutput, RepoRecord};
use bobbin_types::sh_tangled::spindle::member::{
    Member as SpindleMember, MemberRecord as SpindleMemberRecord,
};
use bobbin_types::sh_tangled::spindle::{Spindle, SpindleRecord};
use bobbin_types::sh_tangled::string::{
    TangledString, TangledStringGetRecordOutput, TangledStringRecord,
};

use futures::Stream;
use futures::stream::{self, StreamExt, TryStreamExt};
use jacquard_axum::service_auth::{self, ExtractOptionalServiceAuth, ServiceAuthConfig};
use jacquard_common::types::datetime::Datetime;
use jacquard_common::types::did::Did;
use jacquard_common::types::ident::AtIdentifier;
use jacquard_common::types::nsid::Nsid;
use jacquard_common::types::recordkey::Rkey;
use jacquard_common::types::string::{AtUri, Cid};
use jacquard_common::xrpc::XrpcResp;
use jacquard_common::{DefaultStr, IntoStatic};
use jacquard_identity::JacquardResolver;
use serde::{Deserialize, Deserializer, Serialize};
use std::convert::Infallible;
use std::time::Duration;
use thiserror::Error;
use tokio::time::Instant;
use url::form_urlencoded;

use tower_http::classify::ServerErrorsFailureClass;
use tower_http::cors::{AllowOrigin, CorsLayer};
use tower_http::trace::{DefaultMakeSpan, OnFailure, OnResponse, TraceLayer};
use tracing::{Level, Span};

mod actor;
mod backpressure;
mod change_ids;
mod client_address;
mod codesearch;
mod commit_stats;
mod enrich;
mod feed;
mod filter;
mod pocket;
mod profile_activity;
mod recordpath;
mod repo;
mod view;

pub use backpressure::{
    AwaitLimiter, AwaitPermit, HeavyLimiter, HeavyPermit, MaxAwaiting, MaxInFlight,
    PerRequestAnonBytes, PressureVerdict, ReservedFloor,
};
use client_address::X_FORWARDED_FOR;
pub use client_address::{ClientAddress, SocketPeer};
use filter::{
    CountFilter, IssueFilter, ListFilter, NoFilter, PullFilter, StateViewQuery, VouchNetworkFilter,
};
pub use pocket::PocketProxy;
use trusted_proxies::TrustedProxies;

const DEFAULT_LIMIT: u32 = 50;
const ACTOR_TYPEAHEAD_DEFAULT_LIMIT: u32 = 10;
const ACTOR_TYPEAHEAD_MAX_LIMIT: u32 = 100;
const ACTOR_TYPEAHEAD_MAX_QUERY_BYTES: usize = 253;
const FETCH_CONCURRENCY: usize = 8;
const TANGLED_NSID_PREFIX: &str = "sh.tangled.";
const ORG_TANGLED_NSID_PREFIX: &str = "org.tangled.";
const MAX_OUTSTANDING_OFFERS: usize = 128;

pub type Directory = JacquardResolver<ReqwestHttp>;

#[derive(Clone, Debug, Default)]
pub enum Cors {
    #[default]
    Off,
    Any,
    Origins(Vec<HeaderValue>),
}

#[derive(Debug, Error)]
#[error("{0}")]
pub struct CorsOriginError(String);

impl Cors {
    pub fn parse_env(raw: &str) -> Result<Self, CorsOriginError> {
        Self::from_origins(
            raw.split(',')
                .map(str::trim)
                .filter(|entry| !entry.is_empty()),
        )
    }

    fn from_origins<'a>(
        origins: impl IntoIterator<Item = &'a str>,
    ) -> Result<Self, CorsOriginError> {
        let origins: Vec<&str> = origins.into_iter().collect();
        match origins.as_slice() {
            [] => Ok(Self::Off),
            ["*"] => Ok(Self::Any),
            _ if origins.contains(&"*") => Err(CorsOriginError(
                "CORS wildcard `*` cannot be combined with exact origins".into(),
            )),
            _ => origins
                .into_iter()
                .map(|origin| {
                    let url = url::Url::parse(origin).map_err(|e| {
                        CorsOriginError(format!("invalid CORS origin `{origin}`: {e}"))
                    })?;
                    if !matches!(url.scheme(), "http" | "https")
                        || url.origin().ascii_serialization() != origin
                    {
                        return Err(CorsOriginError(format!(
                            "CORS origin must be an exact http(s) origin: `{origin}`"
                        )));
                    }
                    origin
                        .parse()
                        .map_err(|e| CorsOriginError(format!("invalid CORS header: {e}")))
                })
                .collect::<Result<Vec<_>, _>>()
                .map(Self::Origins),
        }
    }
}

impl<'de> Deserialize<'de> for Cors {
    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
        let origins = Vec::<String>::deserialize(deserializer)?;
        Self::from_origins(origins.iter().map(String::as_str)).map_err(serde::de::Error::custom)
    }
}

pub fn default_directory() -> Directory {
    JacquardResolver::new(ReqwestHttp::new(reqwest::Client::new()), Default::default())
}

#[derive(Clone)]
pub struct AppState {
    pub records: Arc<dyn RecordStore>,
    pub slingshot: SlingshotClient,
    pub edges: Arc<EdgeStore>,
    pub issue_states: Arc<StateIndex<IssueStateKind>>,
    pub pull_statuses: Arc<StateIndex<PullStatusKind>>,
    pub coverage: Arc<CoverageWatch>,
    pub knots: Arc<KnotProxy>,
    pub mirror: Option<Arc<MirrorProxy>>,
    pub mirror_v2: Option<Arc<MirrorProxy>>,
    pub search: Arc<dyn SearchReader>,
    /// Zoekt-backed code search. `None` when unconfigured, and the endpoint answers 501.
    pub codesearch: Option<Arc<CodeSearch>>,
    /// Pocket, holding per-account preferences. `None` when unconfigured, and the
    /// two `com.bad-example.pocket.*` endpoints answer 501.
    pub pocket: Option<Arc<PocketProxy>>,
    pub actors: Arc<ActorIndex>,
    pub resolver: Arc<RepoIdResolver>,
    pub identity: Arc<IdentityResolver>,
    pub directory: Arc<Directory>,
    pub limiter: Option<Arc<HeavyLimiter>>,
    pub awaiting: Arc<AwaitLimiter>,
    pub client_address: Arc<ClientAddress>,
    pub settlements: Arc<Settlements>,
    pub invites: Arc<InviteIndex>,
    /// the ingest stream's liveness, absent when nothing publishes it
    pub stream_health: Option<Arc<StreamHealth>>,
    cors: Cors,
    service_auth_config: service_auth::ServiceAuthConfig<Arc<Directory>>,
    enrich_router: Arc<std::sync::OnceLock<Router>>,
    trending_cache: Arc<tokio::sync::Mutex<Option<(Instant, actor::TrendingCandidates)>>>,
    commit_stats_cache: Arc<commit_stats::CommitStatsCache>,
    change_ids: Arc<change_ids::ChangeIdCache>,
}

impl AppState {
    #[allow(clippy::too_many_arguments)]
    pub fn new(
        records: Arc<dyn RecordStore>,
        slingshot: SlingshotClient,
        edges: Arc<EdgeStore>,
        issue_states: Arc<StateIndex<IssueStateKind>>,
        pull_statuses: Arc<StateIndex<PullStatusKind>>,
        coverage: Arc<CoverageWatch>,
        knots: Arc<KnotProxy>,
        search: Arc<dyn SearchReader>,
        resolver: Arc<RepoIdResolver>,
        directory: Arc<Directory>,
    ) -> Self {
        let identity = Arc::new(IdentityResolver::with_slingshot(
            slingshot.clone(),
            Arc::new(bobbin_runtime::SystemClock::new()),
            bobbin_runtime::RuntimeHasher::default(),
        ));
        Self {
            records,
            slingshot,
            edges,
            issue_states,
            pull_statuses,
            coverage,
            knots,
            mirror: None,
            mirror_v2: None,
            search,
            codesearch: None,
            pocket: None,
            actors: Arc::new(ActorIndex::new()),
            resolver,
            identity,
            directory: directory.clone(),
            limiter: None,
            awaiting: Arc::new(AwaitLimiter::new(MaxAwaiting::default())),
            client_address: Arc::new(ClientAddress::default()),
            settlements: Arc::new(Settlements::new()),
            invites: Arc::new(InviteIndex::new()),
            stream_health: None,
            cors: Cors::Off,
            service_auth_config: ServiceAuthConfig::new(
                Did::new_static("did:web:localhost").unwrap(),
                directory,
            ),
            enrich_router: Arc::new(std::sync::OnceLock::new()),
            trending_cache: Arc::new(tokio::sync::Mutex::new(None)),
            commit_stats_cache: Arc::new(commit_stats::CommitStatsCache::default()),
            change_ids: Arc::new(change_ids::ChangeIdCache::default()),
        }
    }

    pub fn with_actors(mut self, actors: Arc<ActorIndex>) -> Self {
        self.actors = actors;
        self
    }
    pub fn with_limiter(mut self, limiter: Option<Arc<HeavyLimiter>>) -> Self {
        self.limiter = limiter;
        self
    }

    pub fn with_max_awaiting(mut self, max: MaxAwaiting) -> Self {
        self.awaiting = Arc::new(AwaitLimiter::new(max));
        self
    }

    pub fn with_mirror(mut self, mirror: Option<Arc<MirrorProxy>>) -> Self {
        self.mirror = mirror;
        self
    }

    pub fn with_mirror_v2(mut self, mirror_v2: Option<Arc<MirrorProxy>>) -> Self {
        self.mirror_v2 = mirror_v2;
        self
    }

    pub fn with_codesearch(mut self, codesearch: Option<Arc<CodeSearch>>) -> Self {
        self.codesearch = codesearch;
        self
    }

    pub fn with_pocket(mut self, pocket: Option<Arc<PocketProxy>>) -> Self {
        self.pocket = pocket;
        self
    }

    pub fn with_identity(mut self, identity: Arc<IdentityResolver>) -> Self {
        self.identity = identity;
        self
    }

    pub fn with_proxies(mut self, proxies: TrustedProxies) -> Self {
        self.client_address = Arc::new(ClientAddress::new(proxies));
        self
    }

    pub fn with_settlements(mut self, settlements: Arc<Settlements>) -> Self {
        self.settlements = settlements;
        self
    }

    pub fn with_invites(mut self, invites: Arc<InviteIndex>) -> Self {
        self.invites = invites;
        self
    }

    pub fn with_stream_health(mut self, stream_health: Arc<StreamHealth>) -> Self {
        self.stream_health = Some(stream_health);
        self
    }

    pub fn with_cors(mut self, cors: Cors) -> Self {
        self.cors = cors;
        self
    }

    pub fn with_service_did(mut self, did: Did) -> Self {
        self.service_auth_config = ServiceAuthConfig::new(did, self.directory.clone());
        self
    }

    /// for internal xrpc dispatch [`enrich`]
    pub fn self_router(&self) -> Router {
        self.enrich_router
            .get_or_init(|| router(self.clone()))
            .clone()
    }

    pub(crate) fn heavy_permit(&self) -> Result<Option<HeavyPermit>, XrpcError> {
        self.limiter.as_ref().map(|l| l.try_enter()).transpose()
    }
}

impl service_auth::ServiceAuth for AppState {
    type Resolver = Directory;

    fn service_did(&self) -> Did<&str> {
        self.service_auth_config.service_did()
    }
    fn resolver(&self) -> &Self::Resolver {
        self.service_auth_config.resolver()
    }
    fn require_lxm(&self) -> bool {
        service_auth::ServiceAuth::require_lxm(&self.service_auth_config)
    }
    fn allowed_services(&self) -> &[jacquard_common::SmolStr] {
        self.service_auth_config.allowed_services()
    }
    fn replay_protection_enabled(&self) -> bool {
        self.service_auth_config.replay_protection_enabled()
    }
    fn replay_store(&self) -> &dyn jacquard_axum::service_auth::ReplayStore {
        self.service_auth_config.replay_store()
    }
}

pub fn router(state: AppState) -> Router {
    let cors = match &state.cors {
        Cors::Off => None,
        Cors::Any => Some(AllowOrigin::any()),
        Cors::Origins(origins) => Some(AllowOrigin::list(origins.clone())),
    }
    .map(|origins| {
        CorsLayer::new()
            .allow_origin(origins)
            .allow_methods([Method::GET, Method::POST, Method::OPTIONS])
            .allow_headers([
                AUTHORIZATION,
                CONTENT_TYPE,
                HeaderName::from_static("atproto-proxy"),
                HeaderName::from_static("atproto-accept-labelers"),
                HeaderName::from_static("range"),
                HeaderName::from_static("if-range"),
                HeaderName::from_static("if-none-match"),
                HeaderName::from_static("if-modified-since"),
            ])
            .expose_headers([
                CONTENT_TYPE,
                HeaderName::from_static("ratelimit-limit"),
                HeaderName::from_static("ratelimit-remaining"),
                HeaderName::from_static("ratelimit-reset"),
            ])
            .max_age(Duration::from_secs(86400))
    });
    let router = Router::new()
        .route("/xrpc/sh.tangled.repo.getRepo", get(get_repo))
        .route("/xrpc/sh.tangled.repo.getRepos", get(get_repos))
        .route(
            "/xrpc/sh.tangled.repo.getRepoByRepoDid",
            get(get_repo_by_repo_did),
        )
        .route(
            "/xrpc/sh.tangled.repo.getReposByRepoDids",
            get(get_repos_by_repo_dids),
        )
        .route("/xrpc/sh.tangled.repo.getRepoByName", get(get_repo_by_name))
        .route("/xrpc/sh.tangled.actor.getProfile", get(get_profile))
        .route("/xrpc/sh.tangled.actor.getProfiles", get(get_profiles))
        .route(
            "/xrpc/sh.tangled.actor.getProfileActivity",
            get(profile_activity::get_profile_activity),
        )
        .route(
            "/xrpc/sh.tangled.actor.getTrending",
            get(actor::get_trending),
        )
        .route("/xrpc/sh.tangled.repo.getIssue", get(get_issue))
        .route("/xrpc/sh.tangled.repo.getIssues", get(get_issues))
        .route("/xrpc/sh.tangled.repo.getPull", get(get_pull))
        .route("/xrpc/sh.tangled.repo.getPulls", get(get_pulls))
        .route("/xrpc/sh.tangled.feed.listStars", get(list_stars))
        .route("/xrpc/sh.tangled.feed.countStars", get(count_stars))
        .route("/xrpc/sh.tangled.feed.getStar", get(get_star))
        .route("/xrpc/sh.tangled.graph.listFollows", get(list_follows))
        .route("/xrpc/sh.tangled.graph.countFollows", get(count_follows))
        .route("/xrpc/sh.tangled.graph.getFollow", get(get_follow))
        .route("/xrpc/sh.tangled.repo.listIssues", get(list_issues))
        .route("/xrpc/sh.tangled.repo.countIssues", get(count_issues))
        .route("/xrpc/sh.tangled.repo.listPulls", get(list_pulls))
        .route("/xrpc/sh.tangled.repo.countPulls", get(count_pulls))
        .route(
            "/xrpc/sh.tangled.feed.listComments",
            get(feed::list_comments),
        )
        .route(
            "/xrpc/sh.tangled.feed.countComments",
            get(count_feed_comments),
        )
        .route("/xrpc/sh.tangled.feed.listReactions", get(list_reactions))
        .route("/xrpc/sh.tangled.feed.countReactions", get(count_reactions))
        .route(
            "/xrpc/sh.tangled.repo.listCollaborators",
            get(list_collaborators),
        )
        .route(
            "/xrpc/sh.tangled.repo.countCollaborators",
            get(count_collaborators),
        )
        .route(
            "/xrpc/sh.tangled.repo.issue.listStates",
            get(list_issue_states),
        )
        .route(
            "/xrpc/sh.tangled.repo.issue.countStates",
            get(count_issue_states),
        )
        .route(
            "/xrpc/sh.tangled.repo.pull.listStatuses",
            get(list_pull_statuses),
        )
        .route(
            "/xrpc/sh.tangled.repo.pull.countStatuses",
            get(count_pull_statuses),
        )
        .route("/xrpc/sh.tangled.repo.listRepos", get(list_repos))
        .route("/xrpc/sh.tangled.repo.countRepos", get(count_repos))
        .route("/xrpc/sh.tangled.repo.countForks", get(count_forks))
        .route("/xrpc/sh.tangled.knot.listKnots", get(list_knots))
        .route("/xrpc/sh.tangled.knot.countKnots", get(count_knots))
        .route("/xrpc/sh.tangled.spindle.listSpindles", get(list_spindles))
        .route(
            "/xrpc/sh.tangled.spindle.countSpindles",
            get(count_spindles),
        )
        .route(
            "/xrpc/sh.tangled.publicKey.getPublicKey",
            get(get_public_key),
        )
        .route("/xrpc/sh.tangled.publicKey.listKeys", get(list_public_keys))
        .route(
            "/xrpc/sh.tangled.publicKey.countKeys",
            get(count_public_keys),
        )
        .route("/xrpc/sh.tangled.feed.getTimeline", get(feed::get_timeline))
        .route(
            "/xrpc/sh.tangled.graph.listNetworkVouches",
            get(list_network_vouches),
        )
        .route("/xrpc/sh.tangled.feed.listStarsBy", get(list_stars_by))
        .route("/xrpc/sh.tangled.feed.countStarsBy", get(count_stars_by))
        .route(
            "/xrpc/sh.tangled.feed.listReactionsBy",
            get(list_reactions_by),
        )
        .route(
            "/xrpc/sh.tangled.feed.countReactionsBy",
            get(count_reactions_by),
        )
        .route("/xrpc/sh.tangled.graph.listFollowsBy", get(list_follows_by))
        .route(
            "/xrpc/sh.tangled.graph.countFollowsBy",
            get(count_follows_by),
        )
        .route(
            "/xrpc/sh.tangled.repo.listCollaboratorInvitesBy",
            get(list_collaborator_invites_by),
        )
        .route("/xrpc/sh.tangled.label.listOpsBy", get(list_label_ops_by))
        .route("/xrpc/sh.tangled.label.countOpsBy", get(count_label_ops_by))
        .route(
            "/xrpc/sh.tangled.pipeline.listPipelinesBy",
            get(list_pipelines_by),
        )
        .route(
            "/xrpc/sh.tangled.pipeline.countPipelinesBy",
            get(count_pipelines_by),
        )
        .route(
            "/xrpc/sh.tangled.pipeline.listStatusesBy",
            get(list_pipeline_statuses_by),
        )
        .route(
            "/xrpc/sh.tangled.pipeline.countStatusesBy",
            get(count_pipeline_statuses_by),
        )
        .route(
            "/xrpc/sh.tangled.repo.listArtifactsBy",
            get(list_artifacts_by),
        )
        .route(
            "/xrpc/sh.tangled.repo.countArtifactsBy",
            get(count_artifacts_by),
        )
        .route(
            "/xrpc/sh.tangled.repo.listCollaboratorsBy",
            get(list_collaborators_by),
        )
        .route(
            "/xrpc/sh.tangled.repo.countCollaboratorsBy",
            get(count_collaborators_by),
        )
        .route("/xrpc/sh.tangled.repo.listIssuesBy", get(list_issues_by))
        .route("/xrpc/sh.tangled.repo.countIssuesBy", get(count_issues_by))
        .route(
            "/xrpc/sh.tangled.feed.listCommentsBy",
            get(list_feed_comments_by),
        )
        .route(
            "/xrpc/sh.tangled.feed.countCommentsBy",
            get(count_feed_comments_by),
        )
        .route(
            "/xrpc/sh.tangled.repo.issue.listStatesBy",
            get(list_issue_states_by),
        )
        .route(
            "/xrpc/sh.tangled.repo.issue.countStatesBy",
            get(count_issue_states_by),
        )
        .route("/xrpc/sh.tangled.repo.listPullsBy", get(list_pulls_by))
        .route("/xrpc/sh.tangled.repo.countPullsBy", get(count_pulls_by))
        .route(
            "/xrpc/sh.tangled.repo.pull.listStatusesBy",
            get(list_pull_statuses_by),
        )
        .route(
            "/xrpc/sh.tangled.repo.pull.countStatusesBy",
            get(count_pull_statuses_by),
        )
        .route(
            "/xrpc/sh.tangled.pull.getPullView",
            get(repo::get_pull_view),
        )
        // .route("/xrpc/sh.tangled.pull.listPullViews", get(repo::list_pull_views))
        .route(
            "/xrpc/sh.tangled.spindle.listMembersBy",
            get(list_spindle_members_by),
        )
        .route(
            "/xrpc/sh.tangled.spindle.countMembersBy",
            get(count_spindle_members_by),
        )
        .route(
            "/xrpc/sh.tangled.label.listDefinitions",
            get(list_label_definitions),
        )
        .route(
            "/xrpc/sh.tangled.label.countDefinitions",
            get(count_label_definitions),
        )
        .route("/xrpc/sh.tangled.label.listOps", get(list_label_ops))
        .route("/xrpc/sh.tangled.label.countOps", get(count_label_ops))
        .route(
            "/xrpc/sh.tangled.pipeline.listPipelines",
            get(list_pipelines),
        )
        .route(
            "/xrpc/sh.tangled.pipeline.countPipelines",
            get(count_pipelines),
        )
        .route(
            "/xrpc/sh.tangled.pipeline.listStatuses",
            get(list_pipeline_statuses),
        )
        .route(
            "/xrpc/sh.tangled.pipeline.countStatuses",
            get(count_pipeline_statuses),
        )
        .route("/xrpc/sh.tangled.repo.listArtifacts", get(list_artifacts))
        .route("/xrpc/sh.tangled.repo.countArtifacts", get(count_artifacts))
        .route(
            "/xrpc/sh.tangled.spindle.listMembers",
            get(list_spindle_members),
        )
        .route(
            "/xrpc/sh.tangled.spindle.countMembers",
            get(count_spindle_members),
        )
        .route("/xrpc/sh.tangled.string.getString", get(get_string))
        .route("/xrpc/sh.tangled.string.listStrings", get(list_strings))
        .route("/xrpc/sh.tangled.string.countStrings", get(count_strings))
        .route("/xrpc/sh.tangled.search.query", get(search_query))
        .route(
            "/xrpc/org.tangled.temp.search.searchCode",
            get(codesearch::search_code),
        )
        .route(
            "/xrpc/sh.tangled.query.enrichResponse",
            axum::routing::post(enrich::enrich),
        )
        .route(
            "/xrpc/sh.tangled.actor.searchActorsTypeahead",
            get(search_actors_typeahead),
        )
        .route("/xrpc/sh.tangled.bobbin.getCoverage", get(get_coverage))
        .route("/xrpc/sh.tangled.bobbin.awaitRecord", get(await_record))
        .route(
            "/xrpc/org.tangled.temp.notification.listRecipients",
            get(list_recipients),
        )
        .route(
            "/xrpc/org.tangled.feed.subscription.getForActor",
            get(get_subscription_for_actor),
        )
        .route(
            "/xrpc/blue.microcosm.identity.resolveMiniDoc",
            get(resolve_mini_doc),
        )
        .route(
            &format!("/xrpc/{}", pocket::GET_PREFERENCES),
            get(pocket::get_preferences),
        )
        .route(
            &format!("/xrpc/{}", pocket::PUT_PREFERENCES),
            axum::routing::post(pocket::put_preferences),
        )
        .merge(knot_proxied_routes())
        .layer(
            TraceLayer::new_for_http()
                .make_span_with(DefaultMakeSpan::new().level(Level::INFO))
                .on_request(())
                .on_response(LatencyFreeTrace)
                .on_failure(LatencyFreeTrace),
        );
    let router = match cors {
        Some(cors) => router.layer(cors),
        None => router,
    };
    router.with_state(state)
}

#[derive(Clone, Copy, Debug)]
struct LatencyFreeTrace;

impl<B> OnResponse<B> for LatencyFreeTrace {
    fn on_response(self, response: &Response<B>, _latency: Duration, _span: &Span) {
        tracing::event!(
            target: "tower_http::trace::on_response",
            Level::INFO,
            status = response.status().as_u16(),
            "request completed",
        );
    }
}

impl OnFailure<ServerErrorsFailureClass> for LatencyFreeTrace {
    fn on_failure(&mut self, error: ServerErrorsFailureClass, _latency: Duration, _span: &Span) {
        tracing::event!(
            target: "tower_http::trace::on_failure",
            Level::WARN,
            error = %error,
            "request failed",
        );
    }
}

const REPO_PROXIED_NSIDS: &[&str] = &[
    "sh.tangled.repo.archive",
    "sh.tangled.repo.blame",
    "sh.tangled.repo.blob",
    "sh.tangled.repo.branch",
    "sh.tangled.repo.branches",
    "sh.tangled.repo.compare",
    "sh.tangled.repo.describeRepo",
    "sh.tangled.repo.diff",
    "sh.tangled.repo.getDefaultBranch",
    "sh.tangled.repo.languages",
    "sh.tangled.repo.listSecrets",
    "sh.tangled.repo.log",
    "sh.tangled.repo.tag",
    "sh.tangled.repo.tags",
    "sh.tangled.repo.tree",
];

const KNOT_PROXIED_NSIDS: &[&str] = &[
    "sh.tangled.owner",
    "sh.tangled.knot.version",
    "sh.tangled.knot.listKeys",
];

const MIRROR_V2_PROXIED_NSIDS: &[&str] = &[
    "sh.tangled.git.temp2.getBlame",
    "sh.tangled.git.temp2.getDiff",
    "sh.tangled.git.temp2.getInterdiff",
    "sh.tangled.git.temp2.listCommits",
    "sh.tangled.git.temp2.mergeCheck",
    "org.tangled.temp.git.getBlob",
    "org.tangled.temp.git.getBranch",
    "org.tangled.temp.git.getEntry",
    "org.tangled.temp.git.getMergeBase",
    "org.tangled.temp.git.getTree",
];

const MIRROR_V2_WRITE_NSIDS: &[&str] = &[
    "sh.tangled.git.mergeCommit",
    "org.tangled.temp.git.deleteBranch",
];

const PASSTHROUGH_HEADERS: &[&HeaderName] = &[
    &CONTENT_TYPE,
    &CONTENT_LENGTH,
    &CONTENT_ENCODING,
    &ETAG,
    &CACHE_CONTROL,
    &LAST_MODIFIED,
    &CONTENT_DISPOSITION,
    &ACCEPT_RANGES,
    &CONTENT_RANGE,
    &X_CONTENT_TYPE_OPTIONS,
    &CONTENT_SECURITY_POLICY,
];

const RANGE_OR_CONDITIONAL_HEADERS: &[&HeaderName] =
    &[&RANGE, &IF_RANGE, &IF_NONE_MATCH, &IF_MODIFIED_SINCE];

const KNOT_HOST_PARAM: &str = "knot";
const REPO_PARAM: &str = "repo";

type ProxyParams = Vec<(String, String)>;

fn knot_proxied_routes() -> Router<AppState> {
    let with_repo = register_proxied(Router::new(), REPO_PROXIED_NSIDS, proxy_repo_handler);
    let with_knot = register_proxied(with_repo, KNOT_PROXIED_NSIDS, proxy_knot_handler);
    let with_mirror = register_proxied(with_knot, MIRROR_V2_PROXIED_NSIDS, proxy_mirror_v2_handler);
    register_mirror_v2_writes(with_mirror)
}

fn register_mirror_v2_writes(router: Router<AppState>) -> Router<AppState> {
    MIRROR_V2_WRITE_NSIDS
        .iter()
        .fold(router, |router, &nsid_lit| {
            let nsid = nsid_static(nsid_lit);
            router.route(
                &format!("/xrpc/{nsid_lit}"),
                axum::routing::post(
                    move |State(state): State<AppState>, headers: HeaderMap, body: Bytes| {
                        proxy_mirror_v2_write_handler(state, headers, body, nsid.clone())
                    },
                ),
            )
        })
}

fn register_proxied<H, Fut>(
    router: Router<AppState>,
    nsids: &[&'static str],
    handler: H,
) -> Router<AppState>
where
    H: Fn(AppState, HeaderMap, SocketPeer, ProxyParams, Nsid<DefaultStr>) -> Fut
        + Clone
        + Send
        + Sync
        + 'static,
    Fut: Future<Output = Result<Response, XrpcError>> + Send + 'static,
{
    nsids.iter().fold(router, |router, &nsid_lit| {
        let handler = handler.clone();
        let nsid = nsid_static(nsid_lit);
        router.route(
            &format!("/xrpc/{nsid_lit}"),
            get(
                move |State(state): State<AppState>,
                      headers: HeaderMap,
                      socket: SocketPeer,
                      Query(params): Query<ProxyParams>| {
                    handler(state, headers, socket, params, nsid.clone())
                },
            ),
        )
    })
}

#[derive(Clone, Debug)]
pub enum SubjectQuery {
    Did(Did<DefaultStr>),
    Uri(AtUri<DefaultStr>),
}

impl<'de> Deserialize<'de> for SubjectQuery {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: serde::Deserializer<'de>,
    {
        let raw = String::deserialize(deserializer)?;
        if let Ok(did) = Did::<DefaultStr>::new_owned(&raw) {
            return Ok(Self::Did(did));
        }
        AtUri::<DefaultStr>::new_owned(&raw)
            .map(Self::Uri)
            .map_err(serde::de::Error::custom)
    }
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ExpectedNsid {
    canon: Nsid<DefaultStr>,
    aliases: &'static [&'static str],
}

const FEED_COMMENT_LEGACY_ALIASES: &[&str] = &[
    "sh.tangled.repo.issue.comment",
    "sh.tangled.repo.pull.comment",
];

fn aliases_for(nsid: &str) -> &'static [&'static str] {
    match nsid {
        "sh.tangled.feed.comment" => FEED_COMMENT_LEGACY_ALIASES,
        _ => &[],
    }
}

impl ExpectedNsid {
    pub fn new(nsid: Nsid<DefaultStr>) -> Self {
        let aliases = aliases_for(nsid.as_ref());
        Self {
            canon: nsid,
            aliases,
        }
    }

    pub fn from_static(s: &'static str) -> Self {
        let canon = nsid_static(s);
        let aliases = aliases_for(s);
        Self { canon, aliases }
    }

    pub fn as_nsid(&self) -> &Nsid<DefaultStr> {
        &self.canon
    }

    pub fn as_str(&self) -> &str {
        self.canon.as_ref()
    }

    fn accepts(&self, other: &str) -> bool {
        other == self.canon.as_ref() || self.aliases.contains(&other)
    }
}

#[derive(Debug, Deserialize)]
struct GetRepoQuery {
    repo: AtUri<DefaultStr>,
}

#[derive(Debug, Deserialize)]
struct GetRepoByRepoDidQuery {
    #[serde(rename = "repoDid")]
    repo_did: Did<DefaultStr>,
}

#[derive(Debug, Deserialize)]
struct GetRepoByNameQuery {
    owner: Did<DefaultStr>,
    name: String,
}

#[derive(Debug, Deserialize)]
struct GetProfileQuery {
    actor: AtUri<DefaultStr>,
}

#[derive(Debug, Deserialize)]
struct GetIssueQuery {
    issue: AtUri<DefaultStr>,
}

#[derive(Debug, Deserialize)]
struct GetPullQuery {
    pull: AtUri<DefaultStr>,
}

#[derive(Debug, Deserialize)]
struct GetStringQuery {
    uri: AtUri<DefaultStr>,
}

#[derive(Debug, Deserialize)]
struct GetPublicKeyQuery {
    #[serde(rename = "publicKey")]
    public_key: AtUri<DefaultStr>,
}

#[derive(Clone, Copy, Debug, Default, Deserialize, Eq, PartialEq)]
#[serde(rename_all = "lowercase")]
enum Order {
    Asc,
    #[default]
    Desc,
}

impl From<Order> for SortDir {
    fn from(o: Order) -> Self {
        match o {
            Order::Asc => SortDir::Asc,
            Order::Desc => SortDir::Desc,
        }
    }
}

#[derive(Debug, Deserialize)]
struct TypedListQuery<F> {
    subject: SubjectQuery,
    cursor: Option<String>,
    offset: Option<PageOffset>,
    limit: Option<u32>,
    #[serde(default)]
    order: Order,
    #[serde(flatten)]
    filter: F,
}

impl<F> TypedListQuery<F> {
    fn dir(&self) -> SortDir {
        self.order.into()
    }
}

#[derive(Debug, Deserialize)]
struct CountQuery {
    subject: SubjectQuery,
}

#[derive(Debug, Deserialize)]
struct TypedCountQuery<F> {
    subject: SubjectQuery,
    #[serde(flatten)]
    filter: F,
}

#[derive(Debug, Deserialize)]
struct GetEdgeQuery {
    actor: Did<DefaultStr>,
    subject: SubjectQuery,
}

#[derive(Debug, Deserialize)]
struct SearchQueryParams {
    q: String,
    nsid: Option<Nsid<DefaultStr>>,
    author: Option<Did<DefaultStr>>,
    repo: Option<Did<DefaultStr>>,
    since: Option<String>,
    until: Option<String>,
    cursor: Option<String>,
    offset: Option<PageOffset>,
    limit: Option<u32>,
}

#[derive(Debug, Deserialize)]
struct ActorTypeaheadParams {
    q: String,
    limit: Option<u32>,
}

pub struct XrpcQuery<T>(pub T);

impl<S, T> FromRequestParts<S> for XrpcQuery<T>
where
    S: Send + Sync,
    T: serde::de::DeserializeOwned + Send + 'static,
{
    type Rejection = XrpcError;

    async fn from_request_parts(parts: &mut Parts, state: &S) -> Result<Self, Self::Rejection> {
        Query::<T>::from_request_parts(parts, state)
            .await
            .map(|Query(t)| Self(t))
            .map_err(|rej: QueryRejection| XrpcError::InvalidParams(rej.body_text()))
    }
}

#[derive(Debug, Error)]
pub enum XrpcError {
    #[error("invalid request: {0}")]
    InvalidParams(String),
    #[error("authentication required: {0}")]
    AuthRequired(String),
    // /// A service-auth token failed verification. Typed so the jwt-level checks speak jacquard's
    // /// vocabulary; adopting `jacquard-axum` later replaces the producer, not this variant.
    // #[error(transparent)]
    // Auth(#[from] jacquard_common::service_auth::ServiceAuthError),
    #[error("record not found")]
    NotFound,
    #[error("upstream unavailable: {0}")]
    UpstreamUnavailable(String),
    #[error("upstream gone: {0}")]
    UpstreamGone(String),
    #[error("invalid record: {0}")]
    InvalidRecord(String),
    #[error("internal: {0}")]
    Internal(String),
    #[error("overloaded, shedding under memory pressure")]
    Overloaded,
    #[error("invite index is still warming, so it can't list offers yet")]
    Warming,
    #[error("record has not been ingested yet")]
    NotSettled,
    #[error("not implemented: {0}")]
    NotImplemented(&'static str),
}

impl XrpcError {
    pub fn overloaded() -> Self {
        Self::Overloaded
    }
}

#[derive(Serialize)]
struct ErrorBody {
    error: &'static str,
    message: String,
}

impl IntoResponse for XrpcError {
    fn into_response(self) -> Response {
        let (status, error) = match &self {
            Self::InvalidParams(_) => (StatusCode::BAD_REQUEST, "InvalidRequest"),
            Self::AuthRequired(_) => (StatusCode::UNAUTHORIZED, "AuthenticationRequired"),
            Self::NotFound => (StatusCode::NOT_FOUND, "RecordNotFound"),
            Self::UpstreamUnavailable(_) => (StatusCode::BAD_GATEWAY, "UpstreamFailed"),
            Self::UpstreamGone(_) => (StatusCode::BAD_GATEWAY, "UpstreamGone"),
            Self::InvalidRecord(_) => (StatusCode::BAD_GATEWAY, "InvalidRecord"),
            Self::Internal(_) => (StatusCode::INTERNAL_SERVER_ERROR, "InternalError"),
            Self::Overloaded => (StatusCode::SERVICE_UNAVAILABLE, "Overloaded"),
            Self::Warming => (StatusCode::SERVICE_UNAVAILABLE, "Warming"),
            Self::NotSettled => (StatusCode::GATEWAY_TIMEOUT, "NotSettled"),
            Self::NotImplemented(_) => (StatusCode::NOT_IMPLEMENTED, "MethodNotImplemented"),
        };
        let body = ErrorBody {
            error,
            message: self.to_string(),
        };
        (status, Json(body)).into_response()
    }
}

#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
struct CoverageEnvelope {
    ready: bool,
    events_processed: u64,
    last_cursor: u64,
    #[serde(skip_serializing_if = "Option::is_none")]
    stream: Option<StreamEnvelope>,
}

impl From<Coverage> for CoverageEnvelope {
    fn from(c: Coverage) -> Self {
        Self {
            ready: c.is_ready(),
            events_processed: c.events_processed(),
            last_cursor: c.last_cursor().raw(),
            stream: None,
        }
    }
}

/// a wedged ingest shows up as a state that never moved on
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
struct StreamEnvelope {
    state: &'static str,
    state_since: String,
    last_error: Option<Box<str>>,
}

impl From<StreamHealthSnapshot> for StreamEnvelope {
    fn from(snapshot: StreamHealthSnapshot) -> Self {
        Self {
            state: snapshot.state.as_str(),
            state_since: rfc3339_millis(snapshot.state_since_unix_micros),
            last_error: snapshot.last_error,
        }
    }
}

fn rfc3339_millis(stamp: UnixMicros) -> String {
    i64::try_from(stamp.raw())
        .ok()
        .and_then(chrono::DateTime::<chrono::Utc>::from_timestamp_micros)
        .map(|at| at.to_rfc3339_opts(chrono::SecondsFormat::Millis, true))
        .unwrap_or_default()
}

struct Deduped<T>(T);

impl<T: Serialize> Serialize for Deduped<T> {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: serde::Serializer,
    {
        serde_json::to_value(&self.0)
            .map_err(serde::ser::Error::custom)?
            .serialize(serializer)
    }
}

#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
struct RecordView<V> {
    uri: AtUri<DefaultStr>,
    #[serde(skip_serializing_if = "Option::is_none")]
    cid: Option<Cid<DefaultStr>>,
    value: V,
}

#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
struct NetworkVouchItem<V> {
    actor: Did<DefaultStr>,
    #[serde(flatten)]
    view: RecordView<V>,
}

#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
struct CollaboratorListItem {
    subject: Did<DefaultStr>,
    added_by: Did<DefaultStr>,
    created_at: Datetime,
    effective_since: Datetime,
    #[serde(skip_serializing_if = "Option::is_none")]
    uri: Option<AtUri<DefaultStr>>,
    #[serde(skip_serializing_if = "Option::is_none")]
    cid: Option<Cid<DefaultStr>>,
}

#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
struct StatefulItem<V> {
    #[serde(flatten)]
    view: RecordView<V>,
    state: &'static str,
    #[serde(skip_serializing_if = "Option::is_none")]
    state_updated_at: Option<String>,
    comment_count: u64,
}

fn format_micros(micros: u64) -> String {
    let signed = i64::try_from(micros).ok();
    let rfc = signed
        .and_then(chrono::DateTime::<chrono::Utc>::from_timestamp_micros)
        .map(|dt| dt.to_rfc3339_opts(chrono::SecondsFormat::Micros, true));
    rfc.unwrap_or_else(|| micros.to_string())
}

pub(crate) fn source_authority_did(source: &AtUri<DefaultStr>) -> Option<Did<DefaultStr>> {
    match source.authority() {
        AtIdentifier::Did(d) => Some(d.clone().into_static()),
        AtIdentifier::Handle(_) => None,
    }
}

fn enrich_issue_view(
    state: &AppState,
    view: RecordView<Issue<DefaultStr>>,
) -> StatefulItem<Issue<DefaultStr>> {
    let issue_author = source_authority_did(&view.uri);
    let repo_did = view.value.repo.clone();
    enrich_view(
        &state.edges,
        nsid_static("sh.tangled.feed.comment"),
        &state.issue_states,
        view,
        move |src| accept_state_source(src, issue_author.as_ref(), &repo_did),
    )
}

fn enrich_pull_view(
    state: &AppState,
    view: RecordView<Pull<DefaultStr>>,
) -> StatefulItem<Pull<DefaultStr>> {
    let pull_author = source_authority_did(&view.uri);
    let target_repo = view.value.target.repo.clone();
    enrich_view(
        &state.edges,
        nsid_static("sh.tangled.feed.comment"),
        &state.pull_statuses,
        view,
        move |src| accept_state_source(src, pull_author.as_ref(), &target_repo),
    )
}

pub(crate) fn accept_state_source(
    source: &AtUri<DefaultStr>,
    entity_author: Option<&Did<DefaultStr>>,
    repo_owner: &Did<DefaultStr>,
) -> bool {
    let Some(src) = source_authority_did(source) else {
        return false;
    };
    Some(&src) == entity_author || &src == repo_owner
}

fn enrich_view<V, K, F>(
    edges: &EdgeStore,
    comment_nsid: Nsid<DefaultStr>,
    states: &StateIndex<K>,
    view: RecordView<V>,
    accept: F,
) -> StatefulItem<V>
where
    K: StateKind + Default,
    F: Fn(&AtUri<DefaultStr>) -> bool,
{
    let comment_count = edges.count(&EdgeKey::new(
        comment_nsid,
        SubjectRef::Uri(view.uri.clone()),
    ));
    let (state, state_updated_at) = states
        .latest_by(&view.uri, accept)
        .map_or((K::default().wire(), None), |(kind, micros)| {
            (kind.wire(), Some(format_micros(micros)))
        });
    StatefulItem {
        view,
        state,
        state_updated_at,
        comment_count,
    }
}

#[derive(Clone, Copy, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct CountResponse {
    count: u64,
    distinct_authors: u64,
}

#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
struct EdgeUriResponse {
    uri: AtUri<DefaultStr>,
}

#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
struct SearchHitView {
    uri: AtUri<DefaultStr>,
    cid: Option<Cid<DefaultStr>>,
    nsid: Nsid<DefaultStr>,
    score: f32,
    value: SearchableRecord,
}

#[derive(Serialize)]
struct ActorTypeaheadResponse {
    actors: Vec<ActorTypeaheadView>,
}

#[derive(Serialize)]
struct ActorTypeaheadView {
    uri: AtUri<DefaultStr>,
    did: Did<DefaultStr>,
    handle: String,
}

fn map_slingshot(err: SlingshotError) -> XrpcError {
    use SlingshotError as E;
    match err {
        E::NotFound => XrpcError::NotFound,
        e @ (E::Decode(_)
        | E::MissingField(_)
        | E::InvalidAtUri(_)
        | E::InvalidCid(_)
        | E::UriMismatch { .. }) => XrpcError::InvalidRecord(e.to_string()),
        e @ (E::Network(_)
        | E::Build(_)
        | E::Upstream(_)
        | E::BodyTooLarge { .. }
        | E::BadScheme(_)) => XrpcError::UpstreamUnavailable(e.to_string()),
    }
}

fn parse_uri(raw: &str) -> Result<AtUri<DefaultStr>, XrpcError> {
    AtUri::<DefaultStr>::new_owned(raw).map_err(|e| XrpcError::InvalidParams(format!("uri: {e}")))
}

pub enum SubjectShape {
    BareDid,
    Collection(&'static str),
    BareDidOrOneOfCollections(&'static [&'static str]),
    OneOfCollections(&'static [&'static str]),
    AnyAtUri,
    BareDidOrAnyAtUri,
}

pub trait HasSubject {
    const SHAPE: SubjectShape;
}

pub trait MirrorOf {
    type Record: XrpcResp;
    const EDGE_KIND: &'static str;
    const SHAPE: SubjectShape;
}

macro_rules! edge_kinds {
    ($($collection:literal => $record:ty, $shape:expr $(, mirror $by:ident)? ;)*) => {
        $(
            impl HasSubject for $record {
                const SHAPE: SubjectShape = $shape;
            }
        )*
        $($(
            pub struct $by;
            impl MirrorOf for $by {
                type Record = $record;
                const EDGE_KIND: &'static str = concat!($collection, ".by");
                const SHAPE: SubjectShape = SubjectShape::BareDid;
            }
        )?)*

        pub(crate) fn subject_shape(collection: &str) -> Option<(&'static str, SubjectShape)> {
            Some(match collection {
                $($collection => ($collection, <$record as HasSubject>::SHAPE),)*
                _ => return None,
            })
        }

        pub(crate) fn mirror_kind(collection: &str) -> Option<&'static str> {
            Some(match collection {
                $($($collection => <$by as MirrorOf>::EDGE_KIND,)?)*
                _ => return None,
            })
        }
    };
}

edge_kinds! {
    "sh.tangled.feed.star" => StarRecord, SubjectShape::BareDidOrOneOfCollections(&["sh.tangled.string"]), mirror StarBy;
    "sh.tangled.feed.comment" => FeedCommentRecord, SubjectShape::OneOfCollections(&["sh.tangled.repo.issue", "sh.tangled.repo.pull", "sh.tangled.string"]), mirror FeedCommentBy;
    "sh.tangled.feed.reaction" => ReactionRecord, SubjectShape::AnyAtUri, mirror ReactionBy;
    "sh.tangled.graph.follow" => FollowRecord, SubjectShape::BareDid, mirror FollowBy;
    "sh.tangled.graph.vouch" => VouchRecord, SubjectShape::BareDid, mirror VouchBy;
    "sh.tangled.knot" => KnotRecord, SubjectShape::BareDid;
    "sh.tangled.label.definition" => LabelDefinitionRecord, SubjectShape::BareDid;
    "sh.tangled.label.op" => LabelOpRecord, SubjectShape::OneOfCollections(&["sh.tangled.repo.issue", "sh.tangled.repo.pull"]), mirror LabelOpBy;
    "sh.tangled.pipeline" => PipelineRecord, SubjectShape::BareDid, mirror PipelineBy;
    "sh.tangled.pipeline.status" => PipelineStatusRecord, SubjectShape::Collection("sh.tangled.pipeline"), mirror PipelineStatusBy;
    "sh.tangled.publicKey" => PublicKeyRecord, SubjectShape::BareDid;
    "sh.tangled.repo" => RepoRecord, SubjectShape::BareDid;
    "sh.tangled.repo.artifact" => ArtifactRecord, SubjectShape::BareDid, mirror ArtifactBy;
    "sh.tangled.repo.collaborator" => CollaboratorRecord, SubjectShape::BareDid, mirror CollaboratorBy;
    "sh.tangled.repo.issue" => IssueRecord, SubjectShape::BareDid, mirror IssueBy;
    "sh.tangled.repo.issue.state" => IssueStateRecord, SubjectShape::Collection("sh.tangled.repo.issue"), mirror IssueStateBy;
    "sh.tangled.repo.pull" => PullRecord, SubjectShape::BareDid, mirror PullBy;
    "sh.tangled.repo.pull.status" => PullStatusRecord, SubjectShape::Collection("sh.tangled.repo.pull"), mirror PullStatusBy;
    "sh.tangled.spindle" => SpindleRecord, SubjectShape::BareDid;
    "sh.tangled.spindle.member" => SpindleMemberRecord, SubjectShape::BareDid, mirror SpindleMemberBy;
    "sh.tangled.string" => TangledStringRecord, SubjectShape::BareDid;
    "org.tangled.feed.subscription" => SubscriptionRecord, SubjectShape::BareDidOrAnyAtUri;
}

const FORK_SUBJECT_SHAPE: SubjectShape = SubjectShape::BareDid;

fn parse_subject(raw: &SubjectQuery, shape: SubjectShape) -> Result<SubjectRef, XrpcError> {
    let uri = match raw {
        SubjectQuery::Did(did) => {
            return match shape {
                SubjectShape::BareDid
                | SubjectShape::BareDidOrOneOfCollections(_)
                | SubjectShape::BareDidOrAnyAtUri => Ok(SubjectRef::Did(did.clone())),
                SubjectShape::Collection(expected) => Err(XrpcError::InvalidParams(format!(
                    "subject must be at://<did>/{expected}/<rkey>, got bare did"
                ))),
                SubjectShape::OneOfCollections(allowed) => Err(XrpcError::InvalidParams(format!(
                    "subject must be at://<did>/<nsid>/<rkey> with nsid in [{}], got bare did",
                    allowed.join(", "),
                ))),
                SubjectShape::AnyAtUri => Err(XrpcError::InvalidParams(
                    "subject must be at-uri form, got bare did".into(),
                )),
            };
        }
        SubjectQuery::Uri(uri) => uri,
    };
    if matches!(uri.authority(), AtIdentifier::Handle(_)) {
        return Err(XrpcError::InvalidParams(
            "subject authority must be a did, not a handle".into(),
        ));
    }
    let Some(collection) = uri.collection() else {
        return Err(XrpcError::InvalidParams(
            "subject must be a bare did or full at://<did>/<nsid>/<rkey>".into(),
        ));
    };
    let c = collection.as_ref();
    match shape {
        SubjectShape::BareDid => Err(XrpcError::InvalidParams(format!(
            "subject must be a bare did, got at-uri with collection {c}"
        ))),
        SubjectShape::Collection(expected) if c == expected => {
            require_rkey(uri, expected)?;
            Ok(SubjectRef::Uri(uri.clone()))
        }
        SubjectShape::Collection(expected) => Err(XrpcError::InvalidParams(format!(
            "subject must be at://<did>/{expected}/<rkey>, got collection {c}"
        ))),
        SubjectShape::OneOfCollections(allowed) if allowed.contains(&c) => {
            require_rkey(uri, c)?;
            Ok(SubjectRef::Uri(uri.clone()))
        }
        SubjectShape::OneOfCollections(allowed) => Err(XrpcError::InvalidParams(format!(
            "subject must be at://<did>/<nsid>/<rkey> with nsid in [{}], got collection {c}",
            allowed.join(", "),
        ))),
        SubjectShape::BareDidOrOneOfCollections(allowed) if allowed.contains(&c) => {
            require_rkey(uri, c)?;
            Ok(SubjectRef::Uri(uri.clone()))
        }
        SubjectShape::BareDidOrOneOfCollections(allowed) => Err(XrpcError::InvalidParams(format!(
            "subject must be a bare did or at://<did>/<nsid>/<rkey> with nsid in [{}], got collection {c}",
            allowed.join(", "),
        ))),
        SubjectShape::AnyAtUri | SubjectShape::BareDidOrAnyAtUri => {
            Ok(SubjectRef::Uri(uri.clone()))
        }
    }
}

fn require_rkey(uri: &AtUri<DefaultStr>, expected: &str) -> Result<(), XrpcError> {
    uri.rkey().map(|_| ()).ok_or_else(|| {
        XrpcError::InvalidParams(format!(
            "subject must be at://<did>/{expected}/<rkey>; missing rkey"
        ))
    })
}

pub(crate) fn parse_cursor(raw: Option<&str>) -> Result<PageCursor, XrpcError> {
    PageCursor::from_token(raw)
        .map_err(|e: CursorParseError| XrpcError::InvalidParams(format!("cursor: {e}")))
}

// cursor and offset are mutually exclusive, offset pages return a cursor for follow-up
pub(crate) fn parse_start(
    cursor: Option<&str>,
    offset: Option<PageOffset>,
) -> Result<PageStart, XrpcError> {
    match (cursor, offset) {
        (Some(_), Some(_)) => Err(XrpcError::InvalidParams(
            "cursor and offset are mutually exclusive".into(),
        )),
        (Some(c), None) => Ok(PageStart::Cursor(parse_cursor(Some(c))?)),
        (None, Some(o)) => Ok(PageStart::Offset(o)),
        (None, None) => Ok(PageStart::Cursor(PageCursor::Start)),
    }
}

pub(crate) fn parse_limit(raw: Option<u32>) -> Result<PageLimit, XrpcError> {
    PageLimit::new(raw.unwrap_or(DEFAULT_LIMIT))
        .map_err(|e| XrpcError::InvalidParams(format!("limit: {e}")))
}

async fn resolve_for_view(
    state: &AppState,
    expected_nsid: &Nsid<DefaultStr>,
    uri: AtUri<DefaultStr>,
) -> Result<Arc<RecordBody>, XrpcError> {
    let raw = uri.as_ref().to_owned();
    resolve(state, ExpectedNsid::new(expected_nsid.clone()), uri)
        .await
        .map(|(body, _did)| body)
        .map_err(|e| match e {
            XrpcError::NotFound => XrpcError::UpstreamGone(raw),
            other => other,
        })
}

async fn resolve(
    state: &AppState,
    expected: ExpectedNsid,
    uri: AtUri<DefaultStr>,
) -> Result<(Arc<RecordBody>, Did<DefaultStr>), XrpcError> {
    let collection = uri
        .collection()
        .ok_or_else(|| XrpcError::InvalidParams("uri missing collection".into()))?;
    if !expected.accepts(collection.as_ref()) {
        return Err(XrpcError::InvalidParams(format!(
            "collection mismatch: expected {}, got {}",
            expected.as_str(),
            collection.as_ref()
        )));
    }
    let rkey = uri
        .rkey()
        .ok_or_else(|| XrpcError::InvalidParams("uri missing rkey".into()))?;
    let did_ref = match uri.authority() {
        AtIdentifier::Did(d) => d,
        AtIdentifier::Handle(_) => {
            return Err(XrpcError::InvalidParams(
                "uri authority must be a did, not a handle".into(),
            ));
        }
    };
    let did: Did<DefaultStr> = did_ref.clone().into_static();

    if let Some(hit) = state.records.get(&uri) {
        return Ok((hit, did));
    }
    let body = state
        .slingshot
        .get_record(&did_ref, &collection, &rkey)
        .await
        .map_err(map_slingshot)?;
    verify_type_tag(&body, &expected)?;
    state.records.put(uri, body.clone());
    Ok((body, did))
}

#[derive(Deserialize)]
struct TypeTag<'a> {
    #[serde(rename = "$type", borrow)]
    ty: &'a str,
}

fn verify_type_tag(body: &RecordBody, expected: &ExpectedNsid) -> Result<(), XrpcError> {
    let bytes = body.value.as_ref();
    let ty: std::borrow::Cow<'_, str> = match serde_json::from_slice::<TypeTag>(bytes) {
        Ok(t) => std::borrow::Cow::Borrowed(t.ty),
        Err(_) => {
            let value: serde_json::Value = serde_json::from_slice(bytes)
                .map_err(|e| XrpcError::InvalidRecord(format!("$type peek: {e}")))?;
            value
                .as_object()
                .and_then(|m| m.get("$type"))
                .and_then(|v| v.as_str())
                .map(|s| std::borrow::Cow::Owned(s.to_owned()))
                .ok_or_else(|| XrpcError::InvalidRecord("$type peek: missing $type field".into()))?
        }
    };
    if !expected.accepts(ty.as_ref()) {
        return Err(XrpcError::InvalidRecord(format!(
            "$type mismatch: expected {}, got {}",
            expected.as_str(),
            ty
        )));
    }
    Ok(())
}

fn wire_type_nsid(bytes: &[u8]) -> Option<Nsid<DefaultStr>> {
    let ty = serde_json::from_slice::<TypeTag>(bytes).ok()?.ty;
    Nsid::<DefaultStr>::new_owned(ty).ok()
}

async fn deserialize_or_upgrade<V>(
    state: &AppState,
    nsid: &Nsid<DefaultStr>,
    bytes: &[u8],
) -> Result<V, XrpcError>
where
    V: serde::de::DeserializeOwned,
{
    match serde_json::from_slice::<V>(bytes) {
        Ok(v) => Ok(v),
        Err(canon_err) => {
            let normalized = normalize_record_fields(bytes);
            let working: &[u8] = normalized.as_deref().unwrap_or(bytes);
            if normalized.is_some()
                && let Ok(v) = serde_json::from_slice::<V>(working)
            {
                return Ok(v);
            }
            let scrubbed = scrub_record_bytes(nsid, working);
            let retry_bytes: &[u8] = scrubbed.as_deref().unwrap_or(working);
            if scrubbed.is_some()
                && let Ok(v) = serde_json::from_slice::<V>(retry_bytes)
            {
                return Ok(v);
            }
            let wire_nsid = wire_type_nsid(retry_bytes).unwrap_or_else(|| nsid.clone());
            match upgrade_wire_bytes(&wire_nsid, retry_bytes, &state.resolver).await {
                Ok(canon_bytes) => serde_json::from_slice(&canon_bytes)
                    .map_err(|e| XrpcError::InvalidRecord(e.to_string())),
                Err(_) => Err(XrpcError::InvalidRecord(canon_err.to_string())),
            }
        }
    }
}

async fn fetch_from_uri<R, V>(
    state: &AppState,
    uri: AtUri<DefaultStr>,
) -> Result<(Arc<RecordBody>, V), XrpcError>
where
    R: XrpcResp,
    V: serde::de::DeserializeOwned + NormalizeRepoRefs,
{
    let raw = uri.as_str().to_owned();
    let nsid = nsid_static(R::NSID);
    let (body, _did) = resolve(state, ExpectedNsid::new(nsid.clone()), uri).await?;
    let value: V = deserialize_or_upgrade(state, &nsid, &body.value).await?;
    let value = value
        .normalize(&state.resolver)
        .await
        .ok_or(XrpcError::UpstreamGone(raw))?;
    Ok((body, value))
}

pub(crate) async fn fetch<R, V>(
    state: &AppState,
    uri: &AtUri<DefaultStr>,
) -> Result<(Arc<RecordBody>, V), XrpcError>
where
    R: XrpcResp,
    V: serde::de::DeserializeOwned + NormalizeRepoRefs,
{
    fetch_from_uri::<R, V>(state, uri.clone()).await
}

async fn get_repo(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<GetRepoQuery>,
) -> Result<Json<Deduped<RepoGetRecordOutput<DefaultStr>>>, XrpcError> {
    let (body, value) = fetch::<RepoRecord, Repo<DefaultStr>>(&state, &q.repo).await?;
    Ok(Json(Deduped(RepoGetRecordOutput {
        cid: Some(body.cid.clone()),
        uri: body.uri.clone(),
        value,
    })))
}

async fn get_repo_by_repo_did(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<GetRepoByRepoDidQuery>,
) -> Result<Json<Deduped<RepoGetRecordOutput<DefaultStr>>>, XrpcError> {
    let ident = state
        .resolver
        .lookup_by_repo_did(&q.repo_did)
        .await
        .ok_or(XrpcError::NotFound)?;
    let uri = AtUri::<DefaultStr>::from_parts_owned(
        ident.owner.as_str(),
        RepoRecord::NSID,
        ident.rkey.as_str(),
    )
    .expect("Did and Rkey newtypes already validated, at-uri assembly cannot fail");
    let (body, value) = fetch_from_uri::<RepoRecord, Repo<DefaultStr>>(&state, uri).await?;
    Ok(Json(Deduped(RepoGetRecordOutput {
        cid: Some(body.cid.clone()),
        uri: body.uri.clone(),
        value,
    })))
}

async fn get_repo_by_name(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<GetRepoByNameQuery>,
) -> Result<Json<Deduped<RepoGetRecordOutput<DefaultStr>>>, XrpcError> {
    let ident = state
        .resolver
        .lookup_by_name(&q.owner, &q.name)
        .await
        .ok_or(XrpcError::NotFound)?;
    let uri = AtUri::<DefaultStr>::from_parts_owned(
        ident.owner.as_str(),
        RepoRecord::NSID,
        ident.rkey.as_str(),
    )
    .expect("Did and Rkey newtypes already validated, at-uri assembly cannot fail");
    let (body, value) = fetch_from_uri::<RepoRecord, Repo<DefaultStr>>(&state, uri).await?;
    Ok(Json(Deduped(RepoGetRecordOutput {
        cid: Some(body.cid.clone()),
        uri: body.uri.clone(),
        value,
    })))
}

async fn get_profile(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<GetProfileQuery>,
) -> Result<Json<Deduped<ProfileGetRecordOutput<DefaultStr>>>, XrpcError> {
    let (body, value) = fetch::<ProfileRecord, Profile<DefaultStr>>(&state, &q.actor).await?;
    Ok(Json(Deduped(ProfileGetRecordOutput {
        cid: Some(body.cid.clone()),
        uri: body.uri.clone(),
        value,
    })))
}

async fn get_issue(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<GetIssueQuery>,
) -> Result<Json<Deduped<IssueGetRecordOutput<DefaultStr>>>, XrpcError> {
    let (body, value) = fetch::<IssueRecord, Issue<DefaultStr>>(&state, &q.issue).await?;
    Ok(Json(Deduped(IssueGetRecordOutput {
        cid: Some(body.cid.clone()),
        uri: body.uri.clone(),
        value,
    })))
}

async fn get_pull(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<GetPullQuery>,
) -> Result<Json<Deduped<PullGetRecordOutput<DefaultStr>>>, XrpcError> {
    let (body, value) = fetch::<PullRecord, Pull<DefaultStr>>(&state, &q.pull).await?;
    Ok(Json(Deduped(PullGetRecordOutput {
        cid: Some(body.cid.clone()),
        uri: body.uri.clone(),
        value,
    })))
}

async fn get_string(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<GetStringQuery>,
) -> Result<Json<Deduped<TangledStringGetRecordOutput<DefaultStr>>>, XrpcError> {
    let (body, value) =
        fetch::<TangledStringRecord, TangledString<DefaultStr>>(&state, &q.uri).await?;
    Ok(Json(Deduped(TangledStringGetRecordOutput {
        cid: Some(body.cid.clone()),
        uri: body.uri.clone(),
        value,
    })))
}

async fn get_public_key(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<GetPublicKeyQuery>,
) -> Result<Json<Deduped<PublicKeyGetRecordOutput<DefaultStr>>>, XrpcError> {
    let (body, value) =
        fetch::<PublicKeyRecord, PublicKey<DefaultStr>>(&state, &q.public_key).await?;
    Ok(Json(Deduped(PublicKeyGetRecordOutput {
        cid: Some(body.cid.clone()),
        uri: body.uri.clone(),
        value,
    })))
}

async fn get_repos(
    State(state): State<AppState>,
    RawQuery(query): RawQuery,
) -> Result<Response, XrpcError> {
    let uris = collect_repeated(query.as_deref(), BULK_REPOS_KEY);
    bulk_fetch::<RepoRecord, Repo<DefaultStr>>(&state, uris).await
}

async fn get_repos_by_repo_dids(
    State(state): State<AppState>,
    RawQuery(query): RawQuery,
) -> Result<Response, XrpcError> {
    let dids = collect_repeated(query.as_deref(), BULK_REPO_DIDS_KEY);
    if dids.is_empty() {
        return Err(XrpcError::InvalidParams("at least one did required".into()));
    }
    if dids.len() > BULK_LIMIT {
        return Err(XrpcError::InvalidParams(format!(
            "at most {BULK_LIMIT} dids per request"
        )));
    }
    let dids = dids
        .iter()
        .map(|s| {
            Did::<DefaultStr>::new_owned(s)
                .map_err(|_| XrpcError::InvalidParams(format!("invalid did: {s}")))
        })
        .collect::<Result<Vec<_>, _>>()?;
    let mut uris: Vec<AtUri<DefaultStr>> = Vec::new();
    for did in &dids {
        if let Some(ident) = state.resolver.lookup_by_repo_did(did).await {
            uris.push(
                AtUri::<DefaultStr>::from_parts_owned(
                    ident.owner.as_str(),
                    RepoRecord::NSID,
                    ident.rkey.as_str(),
                )
                .expect("Did and Rkey newtypes already validated, at-uri assembly cannot fail"),
            );
        }
    }
    bulk_stream::<RepoRecord, Repo<DefaultStr>>(&state, uris)
}

async fn get_profiles(
    State(state): State<AppState>,
    RawQuery(query): RawQuery,
) -> Result<Response, XrpcError> {
    let uris = collect_repeated(query.as_deref(), BULK_PROFILES_KEY);
    bulk_fetch::<ProfileRecord, Profile<DefaultStr>>(&state, uris).await
}

async fn get_issues(
    State(state): State<AppState>,
    RawQuery(query): RawQuery,
) -> Result<Response, XrpcError> {
    let uris = collect_repeated(query.as_deref(), BULK_ISSUES_KEY);
    bulk_fetch::<IssueRecord, Issue<DefaultStr>>(&state, uris).await
}

async fn get_pulls(
    State(state): State<AppState>,
    RawQuery(query): RawQuery,
) -> Result<Response, XrpcError> {
    let uris = collect_repeated(query.as_deref(), BULK_PULLS_KEY);
    bulk_fetch::<PullRecord, Pull<DefaultStr>>(&state, uris).await
}

const BULK_REPOS_KEY: &str = "repos";
const BULK_REPO_DIDS_KEY: &str = "dids";
const BULK_PROFILES_KEY: &str = "actors";
const BULK_ISSUES_KEY: &str = "issues";
const BULK_PULLS_KEY: &str = "pulls";
const BULK_LIMIT: usize = 50;

fn collect_repeated(query: Option<&str>, key: &str) -> Vec<String> {
    let Some(q) = query else {
        return Vec::new();
    };
    form_urlencoded::parse(q.as_bytes())
        .filter_map(|(k, v)| (k == key).then(|| v.into_owned()))
        .collect()
}

async fn hydrate_record_view<V>(
    state: &AppState,
    nsid: &Nsid<DefaultStr>,
    uri: AtUri<DefaultStr>,
    sort_micros: u64,
) -> Result<Option<RecordView<V>>, XrpcError>
where
    V: serde::de::DeserializeOwned + NormalizeRepoRefs,
{
    if let Some(source) = decode_knot_owned_source(&uri) {
        return synthesize_knot_owned_view::<V>(state, uri, source, sort_micros).await;
    }
    let body = resolve_for_view(state, nsid, uri).await?;
    let value: V = deserialize_or_upgrade::<V>(state, nsid, &body.value).await?;
    let Some(value) = value.normalize(&state.resolver).await else {
        return Ok(None);
    };
    Ok(Some(RecordView {
        uri: body.uri.clone(),
        cid: Some(body.cid.clone()),
        value,
    }))
}

async fn synthesize_knot_owned_view<V>(
    state: &AppState,
    uri: AtUri<DefaultStr>,
    source: KnotOwnedSource,
    sort_micros: u64,
) -> Result<Option<RecordView<V>>, XrpcError>
where
    V: serde::de::DeserializeOwned + NormalizeRepoRefs,
{
    let Some(body) = synth_knot_owned_value(source, sort_micros) else {
        return Ok(None);
    };
    let Ok(value) = serde_json::from_value::<V>(body) else {
        return Ok(None);
    };
    let Some(value) = value.normalize(&state.resolver).await else {
        return Ok(None);
    };
    Ok(Some(RecordView {
        uri,
        cid: None,
        value,
    }))
}

fn synth_knot_owned_value(source: KnotOwnedSource, sort_micros: u64) -> Option<serde_json::Value> {
    let created_at = datetime_of(UnixMicros::new(sort_micros))?;
    match source {
        KnotOwnedSource::Collaborator { repo, subject } => Some(serde_json::json!({
            "repo": repo.as_ref(),
            "subject": subject.as_ref(),
            "createdAt": created_at,
        })),
    }
}

fn datetime_of(at: UnixMicros) -> Option<Datetime> {
    let micros = i64::try_from(at.raw()).ok()?;
    chrono::DateTime::from_timestamp_micros(micros).map(|dt| Datetime::new(dt.fixed_offset()))
}

#[derive(Clone, Copy)]
enum HitProvenance {
    ClientSupplied,
    Indexed,
}

fn is_index_evictable(err: &XrpcError) -> bool {
    matches!(
        err,
        XrpcError::NotFound
            | XrpcError::UpstreamGone(_)
            | XrpcError::InvalidRecord(_)
            | XrpcError::InvalidParams(_)
    )
}

fn drop_unhydratable<V>(
    provenance: HitProvenance,
    nsid: &Nsid<DefaultStr>,
    uri: &AtUri<DefaultStr>,
    result: Result<Option<V>, XrpcError>,
) -> Result<Option<V>, XrpcError> {
    match result {
        Ok(view) => Ok(view),
        Err(err @ (XrpcError::NotFound | XrpcError::UpstreamGone(_))) => {
            tracing::debug!(
                uri = %uri,
                nsid = %nsid.as_ref(),
                error = %err,
                "dropping gone hit during hydration",
            );
            Ok(None)
        }
        Err(err @ XrpcError::UpstreamUnavailable(_)) => {
            tracing::warn!(
                uri = %uri,
                nsid = %nsid.as_ref(),
                error = %err,
                "dropping hit, upstream unavailable during hydration",
            );
            Ok(None)
        }
        Err(err @ XrpcError::InvalidRecord(_)) => {
            tracing::warn!(
                uri = %uri,
                nsid = %nsid.as_ref(),
                error = %err,
                "dropping invalid hit during hydration",
            );
            Ok(None)
        }
        Err(err @ XrpcError::InvalidParams(_)) => match provenance {
            HitProvenance::ClientSupplied => Err(err),
            HitProvenance::Indexed => {
                tracing::warn!(
                    uri = %uri,
                    nsid = %nsid.as_ref(),
                    error = %err,
                    "dropping malformed indexed hit during hydration",
                );
                Ok(None)
            }
        },
        Err(
            err @ (XrpcError::Internal(_)
            | XrpcError::Overloaded
            | XrpcError::Warming
            | XrpcError::NotSettled
            | XrpcError::AuthRequired(_)
            | XrpcError::NotImplemented(_)),
        ) => Err(err),
    }
}

fn hydrate_stream<T, Fut, V>(
    items: impl IntoIterator<Item = T>,
    produce: impl FnMut(T) -> Fut,
) -> impl Stream<Item = Result<V, XrpcError>>
where
    Fut: Future<Output = Result<Option<V>, XrpcError>>,
{
    stream::iter(items)
        .map(produce)
        .buffered(FETCH_CONCURRENCY)
        .try_filter_map(|view| async move { Ok(view) })
}

fn hydrate_keyed_record_stream<K, V>(
    state: &AppState,
    nsid: Nsid<DefaultStr>,
    items: Vec<(K, EdgeItem)>,
    provenance: HitProvenance,
) -> impl Stream<Item = Result<(K, RecordView<V>), XrpcError>> + Send + 'static
where
    K: Send + 'static,
    V: serde::de::DeserializeOwned + Serialize + NormalizeRepoRefs + Send + 'static,
{
    let owned = state.clone();
    hydrate_stream(items, move |(key, item)| {
        let owned = owned.clone();
        let nsid = nsid.clone();
        async move {
            let EdgeItem {
                uri, sort_micros, ..
            } = item;
            let result = hydrate_record_view::<V>(&owned, &nsid, uri.clone(), sort_micros).await;
            if matches!(provenance, HitProvenance::Indexed)
                && let Err(err) = &result
                && is_index_evictable(err)
            {
                owned.edges.remove_source(&uri);
            }
            drop_unhydratable(provenance, &nsid, &uri, result)
                .map(|view| view.map(|view| (key, view)))
        }
    })
}

fn hydrate_record_stream<V>(
    state: &AppState,
    nsid: Nsid<DefaultStr>,
    items: Vec<EdgeItem>,
    provenance: HitProvenance,
) -> impl Stream<Item = Result<RecordView<V>, XrpcError>> + Send + 'static
where
    V: serde::de::DeserializeOwned + Serialize + NormalizeRepoRefs + Send + 'static,
{
    hydrate_keyed_record_stream(
        state,
        nsid,
        items.into_iter().map(|item| ((), item)).collect(),
        provenance,
    )
    .map_ok(|(_, view)| view)
}

enum PagePhase {
    Head,
    Body { first: bool },
    Done,
}

struct PageState<S> {
    items: std::pin::Pin<Box<S>>,
    phase: PagePhase,
    array_key: &'static str,
    tail: Vec<u8>,
    permit: Option<HeavyPermit>,
}

pub(crate) fn paged_tail(cursor: Option<String>, total: Option<TotalCount>) -> Vec<u8> {
    let encoded =
        cursor.map(|value| serde_json::to_string(&value).unwrap_or_else(|_| "\"\"".to_owned()));
    match (encoded, total) {
        (Some(cursor), Some(total)) => {
            format!("],\"cursor\":{cursor},\"total\":{}}}", total.get()).into_bytes()
        }
        (Some(cursor), None) => format!("],\"cursor\":{cursor}}}").into_bytes(),
        (None, Some(total)) => format!("],\"total\":{}}}", total.get()).into_bytes(),
        (None, None) => unpaged_tail(),
    }
}

fn unpaged_tail() -> Vec<u8> {
    b"]}".to_vec()
}

pub(crate) fn json_stream<V, S>(
    array_key: &'static str,
    items: S,
    tail: Vec<u8>,
    permit: Option<HeavyPermit>,
) -> Response
where
    V: Serialize + Send + 'static,
    S: Stream<Item = Result<V, XrpcError>> + Send + 'static,
{
    let init = PageState {
        items: Box::pin(items),
        phase: PagePhase::Head,
        array_key,
        tail,
        permit,
    };
    let chunks = stream::unfold(init, |mut st| async move {
        match st.phase {
            PagePhase::Head => {
                let head = format!("{{\"{}\":[", st.array_key).into_bytes();
                st.phase = PagePhase::Body { first: true };
                Some((Ok::<Vec<u8>, Infallible>(head), st))
            }
            PagePhase::Body { first } => match st.items.next().await {
                Some(Ok(view)) => match serde_json::to_vec(&Deduped(&view)) {
                    Ok(encoded) => {
                        let mut chunk = Vec::with_capacity(encoded.len() + 1);
                        if !first {
                            chunk.push(b',');
                        }
                        chunk.extend_from_slice(&encoded);
                        st.phase = PagePhase::Body { first: false };
                        Some((Ok(chunk), st))
                    }
                    Err(e) => {
                        tracing::warn!(error = %e, "skipping hit, serialize failed mid-stream");
                        Some((Ok(Vec::new()), st))
                    }
                },
                Some(Err(e)) => {
                    tracing::warn!(error = %e, "ending page early, hydration failed mid-stream");
                    let tail = std::mem::take(&mut st.tail);
                    st.phase = PagePhase::Done;
                    Some((Ok(tail), st))
                }
                None => {
                    let tail = std::mem::take(&mut st.tail);
                    st.phase = PagePhase::Done;
                    Some((Ok(tail), st))
                }
            },
            PagePhase::Done => {
                drop(st.permit.take());
                None
            }
        }
    });
    (
        [(CONTENT_TYPE, "application/json")],
        Body::from_stream(chunks),
    )
        .into_response()
}

async fn bulk_fetch<R, V>(state: &AppState, uris: Vec<String>) -> Result<Response, XrpcError>
where
    R: XrpcResp,
    V: serde::de::DeserializeOwned + Serialize + NormalizeRepoRefs + Send + 'static,
{
    if uris.is_empty() {
        return Err(XrpcError::InvalidParams("at least one uri required".into()));
    }
    if uris.len() > BULK_LIMIT {
        return Err(XrpcError::InvalidParams(format!(
            "at most {BULK_LIMIT} uris per request"
        )));
    }
    let parsed: Vec<AtUri<DefaultStr>> = uris
        .iter()
        .map(|s| parse_uri(s))
        .collect::<Result<_, _>>()?;
    let nsid = nsid_static(R::NSID);
    if let Some(bad) = parsed
        .iter()
        .find(|uri| uri.collection().is_none_or(|c| c.as_ref() != nsid.as_ref()))
    {
        return Err(XrpcError::InvalidParams(format!(
            "uri collection must be {}, got {}",
            nsid.as_ref(),
            bad.as_ref()
        )));
    }
    bulk_stream::<R, V>(state, parsed)
}

fn bulk_stream<R, V>(
    state: &AppState,
    parsed: Vec<AtUri<DefaultStr>>,
) -> Result<Response, XrpcError>
where
    R: XrpcResp,
    V: serde::de::DeserializeOwned + Serialize + NormalizeRepoRefs + Send + 'static,
{
    let nsid = nsid_static(R::NSID);
    let permit = state.heavy_permit()?;
    let items = parsed
        .into_iter()
        .map(|uri| EdgeItem {
            uri,
            sort_micros: 0,
            sort_source: 0,
        })
        .collect();
    let views = hydrate_record_stream::<V>(state, nsid, items, HitProvenance::ClientSupplied);
    Ok(json_stream::<RecordView<V>, _>(
        "items",
        views,
        unpaged_tail(),
        permit,
    ))
}

fn record_edge_page<R, F>(
    state: &AppState,
    q: &TypedListQuery<F>,
) -> Result<(EdgePage, Nsid<DefaultStr>), XrpcError>
where
    R: XrpcResp + HasSubject,
    F: ListFilter,
{
    let subject = parse_subject(&q.subject, R::SHAPE)?;
    let start = parse_start(q.cursor.as_deref(), q.offset)?;
    let limit = parse_limit(q.limit)?;
    let dir = q.dir();
    let nsid = nsid_static(R::NSID);
    let page = match (q.filter.is_identity(), q.filter.state_view()) {
        (true, _) => {
            let key = EdgeKey::new(nsid.clone(), subject);
            state.edges.list(&key, start, limit, dir)
        }
        (_, Some(StateViewQuery::Issue { kind, author })) => {
            let key = EdgeKey::new(nsid.clone(), subject);
            state
                .edges
                .page_issue_state(&key, kind, start, limit, dir, author.as_ref())
        }
        (_, Some(StateViewQuery::Pull { kind, author })) => {
            let key = EdgeKey::new(nsid.clone(), subject);
            state
                .edges
                .page_pull_status(&key, kind, start, limit, dir, author.as_ref())
        }
        (_, None) => {
            return Err(XrpcError::Internal(
                "non-identity filter did not resolve to a state view".into(),
            ));
        }
    };
    Ok((page, nsid))
}

fn record_list_response<V>(
    state: &AppState,
    page: EdgePage,
    nsid: Nsid<DefaultStr>,
) -> Result<Response, XrpcError>
where
    V: serde::de::DeserializeOwned + Serialize + NormalizeRepoRefs + Send + 'static,
{
    let permit = state.heavy_permit()?;
    let views = hydrate_record_stream::<V>(state, nsid, page.items, HitProvenance::Indexed);
    Ok(json_stream::<RecordView<V>, _>(
        "items",
        views,
        paged_tail(page.next.map(PageToken::encode_token), page.total),
        permit,
    ))
}

fn collaborator_list_item(
    view: RecordView<Collaborator<DefaultStr>>,
) -> Result<Option<CollaboratorListItem>, XrpcError> {
    let RecordView { uri, cid, value } = view;
    let Some(added_by) = source_authority_did(&uri) else {
        return Ok(None);
    };
    Ok(Some(CollaboratorListItem {
        subject: value.subject,
        added_by,
        created_at: value.created_at.clone(),
        effective_since: value.created_at,
        uri: Some(uri),
        cid,
    }))
}

fn collaborator_list_response(
    state: &AppState,
    page: EdgePage,
    nsid: Nsid<DefaultStr>,
) -> Result<Response, XrpcError> {
    let permit = state.heavy_permit()?;
    let views = hydrate_record_stream::<Collaborator<DefaultStr>>(
        state,
        nsid,
        page.items,
        HitProvenance::Indexed,
    )
    .try_filter_map(|view| async move { collaborator_list_item(view) });
    Ok(json_stream::<CollaboratorListItem, _>(
        "items",
        views,
        paged_tail(page.next.map(PageToken::encode_token), page.total),
        permit,
    ))
}

async fn list_records<R, V, F>(
    state: &AppState,
    q: TypedListQuery<F>,
) -> Result<Response, XrpcError>
where
    R: XrpcResp + HasSubject,
    V: serde::de::DeserializeOwned + Serialize + NormalizeRepoRefs + Send + 'static,
    F: ListFilter,
{
    let (page, nsid) = record_edge_page::<R, F>(state, &q)?;
    record_list_response::<V>(state, page, nsid)
}

async fn list_records_by_distinct_author<R, V>(
    state: &AppState,
    q: TypedListQuery<NoFilter>,
) -> Result<Response, XrpcError>
where
    R: XrpcResp + HasSubject,
    V: serde::de::DeserializeOwned + Serialize + NormalizeRepoRefs + Send + 'static,
{
    let subject = parse_subject(&q.subject, R::SHAPE)?;
    let start = parse_start(q.cursor.as_deref(), q.offset)?;
    let limit = parse_limit(q.limit)?;
    let nsid = nsid_static(R::NSID);
    let key = EdgeKey::new(nsid.clone(), subject);
    let page = state
        .edges
        .list_distinct_authors(&key, start, limit, q.dir());
    record_list_response::<V>(state, page, nsid)
}

fn count_for<R: XrpcResp + HasSubject>(
    state: &AppState,
    q: CountQuery,
) -> Result<CountResponse, XrpcError> {
    let subject = parse_subject(&q.subject, R::SHAPE)?;
    let key = EdgeKey::new(nsid_static(R::NSID), subject);
    Ok(CountResponse {
        count: state.edges.count(&key),
        distinct_authors: state.edges.count_distinct_authors(&key),
    })
}

fn count_typed_for<R, F>(
    state: &AppState,
    q: TypedCountQuery<F>,
) -> Result<CountResponse, XrpcError>
where
    R: XrpcResp + HasSubject,
    F: CountFilter,
{
    let subject = parse_subject(&q.subject, R::SHAPE)?;
    let key = EdgeKey::new(nsid_static(R::NSID), subject);
    let counted = q.filter.count(state, &key);
    Ok(CountResponse {
        count: counted.count.get(),
        distinct_authors: counted.distinct_authors.get(),
    })
}

// does `actor` have an edge of this kind pointing at `subject`, returns its own uri
fn get_for<R: XrpcResp + HasSubject>(
    state: &AppState,
    q: GetEdgeQuery,
) -> Result<EdgeUriResponse, XrpcError> {
    let subject = parse_subject(&q.subject, R::SHAPE)?;
    let key = EdgeKey::new(nsid_static(R::NSID), subject);
    let uri = state
        .edges
        .viewer_source(&key, q.actor.as_str())
        .ok_or(XrpcError::NotFound)?;
    Ok(EdgeUriResponse { uri })
}

fn mirror_edge_page<M, F>(
    state: &AppState,
    q: &TypedListQuery<F>,
) -> Result<(EdgePage, Nsid<DefaultStr>), XrpcError>
where
    M: MirrorOf,
    F: ListFilter,
{
    let subject = parse_subject(&q.subject, M::SHAPE)?;
    let start = parse_start(q.cursor.as_deref(), q.offset)?;
    let limit = parse_limit(q.limit)?;
    let dir = q.dir();
    let edge_nsid = nsid_static(M::EDGE_KIND);
    let page = match (q.filter.is_identity(), q.filter.state_view()) {
        (true, _) => {
            let key = EdgeKey::new(edge_nsid, subject);
            state.edges.list(&key, start, limit, dir)
        }
        (_, Some(StateViewQuery::Issue { kind, author })) => {
            let key = EdgeKey::new(edge_nsid, subject);
            state
                .edges
                .page_issue_state(&key, kind, start, limit, dir, author.as_ref())
        }
        (_, Some(StateViewQuery::Pull { kind, author })) => {
            let key = EdgeKey::new(edge_nsid, subject);
            state
                .edges
                .page_pull_status(&key, kind, start, limit, dir, author.as_ref())
        }
        (_, None) => {
            return Err(XrpcError::Internal(
                "non-identity filter did not resolve to a state view".into(),
            ));
        }
    };
    let record_nsid = nsid_static(<M::Record as XrpcResp>::NSID);
    Ok((page, record_nsid))
}

async fn list_mirror<M, V, F>(state: &AppState, q: TypedListQuery<F>) -> Result<Response, XrpcError>
where
    M: MirrorOf,
    V: serde::de::DeserializeOwned + Serialize + NormalizeRepoRefs + Send + 'static,
    F: ListFilter,
{
    let (page, record_nsid) = mirror_edge_page::<M, F>(state, &q)?;
    let permit = state.heavy_permit()?;
    let views = hydrate_record_stream::<V>(state, record_nsid, page.items, HitProvenance::Indexed);
    Ok(json_stream::<RecordView<V>, _>(
        "items",
        views,
        paged_tail(page.next.map(PageToken::encode_token), page.total),
        permit,
    ))
}

fn count_mirror<M: MirrorOf>(state: &AppState, q: CountQuery) -> Result<CountResponse, XrpcError> {
    let subject = parse_subject(&q.subject, M::SHAPE)?;
    let key = EdgeKey::new(nsid_static(M::EDGE_KIND), subject);
    Ok(CountResponse {
        count: state.edges.count(&key),
        distinct_authors: state.edges.count_distinct_authors(&key),
    })
}

async fn list_stars(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
) -> Result<Response, XrpcError> {
    list_records_by_distinct_author::<StarRecord, Star<DefaultStr>>(&state, q).await
}

async fn count_stars(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<CountQuery>,
) -> Result<Json<CountResponse>, XrpcError> {
    count_for::<StarRecord>(&state, q).map(Json)
}

async fn get_star(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<GetEdgeQuery>,
) -> Result<Json<EdgeUriResponse>, XrpcError> {
    get_for::<StarRecord>(&state, q).map(Json)
}
async fn get_subscription_for_actor(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<GetEdgeQuery>,
) -> Result<Json<EdgeUriResponse>, XrpcError> {
    get_for::<SubscriptionRecord>(&state, q).map(Json)
}

async fn list_follows(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
) -> Result<Response, XrpcError> {
    list_records_by_distinct_author::<FollowRecord, Follow<DefaultStr>>(&state, q).await
}

async fn count_follows(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<CountQuery>,
) -> Result<Json<CountResponse>, XrpcError> {
    count_for::<FollowRecord>(&state, q).map(Json)
}

async fn get_follow(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<GetEdgeQuery>,
) -> Result<Json<EdgeUriResponse>, XrpcError> {
    get_for::<FollowRecord>(&state, q).map(Json)
}

async fn list_issues(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<TypedListQuery<IssueFilter>>,
) -> Result<Response, XrpcError> {
    let (page, nsid) = record_edge_page::<IssueRecord, _>(&state, &q)?;
    let permit = state.heavy_permit()?;
    let owned = state.clone();
    let items = hydrate_record_stream::<Issue<DefaultStr>>(
        &state,
        nsid,
        page.items,
        HitProvenance::Indexed,
    )
    .map(move |view| view.map(|v| enrich_issue_view(&owned, v)));
    Ok(json_stream::<StatefulItem<Issue<DefaultStr>>, _>(
        "items",
        items,
        paged_tail(page.next.map(PageToken::encode_token), page.total),
        permit,
    ))
}

async fn count_issues(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<TypedCountQuery<IssueFilter>>,
) -> Result<Json<CountResponse>, XrpcError> {
    count_typed_for::<IssueRecord, _>(&state, q).map(Json)
}

async fn list_pulls(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<TypedListQuery<PullFilter>>,
) -> Result<Response, XrpcError> {
    let (page, nsid) = record_edge_page::<PullRecord, _>(&state, &q)?;
    let permit = state.heavy_permit()?;
    let owned = state.clone();
    let items =
        hydrate_record_stream::<Pull<DefaultStr>>(&state, nsid, page.items, HitProvenance::Indexed)
            .map(move |view| view.map(|v| enrich_pull_view(&owned, v)));
    Ok(json_stream::<StatefulItem<Pull<DefaultStr>>, _>(
        "items",
        items,
        paged_tail(page.next.map(PageToken::encode_token), page.total),
        permit,
    ))
}

async fn count_pulls(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<TypedCountQuery<PullFilter>>,
) -> Result<Json<CountResponse>, XrpcError> {
    count_typed_for::<PullRecord, _>(&state, q).map(Json)
}

#[allow(dead_code)]
async fn list_feed_comments(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
) -> Result<Response, XrpcError> {
    list_records::<FeedCommentRecord, FeedComment<DefaultStr>, _>(&state, q).await
}

async fn count_feed_comments(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<CountQuery>,
) -> Result<Json<CountResponse>, XrpcError> {
    count_for::<FeedCommentRecord>(&state, q).map(Json)
}

async fn list_reactions(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
) -> Result<Response, XrpcError> {
    list_records::<ReactionRecord, Reaction<DefaultStr>, _>(&state, q).await
}

async fn count_reactions(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<CountQuery>,
) -> Result<Json<CountResponse>, XrpcError> {
    count_for::<ReactionRecord>(&state, q).map(Json)
}

async fn list_collaborators(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
) -> Result<Response, XrpcError> {
    let (page, nsid) = record_edge_page::<CollaboratorRecord, NoFilter>(&state, &q)?;
    collaborator_list_response(&state, page, nsid)
}

async fn count_collaborators(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<CountQuery>,
) -> Result<Json<CountResponse>, XrpcError> {
    count_for::<CollaboratorRecord>(&state, q).map(Json)
}

async fn list_issue_states(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
) -> Result<Response, XrpcError> {
    list_records::<IssueStateRecord, IssueState<DefaultStr>, _>(&state, q).await
}

async fn count_issue_states(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<CountQuery>,
) -> Result<Json<CountResponse>, XrpcError> {
    count_for::<IssueStateRecord>(&state, q).map(Json)
}

async fn list_pull_statuses(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
) -> Result<Response, XrpcError> {
    list_records::<PullStatusRecord, PullStatus<DefaultStr>, _>(&state, q).await
}

async fn count_pull_statuses(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<CountQuery>,
) -> Result<Json<CountResponse>, XrpcError> {
    count_for::<PullStatusRecord>(&state, q).map(Json)
}

async fn list_repos(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
) -> Result<Response, XrpcError> {
    list_records::<RepoRecord, Repo<DefaultStr>, _>(&state, q).await
}

async fn count_repos(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<CountQuery>,
) -> Result<Json<CountResponse>, XrpcError> {
    count_for::<RepoRecord>(&state, q).map(Json)
}

// forks are repo records pointing back at a repo, so they get their own edge
// kind instead of a collection
async fn count_forks(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<CountQuery>,
) -> Result<Json<CountResponse>, XrpcError> {
    let subject = parse_subject(&q.subject, FORK_SUBJECT_SHAPE)?;
    let key = EdgeKey::new(nsid_static(REPO_SOURCE_EDGE_KIND), subject);
    Ok(Json(CountResponse {
        count: state.edges.count(&key),
        distinct_authors: state.edges.count_distinct_authors(&key),
    }))
}

async fn list_knots(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
) -> Result<Response, XrpcError> {
    list_records::<KnotRecord, Knot<DefaultStr>, _>(&state, q).await
}

async fn count_knots(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<CountQuery>,
) -> Result<Json<CountResponse>, XrpcError> {
    count_for::<KnotRecord>(&state, q).map(Json)
}

async fn list_spindles(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
) -> Result<Response, XrpcError> {
    list_records::<SpindleRecord, Spindle<DefaultStr>, _>(&state, q).await
}

async fn count_spindles(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<CountQuery>,
) -> Result<Json<CountResponse>, XrpcError> {
    count_for::<SpindleRecord>(&state, q).map(Json)
}

async fn list_public_keys(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
) -> Result<Response, XrpcError> {
    list_records::<PublicKeyRecord, PublicKey<DefaultStr>, _>(&state, q).await
}

async fn count_public_keys(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<CountQuery>,
) -> Result<Json<CountResponse>, XrpcError> {
    count_for::<PublicKeyRecord>(&state, q).map(Json)
}

struct NetworkVouchQuery {
    actors: Vec<Did<DefaultStr>>,
    start: PageCursor,
    limit: PageLimit,
}

fn one_query_value(query: Option<&str>, key: &str) -> Result<Option<String>, XrpcError> {
    let values = collect_repeated(query, key);
    match values.as_slice() {
        [] => Ok(None),
        [value] => Ok(Some(value.clone())),
        _ => Err(XrpcError::InvalidParams(format!(
            "{key} may only be specified once"
        ))),
    }
}

fn parse_network_vouch_query(query: Option<&str>) -> Result<NetworkVouchQuery, XrpcError> {
    let actors = collect_repeated(query, "actors");
    if actors.is_empty() {
        return Err(XrpcError::InvalidParams(
            "at least one actor required".into(),
        ));
    }
    if actors.len() > BULK_LIMIT {
        return Err(XrpcError::InvalidParams(format!(
            "at most {BULK_LIMIT} actors per request"
        )));
    }
    let actors = actors
        .iter()
        .map(|actor| {
            Did::<DefaultStr>::new_owned(actor)
                .map_err(|_| XrpcError::InvalidParams(format!("invalid did: {actor}")))
        })
        .collect::<Result<Vec<_>, _>>()?;
    let mut seen = std::collections::HashSet::new();
    let actors = actors
        .into_iter()
        .filter(|actor| seen.insert(actor.as_ref().to_owned()))
        .collect::<Vec<_>>();
    let cursor = one_query_value(query, "cursor")?;
    if actors.len() > 1 && cursor.is_some() {
        return Err(XrpcError::InvalidParams(
            "cursor requires exactly one actor".into(),
        ));
    }
    let limit = one_query_value(query, "limit")?
        .map(|limit| {
            limit
                .parse::<u32>()
                .map_err(|_| XrpcError::InvalidParams(format!("invalid limit: {limit}")))
        })
        .transpose()?;
    Ok(NetworkVouchQuery {
        actors,
        start: parse_vouch_cursor(cursor.as_deref())?,
        limit: parse_limit(limit)?,
    })
}

const VOUCH_KIND_VOUCH: &str = "vouch";

fn network_of(actor: &Did<DefaultStr>, vouched: Vec<String>) -> std::collections::HashSet<String> {
    vouched
        .into_iter()
        .chain(std::iter::once(actor.as_ref().to_owned()))
        .collect()
}

async fn vouch_network(
    state: &AppState,
    actor: &Did<DefaultStr>,
) -> std::collections::HashSet<String> {
    let nsid = nsid_static("sh.tangled.graph.vouch");
    let sources = state.edges.sources_for(&EdgeKey::new(
        nsid_static("sh.tangled.graph.vouch.by"),
        SubjectRef::Did(actor.clone()),
    ));
    let vouched = stream::iter(sources)
        .map(|uri| {
            let nsid = nsid.clone();
            async move {
                let subject = uri.rkey().map(|rkey| rkey.as_ref().to_owned())?;
                let view = hydrate_record_view::<Vouch<DefaultStr>>(state, &nsid, uri, 0)
                    .await
                    .ok()??;
                (AsRef::<str>::as_ref(&view.value.kind) == VOUCH_KIND_VOUCH).then_some(subject)
            }
        })
        .buffer_unordered(FETCH_CONCURRENCY)
        .filter_map(|subject| async move { subject })
        .collect()
        .await;
    network_of(actor, vouched)
}

#[derive(Clone, Copy)]
enum VouchList {
    Received,
    Sent,
}

fn vouch_page(edges: &EdgeStore, key: &EdgeKey, list: VouchList, cursor: PageCursor) -> EdgePage {
    let limit = PageLimit::new(PageLimit::MAX).unwrap();
    match list {
        VouchList::Received => edges.list_distinct_authors(key, cursor, limit, SortDir::Desc),
        VouchList::Sent => edges.list(key, cursor, limit, SortDir::Desc),
    }
}

fn vouch_items(
    edges: &EdgeStore,
    key: EdgeKey,
    list: VouchList,
    start: PageCursor,
) -> impl Iterator<Item = EdgeItem> + '_ {
    vouch_items_from(start, move |cursor| vouch_page(edges, &key, list, cursor))
}

fn vouch_items_from(
    start: PageCursor,
    mut fetch: impl FnMut(PageCursor) -> EdgePage,
) -> impl Iterator<Item = EdgeItem> {
    let mut next = Some(start);
    let mut items = Vec::<EdgeItem>::new().into_iter();
    std::iter::from_fn(move || {
        loop {
            if let Some(item) = items.next() {
                return Some(item);
            }
            let cursor = next.take()?;
            let page = fetch(cursor);
            next = page.next.map(PageCursor::After);
            items = page.items.into_iter();
        }
    })
}

fn merge_vouches(
    received: impl Iterator<Item = EdgeItem>,
    sent: impl Iterator<Item = EdgeItem>,
) -> impl Iterator<Item = EdgeItem> {
    let mut received = received.peekable();
    let mut sent = sent.peekable();
    std::iter::from_fn(move || match (received.peek(), sent.peek()) {
        (Some(left), Some(right)) => match left.sort_token().cmp(&right.sort_token()) {
            std::cmp::Ordering::Greater => received.next(),
            std::cmp::Ordering::Less => sent.next(),
            std::cmp::Ordering::Equal => {
                sent.next();
                received.next()
            }
        },
        (Some(_), None) => received.next(),
        (None, Some(_)) => sent.next(),
        (None, None) => None,
    })
}

fn parse_vouch_cursor(raw: Option<&str>) -> Result<PageCursor, XrpcError> {
    let Some(raw) = raw else {
        return Ok(PageCursor::Start);
    };
    PageToken::decode_token(raw)
        .or_else(|_| {
            raw.parse::<u64>()
                .map(|micros| PageToken::new(micros, 0))
                .map_err(|_| CursorParseError::Malformed)
        })
        .map(PageCursor::After)
        .map_err(|error| XrpcError::InvalidParams(format!("cursor: {error}")))
}

fn paginate_vouches(
    candidates: impl Iterator<Item = EdgeItem>,
    limit: PageLimit,
) -> (Vec<EdgeItem>, Option<String>) {
    let limit = limit.get() as usize;
    let mut seen = std::collections::HashSet::new();
    let mut items = candidates
        .filter(|item| seen.insert(item.uri.clone()))
        .take(limit + 1)
        .collect::<Vec<_>>();
    let has_more = items.len() > limit;
    items.truncate(limit);
    let next = has_more.then(|| items.last().unwrap().sort_token().encode_token());
    (items, next)
}

fn network_vouch_page(
    edges: &EdgeStore,
    actor: &Did<DefaultStr>,
    network: &VouchNetworkFilter,
    start: PageCursor,
    limit: PageLimit,
) -> (Vec<EdgeItem>, Option<String>) {
    let received = vouch_items(
        edges,
        EdgeKey::new(
            nsid_static("sh.tangled.graph.vouch"),
            SubjectRef::Did(actor.clone()),
        ),
        VouchList::Received,
        start,
    )
    .filter(|item| network.matches_author(item));
    let sent = vouch_items(
        edges,
        EdgeKey::new(
            nsid_static("sh.tangled.graph.vouch.by"),
            SubjectRef::Did(actor.clone()),
        ),
        VouchList::Sent,
        start,
    )
    .filter(|item| network.matches_subject(item));
    paginate_vouches(merge_vouches(received, sent), limit)
}

fn network_vouch_pages(
    edges: &EdgeStore,
    actors: &[Did<DefaultStr>],
    network: &VouchNetworkFilter,
    start: PageCursor,
    limit: PageLimit,
) -> (Vec<(Did<DefaultStr>, EdgeItem)>, Option<String>) {
    let single = actors.len() == 1;
    let mut next_cursor = None;
    let mut tagged = Vec::new();
    for actor in actors {
        let (items, cursor) = network_vouch_page(edges, actor, network, start, limit);
        tagged.extend(items.into_iter().map(|item| (actor.clone(), item)));
        if single {
            next_cursor = cursor;
        }
    }
    (tagged, next_cursor)
}

#[cfg(test)]
mod vouch_pagination_tests {
    use std::collections::HashSet;

    use super::*;
    use bobbin_runtime::RuntimeHasher;
    use bobbin_types::edges::Edge;

    fn did(value: &str) -> Did<DefaultStr> {
        Did::new_owned(value).unwrap()
    }

    fn uri(value: String) -> AtUri<DefaultStr> {
        AtUri::new_owned(value).unwrap()
    }

    fn edge(kind: &'static str, subject: &Did<DefaultStr>, author: usize, micros: u64) -> Edge {
        Edge {
            kind: nsid_static(kind),
            subject: SubjectRef::Did(subject.clone()),
            source: uri(format!(
                "at://did:plc:author{author}/sh.tangled.graph.vouch/r{author}"
            )),
            sort_micros: micros,
        }
    }

    #[test]
    fn received_filter_reaches_beyond_the_first_candidate_page() {
        let store = EdgeStore::new(RuntimeHasher::from_seeds(1, 2, 3, 4));
        let subject = did("did:plc:subject");
        (0..=PageLimit::MAX as usize).for_each(|author| {
            store.add(edge(
                "sh.tangled.graph.vouch",
                &subject,
                author,
                author as u64,
            ));
        });
        let key = EdgeKey::new(
            nsid_static("sh.tangled.graph.vouch"),
            SubjectRef::Did(subject),
        );
        let candidates = vouch_items(&store, key, VouchList::Received, PageCursor::Start)
            .filter(|item| item.uri.as_ref().starts_with("at://did:plc:author0/"));
        let (items, cursor) = paginate_vouches(candidates, PageLimit::new(1).unwrap());

        assert_eq!(items.len(), 1);
        assert!(items[0].uri.as_ref().starts_with("at://did:plc:author0/"));
        assert!(cursor.is_none());
    }

    #[test]
    fn cursor_walk_keeps_every_timestamp_tie() {
        let store = EdgeStore::new(RuntimeHasher::from_seeds(1, 2, 3, 4));
        let subject = did("did:plc:subject");
        store.add(edge("sh.tangled.graph.vouch", &subject, 1, 42));
        store.add(edge("sh.tangled.graph.vouch.by", &subject, 2, 42));
        store.add(edge("sh.tangled.graph.vouch", &subject, 3, 42));
        let received = EdgeKey::new(
            nsid_static("sh.tangled.graph.vouch"),
            SubjectRef::Did(subject.clone()),
        );
        let sent = EdgeKey::new(
            nsid_static("sh.tangled.graph.vouch.by"),
            SubjectRef::Did(subject),
        );
        let mut start = PageCursor::Start;
        let mut uris = Vec::new();

        loop {
            let candidates = merge_vouches(
                vouch_items(&store, received.clone(), VouchList::Received, start),
                vouch_items(&store, sent.clone(), VouchList::Sent, start),
            );
            let (items, cursor) = paginate_vouches(candidates, PageLimit::new(1).unwrap());
            uris.extend(items.into_iter().map(|item| item.uri));
            let Some(cursor) = cursor else {
                break;
            };
            assert_eq!(cursor.len(), 24);
            start = parse_vouch_cursor(Some(&cursor)).unwrap();
        }

        assert_eq!(uris.len(), 3);
        assert_eq!(uris.iter().collect::<HashSet<_>>().len(), 3);
    }

    #[test]
    fn page_fetch_is_deferred_until_items_are_needed() {
        let calls = std::cell::Cell::new(0usize);
        let item = |source: u32| EdgeItem {
            uri: uri(format!(
                "at://did:plc:author/sh.tangled.graph.vouch/r{source}"
            )),
            sort_micros: 42,
            sort_source: source,
        };
        let mut items = vouch_items_from(PageCursor::Start, |cursor| {
            calls.set(calls.get() + 1);
            match cursor {
                PageCursor::Start => EdgePage {
                    items: vec![item(3), item(2)],
                    next: Some(PageToken::new(42, 2)),
                    total: None,
                },
                PageCursor::After(token) => {
                    assert_eq!(token, PageToken::new(42, 2));
                    EdgePage {
                        items: vec![item(1)],
                        next: None,
                        total: None,
                    }
                }
            }
        });
        assert_eq!(calls.get(), 0);
        assert_eq!(items.next().unwrap().sort_source, 3);
        assert_eq!(calls.get(), 1);
        assert_eq!(items.next().unwrap().sort_source, 2);
        assert_eq!(calls.get(), 1);
        assert_eq!(items.next().unwrap().sort_source, 1);
        assert_eq!(calls.get(), 2);
        assert!(items.next().is_none());
        assert!(items.next().is_none());
        assert_eq!(calls.get(), 2);
    }

    #[test]
    fn deferred_fetch_continues_through_an_empty_page_with_a_cursor() {
        let calls = std::cell::Cell::new(0usize);
        let mut items = vouch_items_from(PageCursor::Start, |cursor| {
            calls.set(calls.get() + 1);
            match cursor {
                PageCursor::Start => EdgePage {
                    items: Vec::new(),
                    next: Some(PageToken::new(42, 2)),
                    total: None,
                },
                PageCursor::After(token) => {
                    assert_eq!(token, PageToken::new(42, 2));
                    EdgePage {
                        items: vec![EdgeItem {
                            uri: uri("at://did:plc:author/sh.tangled.graph.vouch/r1".to_owned()),
                            sort_micros: 42,
                            sort_source: 1,
                        }],
                        next: None,
                        total: None,
                    }
                }
            }
        });
        assert_eq!(items.next().unwrap().sort_source, 1);
        assert_eq!(calls.get(), 2);
        assert!(items.next().is_none());
        assert_eq!(calls.get(), 2);
    }

    #[test]
    fn decimal_vouch_cursor_keeps_legacy_resume_semantics() {
        assert_eq!(
            parse_vouch_cursor(Some("42")).unwrap(),
            PageCursor::After(PageToken::new(42, 0)),
        );
    }

    fn vouch_edge(
        kind: &'static str,
        index_subject: &Did<DefaultStr>,
        author: &Did<DefaultStr>,
        record_subject: &Did<DefaultStr>,
        micros: u64,
    ) -> Edge {
        Edge {
            kind: nsid_static(kind),
            subject: SubjectRef::Did(index_subject.clone()),
            source: uri(format!(
                "at://{author}/sh.tangled.graph.vouch/{record_subject}"
            )),
            sort_micros: micros,
        }
    }

    #[test]
    fn viewer_authored_direct_vouch_is_returned() {
        let store = EdgeStore::new(RuntimeHasher::from_seeds(1, 2, 3, 4));
        let viewer = did("did:plc:viewer");
        let actor = did("did:plc:actor");
        store.add(vouch_edge(
            "sh.tangled.graph.vouch",
            &actor,
            &viewer,
            &actor,
            42,
        ));
        let network = VouchNetworkFilter::new(network_of(&viewer, Vec::new()));
        let (items, _) = network_vouch_page(
            &store,
            &actor,
            &network,
            PageCursor::Start,
            PageLimit::new(10).unwrap(),
        );

        assert_eq!(items.len(), 1);
        assert_eq!(source_authority_did(&items[0].uri), Some(viewer));
    }

    #[test]
    fn sent_vouches_match_their_record_subject_against_the_network() {
        let store = EdgeStore::new(RuntimeHasher::from_seeds(1, 2, 3, 4));
        let actor = did("did:plc:actor");
        let member = did("did:plc:member");
        let outsider = did("did:plc:outsider");
        store.add(vouch_edge(
            "sh.tangled.graph.vouch.by",
            &actor,
            &actor,
            &member,
            42,
        ));
        store.add(vouch_edge(
            "sh.tangled.graph.vouch.by",
            &actor,
            &actor,
            &outsider,
            41,
        ));
        let network = VouchNetworkFilter::new(HashSet::from([member.as_ref().to_owned()]));
        let (items, _) = network_vouch_page(
            &store,
            &actor,
            &network,
            PageCursor::Start,
            PageLimit::new(10).unwrap(),
        );

        assert_eq!(items.len(), 1);
        assert_eq!(items[0].uri.rkey().unwrap().as_ref(), member.as_ref());
    }

    #[test]
    fn batch_tags_each_actor_and_applies_the_limit_per_actor() {
        let store = EdgeStore::new(RuntimeHasher::from_seeds(1, 2, 3, 4));
        let first = did("did:plc:first");
        let second = did("did:plc:second");
        let authors = [
            did("did:plc:author1"),
            did("did:plc:author2"),
            did("did:plc:author3"),
            did("did:plc:author4"),
        ];
        for (index, actor) in [first.clone(), second.clone()].iter().enumerate() {
            for offset in 0..2 {
                let author = &authors[index * 2 + offset];
                store.add(vouch_edge(
                    "sh.tangled.graph.vouch",
                    actor,
                    author,
                    actor,
                    10 - offset as u64,
                ));
            }
        }
        let query = parse_network_vouch_query(Some(
            "actors=did%3Aplc%3Afirst&actors=did%3Aplc%3Asecond&limit=1",
        ))
        .unwrap();
        let network = VouchNetworkFilter::new(
            authors
                .iter()
                .map(|author| author.as_ref().to_owned())
                .collect(),
        );
        let (items, cursor) =
            network_vouch_pages(&store, &query.actors, &network, query.start, query.limit);

        assert_eq!(
            items
                .iter()
                .map(|(actor, _)| actor.clone())
                .collect::<Vec<_>>(),
            vec![first, second]
        );
        assert!(cursor.is_none());
    }

    #[test]
    fn repeated_actors_collapse_into_one_group() {
        let query = parse_network_vouch_query(Some(
            "actors=did%3Aplc%3Afirst&actors=did%3Aplc%3Afirst&actors=did%3Aplc%3Asecond",
        ))
        .unwrap();

        assert_eq!(
            query.actors,
            vec![did("did:plc:first"), did("did:plc:second")]
        );
    }

    fn vouch_state(edges: Arc<EdgeStore>) -> AppState {
        AppState::new(
            Arc::new(bobbin_record_lru::LruRecordStore::new(
                bobbin_record_lru::CacheCapacity::from_bytes(64 * 1024),
            )),
            SlingshotClient::with_default_http(url::Url::parse("http://127.0.0.1:1").unwrap())
                .unwrap(),
            edges,
            Arc::new(StateIndex::new(RuntimeHasher::default())),
            Arc::new(StateIndex::new(RuntimeHasher::default())),
            Arc::new(CoverageWatch::new()),
            Arc::new(
                KnotProxy::new(
                    bobbin_knot_proxy::KnotProxyConfig::default(),
                    bobbin_knot_proxy::KnotHttpConfig::default(),
                    Arc::new(bobbin_runtime::SystemClock::new()),
                    RuntimeHasher::default(),
                )
                .unwrap(),
            ),
            Arc::new(
                bobbin_search::SearchIndex::new(
                    bobbin_search::DEFAULT_WRITER_HEAP_BYTES,
                    Arc::new(bobbin_runtime::SystemClock::new()),
                )
                .unwrap(),
            ) as Arc<dyn SearchReader>,
            Arc::new(RepoIdResolver::detached(RuntimeHasher::default())),
            Arc::new(crate::default_directory()),
        )
    }

    fn cache_vouch(
        state: &AppState,
        author: &Did<DefaultStr>,
        subject: &Did<DefaultStr>,
        kind: &str,
    ) {
        let at = uri(format!("at://{author}/sh.tangled.graph.vouch/{subject}"));
        let value = serde_json::json!({
            "$type": "sh.tangled.graph.vouch",
            "kind": kind,
            "createdAt": "2026-05-01T00:00:00Z"
        });
        state.records.put(
            at.clone(),
            Arc::new(RecordBody {
                uri: at,
                cid: Cid::new(b"bafyreieqygohnz2zqyvtvktbjpvhutphobcmbsnt4q5lc36ri7vpcmoz4i")
                    .unwrap(),
                value: serde_json::to_vec(&value).unwrap().into(),
            }),
        );
    }

    #[tokio::test]
    async fn a_denounced_did_stays_out_of_the_viewer_network() {
        let edges = Arc::new(EdgeStore::new(RuntimeHasher::from_seeds(1, 2, 3, 4)));
        let viewer = did("did:plc:viewer");
        let friend = did("did:plc:friend");
        let enemy = did("did:plc:enemy");
        edges.add(vouch_edge(
            "sh.tangled.graph.vouch.by",
            &viewer,
            &viewer,
            &friend,
            42,
        ));
        edges.add(vouch_edge(
            "sh.tangled.graph.vouch.by",
            &viewer,
            &viewer,
            &enemy,
            41,
        ));
        let state = vouch_state(edges);
        cache_vouch(&state, &viewer, &friend, "vouch");
        cache_vouch(&state, &viewer, &enemy, "denounce");

        let network = vouch_network(&state, &viewer).await;

        assert!(network.contains(friend.as_ref()));
        assert!(!network.contains(enemy.as_ref()));
        assert!(network.contains(viewer.as_ref()));
    }

    #[test]
    fn cursor_is_rejected_for_multiple_actors() {
        assert!(matches!(
            parse_network_vouch_query(Some(
                "actors=did%3Aplc%3Afirst&actors=did%3Aplc%3Asecond&cursor=42"
            )),
            Err(XrpcError::InvalidParams(_))
        ));
    }
}

async fn list_network_vouches(
    State(state): State<AppState>,
    ExtractOptionalServiceAuth(auth): ExtractOptionalServiceAuth,
    RawQuery(query): RawQuery,
) -> Result<Response, XrpcError> {
    let q = parse_network_vouch_query(query.as_deref())?;
    let permit = state.heavy_permit()?;
    let vouch_nsid = nsid_static("sh.tangled.graph.vouch");
    let (page_items, next_cursor) = match auth.as_ref() {
        Some(auth) => {
            let viewer = auth.did().into_static();
            let network = VouchNetworkFilter::new(vouch_network(&state, &viewer).await);
            network_vouch_pages(&state.edges, &q.actors, &network, q.start, q.limit)
        }
        None => (Vec::new(), None),
    };
    let views = hydrate_keyed_record_stream::<_, Vouch<DefaultStr>>(
        &state,
        vouch_nsid,
        page_items,
        HitProvenance::Indexed,
    )
    .map_ok(|(actor, view)| NetworkVouchItem { actor, view });
    Ok(json_stream::<NetworkVouchItem<Vouch<DefaultStr>>, _>(
        "items",
        views,
        paged_tail(next_cursor, None),
        permit,
    ))
}

async fn list_stars_by(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
) -> Result<Response, XrpcError> {
    list_mirror::<StarBy, Star<DefaultStr>, _>(&state, q).await
}
async fn count_stars_by(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<CountQuery>,
) -> Result<Json<CountResponse>, XrpcError> {
    count_mirror::<StarBy>(&state, q).map(Json)
}

async fn list_reactions_by(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
) -> Result<Response, XrpcError> {
    list_mirror::<ReactionBy, Reaction<DefaultStr>, _>(&state, q).await
}
async fn count_reactions_by(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<CountQuery>,
) -> Result<Json<CountResponse>, XrpcError> {
    count_mirror::<ReactionBy>(&state, q).map(Json)
}

async fn list_follows_by(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
) -> Result<Response, XrpcError> {
    list_mirror::<FollowBy, Follow<DefaultStr>, _>(&state, q).await
}
async fn count_follows_by(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<CountQuery>,
) -> Result<Json<CountResponse>, XrpcError> {
    count_mirror::<FollowBy>(&state, q).map(Json)
}

#[derive(serde::Deserialize)]
struct InvitesByQuery {
    subject: Did<DefaultStr>,
}

#[derive(serde::Serialize)]
#[serde(rename_all = "camelCase")]
struct InviteItem {
    uri: AtUri<DefaultStr>,
    knot: Did<DefaultStr>,
    repo: Did<DefaultStr>,
    added_by: Did<DefaultStr>,
    created_at: Datetime,
}

#[derive(serde::Serialize)]
struct InvitesByOutput {
    items: Vec<InviteItem>,
    pending: Vec<Did<DefaultStr>>,
    truncated: bool,
}

const COLLABORATOR_INVITE_COLLECTION: &str = "sh.tangled.repo.collaboratorInvite";

fn list_invites_by(
    state: &AppState,
    q: InvitesByQuery,
) -> Result<Json<InvitesByOutput>, XrpcError> {
    let subject = q.subject;
    if !state.invites.discovered() {
        return Err(XrpcError::Warming);
    }

    let offered: Vec<(InviteScope, Offer)> = state.invites.outstanding(&subject);
    let truncated = offered.len() > MAX_OUTSTANDING_OFFERS;

    let items = offered
        .into_iter()
        .take(MAX_OUTSTANDING_OFFERS)
        .map(|(scope, offer)| {
            let created_at = datetime_of(offer.offered_at).ok_or_else(|| {
                XrpcError::Internal(format!(
                    "offer to {} has timestamp we can't render, {}",
                    subject.as_ref(),
                    offer.offered_at.raw()
                ))
            })?;
            let uri = AtUri::<DefaultStr>::new_owned(format!(
                "at://{}/{}/{}",
                scope.repo.as_ref(),
                COLLABORATOR_INVITE_COLLECTION,
                subject.as_ref()
            ))
            .map_err(|e| {
                XrpcError::Internal(format!(
                    "offer to {} won't form record uri, {e}",
                    subject.as_ref()
                ))
            })?;
            Ok(InviteItem {
                uri,
                knot: scope.knot,
                repo: scope.repo,
                added_by: offer.invited_by,
                created_at,
            })
        })
        .collect::<Result<Vec<InviteItem>, XrpcError>>()?;

    Ok(Json(InvitesByOutput {
        items,
        pending: state.invites.pending(),
        truncated,
    }))
}

async fn list_collaborator_invites_by(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<InvitesByQuery>,
) -> Result<Json<InvitesByOutput>, XrpcError> {
    list_invites_by(&state, q)
}

async fn list_label_ops_by(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
) -> Result<Response, XrpcError> {
    list_mirror::<LabelOpBy, LabelOp<DefaultStr>, _>(&state, q).await
}
async fn count_label_ops_by(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<CountQuery>,
) -> Result<Json<CountResponse>, XrpcError> {
    count_mirror::<LabelOpBy>(&state, q).map(Json)
}

async fn list_pipelines_by(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
) -> Result<Response, XrpcError> {
    list_mirror::<PipelineBy, Pipeline<DefaultStr>, _>(&state, q).await
}
async fn count_pipelines_by(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<CountQuery>,
) -> Result<Json<CountResponse>, XrpcError> {
    count_mirror::<PipelineBy>(&state, q).map(Json)
}

async fn list_pipeline_statuses_by(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
) -> Result<Response, XrpcError> {
    list_mirror::<PipelineStatusBy, PipelineStatus<DefaultStr>, _>(&state, q).await
}
async fn count_pipeline_statuses_by(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<CountQuery>,
) -> Result<Json<CountResponse>, XrpcError> {
    count_mirror::<PipelineStatusBy>(&state, q).map(Json)
}

async fn list_artifacts_by(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
) -> Result<Response, XrpcError> {
    list_mirror::<ArtifactBy, Artifact<DefaultStr>, _>(&state, q).await
}
async fn count_artifacts_by(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<CountQuery>,
) -> Result<Json<CountResponse>, XrpcError> {
    count_mirror::<ArtifactBy>(&state, q).map(Json)
}

async fn list_collaborators_by(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
) -> Result<Response, XrpcError> {
    let (page, nsid) = mirror_edge_page::<CollaboratorBy, NoFilter>(&state, &q)?;
    collaborator_list_response(&state, page, nsid)
}
async fn count_collaborators_by(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<CountQuery>,
) -> Result<Json<CountResponse>, XrpcError> {
    count_mirror::<CollaboratorBy>(&state, q).map(Json)
}

async fn list_issues_by(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<TypedListQuery<IssueFilter>>,
) -> Result<Response, XrpcError> {
    let (page, nsid) = mirror_edge_page::<IssueBy, _>(&state, &q)?;
    let permit = state.heavy_permit()?;
    let owned = state.clone();
    let items = hydrate_record_stream::<Issue<DefaultStr>>(
        &state,
        nsid,
        page.items,
        HitProvenance::Indexed,
    )
    .map(move |view| view.map(|v| enrich_issue_view(&owned, v)));
    Ok(json_stream::<StatefulItem<Issue<DefaultStr>>, _>(
        "items",
        items,
        paged_tail(page.next.map(PageToken::encode_token), page.total),
        permit,
    ))
}
async fn count_issues_by(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<CountQuery>,
) -> Result<Json<CountResponse>, XrpcError> {
    count_mirror::<IssueBy>(&state, q).map(Json)
}

async fn list_feed_comments_by(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
) -> Result<Response, XrpcError> {
    list_mirror::<FeedCommentBy, FeedComment<DefaultStr>, _>(&state, q).await
}
async fn count_feed_comments_by(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<CountQuery>,
) -> Result<Json<CountResponse>, XrpcError> {
    count_mirror::<FeedCommentBy>(&state, q).map(Json)
}

async fn list_issue_states_by(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
) -> Result<Response, XrpcError> {
    list_mirror::<IssueStateBy, IssueState<DefaultStr>, _>(&state, q).await
}
async fn count_issue_states_by(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<CountQuery>,
) -> Result<Json<CountResponse>, XrpcError> {
    count_mirror::<IssueStateBy>(&state, q).map(Json)
}

async fn list_pulls_by(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<TypedListQuery<PullFilter>>,
) -> Result<Response, XrpcError> {
    let (page, nsid) = mirror_edge_page::<PullBy, _>(&state, &q)?;
    let permit = state.heavy_permit()?;
    let owned = state.clone();
    let items =
        hydrate_record_stream::<Pull<DefaultStr>>(&state, nsid, page.items, HitProvenance::Indexed)
            .map(move |view| view.map(|v| enrich_pull_view(&owned, v)));
    Ok(json_stream::<StatefulItem<Pull<DefaultStr>>, _>(
        "items",
        items,
        paged_tail(page.next.map(PageToken::encode_token), page.total),
        permit,
    ))
}
async fn count_pulls_by(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<CountQuery>,
) -> Result<Json<CountResponse>, XrpcError> {
    count_mirror::<PullBy>(&state, q).map(Json)
}

async fn list_pull_statuses_by(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
) -> Result<Response, XrpcError> {
    list_mirror::<PullStatusBy, PullStatus<DefaultStr>, _>(&state, q).await
}
async fn count_pull_statuses_by(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<CountQuery>,
) -> Result<Json<CountResponse>, XrpcError> {
    count_mirror::<PullStatusBy>(&state, q).map(Json)
}

async fn list_spindle_members_by(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
) -> Result<Response, XrpcError> {
    list_mirror::<SpindleMemberBy, SpindleMember<DefaultStr>, _>(&state, q).await
}
async fn count_spindle_members_by(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<CountQuery>,
) -> Result<Json<CountResponse>, XrpcError> {
    count_mirror::<SpindleMemberBy>(&state, q).map(Json)
}

async fn list_label_definitions(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
) -> Result<Response, XrpcError> {
    list_records::<LabelDefinitionRecord, LabelDefinition<DefaultStr>, _>(&state, q).await
}

async fn count_label_definitions(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<CountQuery>,
) -> Result<Json<CountResponse>, XrpcError> {
    count_for::<LabelDefinitionRecord>(&state, q).map(Json)
}

async fn list_label_ops(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
) -> Result<Response, XrpcError> {
    list_records::<LabelOpRecord, LabelOp<DefaultStr>, _>(&state, q).await
}

async fn count_label_ops(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<CountQuery>,
) -> Result<Json<CountResponse>, XrpcError> {
    count_for::<LabelOpRecord>(&state, q).map(Json)
}

async fn list_pipelines(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
) -> Result<Response, XrpcError> {
    list_records::<PipelineRecord, Pipeline<DefaultStr>, _>(&state, q).await
}

async fn count_pipelines(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<CountQuery>,
) -> Result<Json<CountResponse>, XrpcError> {
    count_for::<PipelineRecord>(&state, q).map(Json)
}

async fn list_pipeline_statuses(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
) -> Result<Response, XrpcError> {
    list_records::<PipelineStatusRecord, PipelineStatus<DefaultStr>, _>(&state, q).await
}

async fn count_pipeline_statuses(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<CountQuery>,
) -> Result<Json<CountResponse>, XrpcError> {
    count_for::<PipelineStatusRecord>(&state, q).map(Json)
}

async fn list_artifacts(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
) -> Result<Response, XrpcError> {
    list_records::<ArtifactRecord, Artifact<DefaultStr>, _>(&state, q).await
}

async fn count_artifacts(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<CountQuery>,
) -> Result<Json<CountResponse>, XrpcError> {
    count_for::<ArtifactRecord>(&state, q).map(Json)
}

async fn list_spindle_members(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
) -> Result<Response, XrpcError> {
    list_records::<SpindleMemberRecord, SpindleMember<DefaultStr>, _>(&state, q).await
}

async fn count_spindle_members(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<CountQuery>,
) -> Result<Json<CountResponse>, XrpcError> {
    count_for::<SpindleMemberRecord>(&state, q).map(Json)
}

async fn list_strings(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<TypedListQuery<NoFilter>>,
) -> Result<Response, XrpcError> {
    list_records::<TangledStringRecord, TangledString<DefaultStr>, _>(&state, q).await
}

async fn count_strings(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<CountQuery>,
) -> Result<Json<CountResponse>, XrpcError> {
    count_for::<TangledStringRecord>(&state, q).map(Json)
}

#[derive(Deserialize)]
struct ResolveMiniDocParams {
    identifier: AtIdentifier<DefaultStr>,
}

async fn resolve_mini_doc(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<ResolveMiniDocParams>,
) -> Result<Response, XrpcError> {
    let doc = state
        .identity
        .resolve_minidoc(&q.identifier)
        .await
        .map_err(|error| match error {
            IdentityResolveError::NotFound => XrpcError::NotFound,
            IdentityResolveError::Upstream(message) => XrpcError::UpstreamUnavailable(message),
            IdentityResolveError::Decode(message) => XrpcError::InvalidRecord(message),
        })?;
    Ok(Json(doc).into_response())
}

async fn get_coverage(State(state): State<AppState>) -> Json<CoverageEnvelope> {
    let mut envelope = CoverageEnvelope::from(state.coverage.snapshot());
    envelope.stream = state
        .stream_health
        .as_deref()
        .map(|health| health.snapshot().into());
    Json(envelope)
}
// enough for a write to travel pds -> relay -> indexer -> bobbin, plus slack
// for a slow hop
const AWAIT_RECORD_TIMEOUT: Duration = Duration::from_secs(5);

#[derive(Debug, Deserialize)]
struct AwaitRecordQuery {
    uri: AtUri<DefaultStr>,
    cid: Option<ParsedCid>,
}

#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
struct AwaitRecordResponse {
    status: &'static str,
    #[serde(skip_serializing_if = "Option::is_none")]
    cid: Option<ParsedCid>,
    #[serde(skip_serializing_if = "Option::is_none")]
    reason: Option<&'static str>,
    #[serde(skip_serializing_if = "Option::is_none")]
    detail: Option<Box<str>>,
}

impl From<Settlement> for AwaitRecordResponse {
    fn from(s: Settlement) -> Self {
        Self {
            status: s.outcome.as_str(),
            cid: s.cid,
            reason: s.outcome.rejection().map(Rejection::as_str),
            detail: s.outcome.detail().map(Box::from),
        }
    }
}

async fn await_record(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<AwaitRecordQuery>,
) -> Result<Json<AwaitRecordResponse>, XrpcError> {
    let Some((collection, _)) = q.uri.collection().zip(q.uri.rkey()) else {
        return Err(XrpcError::InvalidParams(
            "uri must be at://<did>/<collection>/<rkey>".into(),
        ));
    };
    if !collection.as_str().starts_with(TANGLED_NSID_PREFIX)
        && !collection.as_str().starts_with(ORG_TANGLED_NSID_PREFIX)
    {
        return Err(XrpcError::InvalidParams(format!(
            "bobbin never sees {collection} records"
        )));
    }
    // records are only ever indexed under their did, a handle would wait forever
    if source_authority_did(&q.uri).is_none() {
        return Err(XrpcError::InvalidParams(
            "uri authority must be a did, not a handle".into(),
        ));
    }
    // Drop releases the await slot when the handler returns
    let _awaiting = state.awaiting.try_enter()?;
    let settled = match state.settlements.watch(&q.uri, q.cid) {
        Watch::Settled(s) => s,
        // Drop removes the waiter from the list when the handler returns,
        // so a timeout won't leave a channel no one is waiting on
        Watch::Waiting(_registration, rx) => {
            match tokio::time::timeout(AWAIT_RECORD_TIMEOUT, rx).await {
                Ok(Ok(s)) => s,
                Ok(Err(_)) | Err(_) => return Err(XrpcError::NotSettled),
            }
        }
    };
    Ok(Json(settled.into()))
}
#[derive(Deserialize)]
#[allow(dead_code)]
struct ListRecipientsQuery {
    subject: String,
    collection: Option<String>,
}

#[derive(Serialize)]
struct ListRecipientsResponse {
    dids: Vec<String>,
}

async fn list_recipients(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<ListRecipientsQuery>,
) -> Result<Json<ListRecipientsResponse>, XrpcError> {
    use std::collections::BTreeSet;

    let subject_ref = if let Ok(uri) = AtUri::<DefaultStr>::new_owned(&q.subject) {
        SubjectRef::Uri(uri)
    } else if let Ok(did) = Did::<DefaultStr>::new_owned(&q.subject) {
        SubjectRef::Did(did)
    } else {
        return Err(XrpcError::InvalidParams(
            "subject must be an at-uri or did".into(),
        ));
    };

    let key = EdgeKey::new(nsid_static("org.tangled.feed.subscription"), subject_ref);
    let sources = state.edges.sources_for(&key);

    // NOTE: collection filtering intentionally disabled.
    // All subscribers are returned regardless of their stored collection filter.
    let filtered = sources;
    let mut seen = BTreeSet::new();
    let mut dids = Vec::new();
    for src in &filtered {
        if let Some(did) = owner_did_from_aturi(src)
            && seen.insert(did.to_string())
        {
            dids.push(did.to_string());
        }
    }

    Ok(Json(ListRecipientsResponse { dids }))
}

async fn search_actors_typeahead(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<ActorTypeaheadParams>,
) -> Result<Json<ActorTypeaheadResponse>, XrpcError> {
    let query = q.q.trim().trim_start_matches('@');
    if query.is_empty() {
        return Err(XrpcError::InvalidParams("q must not be empty".into()));
    }
    if query.len() > ACTOR_TYPEAHEAD_MAX_QUERY_BYTES {
        return Err(XrpcError::InvalidParams(format!(
            "q must be at most {ACTOR_TYPEAHEAD_MAX_QUERY_BYTES} bytes",
        )));
    }
    let limit = q.limit.unwrap_or(ACTOR_TYPEAHEAD_DEFAULT_LIMIT);
    if !(1..=ACTOR_TYPEAHEAD_MAX_LIMIT).contains(&limit) {
        return Err(XrpcError::InvalidParams(format!(
            "limit must be between 1 and {ACTOR_TYPEAHEAD_MAX_LIMIT}",
        )));
    }
    let _permit = state.heavy_permit()?;
    let suggestions = state
        .actors
        .suggest(query, limit)
        .await
        .map_err(map_actor_search_err)?;
    let actors = suggestions
        .into_iter()
        .map(|suggestion| {
            let uri = AtUri::new_owned(format!(
                "at://{}/sh.tangled.actor.profile/self",
                suggestion.did
            ))
            .map_err(|error| {
                XrpcError::Internal(format!("could not build profile URI: {error}"))
            })?;
            Ok(ActorTypeaheadView {
                uri,
                did: suggestion.did,
                handle: suggestion.handle,
            })
        })
        .collect::<Result<Vec<_>, XrpcError>>()?;
    Ok(Json(ActorTypeaheadResponse { actors }))
}
async fn search_query(
    State(state): State<AppState>,
    XrpcQuery(q): XrpcQuery<SearchQueryParams>,
) -> Result<Response, XrpcError> {
    if q.q.trim().is_empty() {
        return Err(XrpcError::InvalidParams("q must not be empty".into()));
    }
    let cursor = match (q.cursor.as_deref(), q.offset) {
        (Some(_), Some(_)) => {
            return Err(XrpcError::InvalidParams(
                "cursor and offset are mutually exclusive".into(),
            ));
        }
        (Some(c), None) => SearchCursor::from_token(Some(c))
            .map_err(|e| XrpcError::InvalidParams(format!("cursor: {e}")))?,
        (None, Some(o)) => SearchCursor::At(SearchOffset::new(o.get() as u32)),
        (None, None) => SearchCursor::Start,
    };
    let limit = parse_limit(q.limit)?;
    let filters = build_search_filters(&q)?;
    let permit = state.heavy_permit()?;
    let page = state
        .search
        .search(&q.q, filters, cursor, limit.get())
        .await
        .map_err(map_search_err)?;
    let next = page.next.map(SearchOffset::encode_token);
    let owned = state.clone();
    let hits = hydrate_stream(page.hits, move |hit| {
        let owned = owned.clone();
        let uri = hit.uri.clone();
        let nsid = hit.nsid.clone();
        async move {
            let result = hydrate_search_hit(&owned, hit).await;
            drop_unhydratable(HitProvenance::Indexed, &nsid, &uri, result)
        }
    });
    Ok(json_stream::<SearchHitView, _>(
        "hits",
        hits,
        paged_tail(next, None),
        permit,
    ))
}

fn build_search_filters(q: &SearchQueryParams) -> Result<SearchFilters, XrpcError> {
    let since = q
        .since
        .as_deref()
        .map(parse_rfc3339_seconds)
        .transpose()
        .map_err(|e| XrpcError::InvalidParams(format!("since: {e}")))?;
    let until = q
        .until
        .as_deref()
        .map(parse_rfc3339_seconds)
        .transpose()
        .map_err(|e| XrpcError::InvalidParams(format!("until: {e}")))?;
    if let (Some(s), Some(u)) = (since, until)
        && s > u
    {
        return Err(XrpcError::InvalidParams("since must be <= until".into()));
    }
    Ok(SearchFilters {
        nsid: q.nsid.clone(),
        author: q.author.clone(),
        repo: q.repo.clone(),
        since,
        until,
    })
}

fn parse_rfc3339_seconds(raw: &str) -> Result<i64, String> {
    chrono::DateTime::parse_from_rfc3339(raw)
        .map(|dt| dt.timestamp())
        .map_err(|e| format!("expected RFC3339, got {raw}: {e}"))
}

async fn hydrate_search_hit(
    state: &AppState,
    hit: SearchHit,
) -> Result<Option<SearchHitView>, XrpcError> {
    let SearchHit { uri, nsid, score } = hit;
    let body = resolve_for_view(state, &nsid, uri).await?;
    let record = decode_canon_or_upgrade(&nsid, &body.value, &state.resolver)
        .await
        .map_err(|err| XrpcError::InvalidRecord(err.to_string()))?;
    let Some(value) = SearchableRecord::try_from_record(record) else {
        return Ok(None);
    };
    let Some(value) = value.normalize(&state.resolver).await else {
        return Ok(None);
    };
    Ok(Some(SearchHitView {
        uri: body.uri.clone(),
        cid: Some(body.cid.clone()),
        nsid,
        score,
        value,
    }))
}

fn map_search_err(err: SearchError) -> XrpcError {
    use SearchError as E;
    match err {
        E::Query(e) => XrpcError::InvalidParams(format!("query: {e}")),
        e @ (E::Tantivy(_)
        | E::InvalidUri(_)
        | E::InvalidNsid(_)
        | E::MissingField(_)
        | E::Cancelled(_)) => XrpcError::Internal(format!("search: {e}")),
    }
}
fn map_actor_search_err(err: ActorIndexError) -> XrpcError {
    XrpcError::Internal(format!("actor search: {err}"))
}

fn map_proxy_error(err: KnotProxyError) -> XrpcError {
    match err {
        KnotProxyError::CircuitOpen => {
            XrpcError::UpstreamUnavailable("knot circuit breaker open".into())
        }
        KnotProxyError::BlockedHost { host, reason } => {
            XrpcError::InvalidRecord(format!("knot host {host} is {reason} address space"))
        }
        KnotProxyError::PlaintextHttp { host } => {
            XrpcError::InvalidRecord(format!("knot host {host} requires https"))
        }
        KnotProxyError::Connect(e) => XrpcError::UpstreamUnavailable(format!("connect: {e}")),
        KnotProxyError::Timeout(e) => {
            XrpcError::UpstreamUnavailable(format!("upstream timeout: {e}"))
        }
        KnotProxyError::Redirect(e) => XrpcError::UpstreamUnavailable(format!("redirect: {e}")),
        KnotProxyError::Transport(e) => XrpcError::UpstreamUnavailable(format!("transport: {e}")),
        KnotProxyError::Upstream(s) => XrpcError::UpstreamUnavailable(format!("status {s}")),
    }
}

fn validate_client_supplied_knot(state: &AppState, host: &KnotHost) -> Result<(), XrpcError> {
    let host_str = || host.url().host_str().unwrap_or_default().to_owned();
    if state.knots.requires_https() && host.url().scheme() != "https" {
        return Err(XrpcError::InvalidParams(format!(
            "knot host {} must be https",
            host_str(),
        )));
    }
    if state.knots.allows_private_hosts() {
        return Ok(());
    }
    match host.private_literal_reason() {
        None => Ok(()),
        Some(reason) => Err(XrpcError::InvalidParams(format!(
            "knot host {} blocked: {} address space",
            host_str(),
            reason,
        ))),
    }
}

struct RepoTarget {
    host: KnotHost,
    slug: RepoSlug,
    repo_did: Option<Did<DefaultStr>>,
}

async fn resolve_knot_target(
    state: &AppState,
    repo_uri: AtUri<DefaultStr>,
) -> Result<RepoTarget, XrpcError> {
    let rkey: Option<Rkey<DefaultStr>> = repo_uri.rkey().map(|r| r.clone().into_static());
    let (body, did) = resolve(state, ExpectedNsid::from_static(RepoRecord::NSID), repo_uri).await?;
    let value: Repo<DefaultStr> = serde_json::from_slice(&body.value)
        .map_err(|e| XrpcError::InvalidRecord(format!("decode repo record: {e}")))?;
    let host = KnotHost::parse(value.knot.as_ref())
        .map_err(|e| XrpcError::InvalidRecord(format!("knot field: {e}")))?;
    let name = pick_human_slug(rkey.as_ref(), value.name.as_deref()).ok_or_else(|| {
        XrpcError::InvalidRecord("at-uri missing rkey and record missing name".to_string())
    })?;
    let slug = RepoSlug::new(&did, &name)
        .map_err(|e| XrpcError::InvalidRecord(format!("repo slug: {e}")))?;
    Ok(RepoTarget {
        host,
        slug,
        repo_did: value.repo_did,
    })
}

fn pick_human_slug(rkey: Option<&Rkey<DefaultStr>>, name: Option<&str>) -> Option<String> {
    match rkey {
        Some(r) if jacquard_common::types::tid::Tid::new(r.as_ref()).is_ok() => {
            Some(name.unwrap_or(r.as_ref()).to_owned())
        }
        Some(r) => Some(r.as_ref().to_owned()),
        None => name.map(str::to_owned),
    }
}

fn filter_request_headers(
    client: &HeaderMap,
    socket: SocketPeer,
    address: &ClientAddress,
) -> HeaderMap {
    let forwarded = RANGE_OR_CONDITIONAL_HEADERS
        .iter()
        .fold(HeaderMap::new(), |mut acc, name| {
            if let Some(value) = client.get(*name) {
                acc.insert((*name).clone(), value.clone());
            }
            acc
        });
    address
        .of(client, socket)
        .into_iter()
        .fold(forwarded, |mut acc, address| {
            acc.insert(X_FORWARDED_FOR.clone(), address);
            acc
        })
}

fn upstream_to_axum(resp: ProxyResponse) -> Response {
    let status = resp.status();
    let upstream_headers = resp.headers().clone();
    let body = Body::from_stream(resp.into_body_stream());
    let mut response = Response::builder()
        .status(status)
        .body(body)
        .expect("response body construction must succeed");
    let response_headers = response.headers_mut();
    PASSTHROUGH_HEADERS.iter().for_each(|name| {
        if let Some(value) = upstream_headers.get(*name) {
            response_headers.insert((*name).clone(), value.clone());
        }
    });
    response
}

async fn dispatch_knot(
    state: &AppState,
    nsid: &Nsid<DefaultStr>,
    host: &KnotHost,
    query: &[(&str, &str)],
    headers: HeaderMap,
) -> Result<Response, XrpcError> {
    let upstream = state
        .knots
        .forward(host, nsid, query, headers)
        .await
        .map_err(map_proxy_error)?;
    Ok(upstream_to_axum(upstream))
}

async fn dispatch_mirror(
    state: &AppState,
    nsid: &Nsid<DefaultStr>,
    target: &RepoTarget,
    query: &[(&str, &str)],
    headers: &HeaderMap,
) -> Option<Response> {
    let mirror = state.mirror.as_ref()?;
    let repo_did = target.repo_did.as_ref()?;
    if RANGE_OR_CONDITIONAL_HEADERS
        .iter()
        .any(|name| headers.contains_key(*name))
    {
        return None;
    }
    let mirror_nsid = MirrorNsid::route(nsid.as_ref(), query)?;
    let refused = match mirror
        .forward(&mirror_nsid, repo_did, query, headers.clone())
        .await
    {
        Ok(upstream) if !upstream.status().is_client_error() => {
            return Some(upstream_to_axum(upstream));
        }
        Ok(upstream) => {
            let status = upstream.status();
            upstream.discard().await;
            status.to_string()
        }
        Err(KnotProxyError::Upstream(status)) => status.to_string(),
        Err(err @ KnotProxyError::CircuitOpen) => err.to_string(),
        Err(err) => {
            tracing::warn!(
                nsid = mirror_nsid.as_str(),
                repo = repo_did.as_ref(),
                error = %err,
                "mirror call failed, asking the knot",
            );
            return None;
        }
    };
    tracing::debug!(
        nsid = mirror_nsid.as_str(),
        repo = repo_did.as_ref(),
        refused,
        "mirror couldn't serve this repo, asking the knot",
    );
    None
}

fn extract_param(
    params: ProxyParams,
    key: &str,
) -> Result<Option<(String, ProxyParams)>, XrpcError> {
    let (matching, rest): (ProxyParams, ProxyParams) =
        params.into_iter().partition(|(k, _)| k == key);
    match matching.as_slice() {
        [] => Ok(None),
        [_] => Ok(matching.into_iter().next().map(|(_, v)| (v, rest))),
        _ => Err(XrpcError::InvalidParams(format!(
            "{key} parameter must appear at most once, got {}",
            matching.len(),
        ))),
    }
}

async fn proxy_repo_handler(
    state: AppState,
    headers: HeaderMap,
    socket: SocketPeer,
    params: ProxyParams,
    nsid: Nsid<DefaultStr>,
) -> Result<Response, XrpcError> {
    let (repo_raw, rest) = extract_param(params, REPO_PARAM)?
        .ok_or_else(|| XrpcError::InvalidParams("missing repo".into()))?;
    let repo_uri = parse_uri(&repo_raw)?;
    let target = resolve_knot_target(&state, repo_uri).await?;
    let allowed = filter_request_headers(&headers, socket, &state.client_address);
    let query: Vec<(&str, &str)> = rest.iter().map(|(k, v)| (k.as_str(), v.as_str())).collect();
    if let Some(response) = dispatch_mirror(&state, &nsid, &target, &query, &allowed).await {
        return Ok(response);
    }
    let forward: Vec<(&str, &str)> = query
        .into_iter()
        .chain(std::iter::once((REPO_PARAM, target.slug.as_str())))
        .collect();
    dispatch_knot(&state, &nsid, &target.host, &forward, allowed).await
}

async fn proxy_knot_handler(
    state: AppState,
    headers: HeaderMap,
    socket: SocketPeer,
    params: ProxyParams,
    nsid: Nsid<DefaultStr>,
) -> Result<Response, XrpcError> {
    let (knot_raw, rest) = extract_param(params, KNOT_HOST_PARAM)?
        .ok_or_else(|| XrpcError::InvalidParams("missing knot".into()))?;
    let host =
        KnotHost::parse(&knot_raw).map_err(|e| XrpcError::InvalidParams(format!("knot: {e}")))?;
    validate_client_supplied_knot(&state, &host)?;
    let allowed = filter_request_headers(&headers, socket, &state.client_address);
    let forward: Vec<(&str, &str)> = rest.iter().map(|(k, v)| (k.as_str(), v.as_str())).collect();
    dispatch_knot(&state, &nsid, &host, &forward, allowed).await
}

async fn proxy_mirror_v2_handler(
    state: AppState,
    headers: HeaderMap,
    socket: SocketPeer,
    params: ProxyParams,
    nsid: Nsid<DefaultStr>,
) -> Result<Response, XrpcError> {
    let mirror_v2 = state
        .mirror_v2
        .as_ref()
        .ok_or_else(|| XrpcError::UpstreamUnavailable("mirror_v2.url is unset".into()))?;
    let allowed = filter_request_headers(&headers, socket, &state.client_address);
    let forward: Vec<(&str, &str)> = params
        .iter()
        .map(|(k, v)| (k.as_str(), v.as_str()))
        .collect();
    mirror_v2
        .forward_raw(&nsid, &forward, allowed)
        .await
        .map(upstream_to_axum)
        .map_err(map_proxy_error)
}

async fn proxy_mirror_v2_write_handler(
    state: AppState,
    headers: HeaderMap,
    body: Bytes,
    nsid: Nsid<DefaultStr>,
) -> Result<Response, XrpcError> {
    let mirror_v2 = state
        .mirror_v2
        .as_ref()
        .ok_or_else(|| XrpcError::UpstreamUnavailable("mirror_v2.url is unset".into()))?;
    let token = headers
        .get(AUTHORIZATION)
        .ok_or_else(|| XrpcError::AuthRequired("the mirror needs a service-auth token".into()))?;

    let url = format!("{}xrpc/{nsid}", mirror_v2.host().url().as_str(),);
    let request = reqwest::Client::new()
        .post(url)
        .header(AUTHORIZATION, token)
        .header(
            CONTENT_TYPE,
            headers
                .get(CONTENT_TYPE)
                .cloned()
                .unwrap_or(axum::http::HeaderValue::from_static("application/json")),
        )
        .body(body);

    let upstream = request
        .send()
        .await
        .map_err(|e| XrpcError::UpstreamUnavailable(format!("mirror: {e}")))?;

    let status = upstream.status();
    let upstream_headers = upstream.headers().clone();
    let mut response = Response::builder()
        .status(status)
        .body(Body::from_stream(upstream.bytes_stream()))
        .expect("response body construction must succeed");
    let response_headers = response.headers_mut();
    PASSTHROUGH_HEADERS.iter().for_each(|name| {
        if let Some(value) = upstream_headers.get(*name) {
            response_headers.insert((*name).clone(), value.clone());
        }
    });
    Ok(response)
}