use std::ops::Not;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::{Duration, Instant};

use futures::future::BoxFuture;
use hickory_resolver::TokioResolver;
use hickory_resolver::proto::rr::{RData, rdata::TXT};
use jacquard_common::Data;
use jacquard_common::IntoStatic;
use jacquard_common::types::crypto::PublicKey;
use jacquard_common::types::did_doc::DidDocument;
use jacquard_common::types::ident::AtIdentifier;
use jacquard_common::types::string::{Did, Handle};
use miette::{Diagnostic, IntoDiagnostic};
use reqwest::StatusCode;
use scc::HashCache;
use serde::Deserialize;
use smol_str::SmolStr;
use thiserror::Error;
use url::Url;

use crate::net::{
    HANDLE_REQUEST_TIMEOUT, PublicHttpClient, PublicHttpError, handle_http, public_http,
};

// as per spec: https://web.plc.directory/spec/v0.1/did-plc
// "As an anti-abuse mechanism, operations have a maximum size when encoded as
// DAG-CBOR. The current limit is 7500 bytes."
const MAX_DID_DOC_BYTES: usize = 16384; // this is generous i think but we deal with json so..
const MAX_HANDLE_BODY_BYTES: usize = 8192;
const PUBLIC_HANDLE_RESOLVER: &str = "https://public.api.bsky.app/";

/// the total budget for every source used to resolve one handle.
const HANDLE_VERIFY_TIMEOUT: Duration = HANDLE_REQUEST_TIMEOUT;

/// how long a successful handle verdict is reused; failures are never stored.
const HANDLE_VERIFY_TTL: Duration = Duration::from_secs(30);

trait HandleDns: Send + Sync {
    fn resolve(&self, handle: Handle) -> BoxFuture<'static, Option<Did>>;
}

struct SystemHandleDns(TokioResolver);

impl SystemHandleDns {
    fn new() -> Self {
        let mut builder = TokioResolver::builder_tokio().expect("system DNS config must load");
        builder.options_mut().timeout = HANDLE_REQUEST_TIMEOUT;
        Self(builder.build().expect("system DNS resolver must build"))
    }
}

impl HandleDns for SystemHandleDns {
    fn resolve(&self, handle: Handle) -> BoxFuture<'static, Option<Did>> {
        let resolver = self.0.clone();
        Box::pin(async move {
            let records = resolver
                .txt_lookup(format!("_atproto.{}.", handle.as_str()))
                .await
                .ok()?;
            records
                .answers()
                .iter()
                .find_map(|record| match &record.data {
                    RData::TXT(txt) => did_from_txt(txt),
                    _ => None,
                })
        })
    }
}

fn did_from_txt(txt: &TXT) -> Option<Did> {
    let value = txt
        .txt_data
        .iter()
        .flat_map(|part| part.iter().copied())
        .collect::<Vec<_>>();
    std::str::from_utf8(&value)
        .ok()?
        .strip_prefix("did=")
        .and_then(|did| Did::new_owned(did).ok())
}

#[cfg(test)]
struct NoHandleDns;

#[cfg(test)]
impl HandleDns for NoHandleDns {
    fn resolve(&self, _handle: Handle) -> BoxFuture<'static, Option<Did>> {
        Box::pin(async { None })
    }
}

#[derive(Debug, Diagnostic, Error)]
pub enum ResolverError {
    #[error("{0}")]
    Generic(miette::Report),
    #[error("too many requests")]
    Ratelimited,
    #[error("handle resolution exhausted: {0}")]
    HandleResolutionExhausted(miette::Report),
    #[error("transport error: {0}")]
    Transport(SmolStr),
    #[error("no PDS service found in DID Doc for {0}")]
    MissingPds(Did),
}

impl From<miette::Report> for ResolverError {
    fn from(report: miette::Report) -> Self {
        ResolverError::Generic(report)
    }
}

#[derive(Clone, Debug)]
pub struct MiniDoc {
    pub pds: Url,
    pub handle: Option<Handle>,
    pub key: Option<PublicKey<'static>>,
}

#[derive(Clone)]
enum HandleSource {
    HttpsWellKnown,
    Xrpc(Url),
}

#[derive(Deserialize)]
struct ResolveHandleOutput {
    did: Did,
}

struct ResolverInner {
    plc_urls: Vec<Url>,
    handle_sources: Vec<HandleSource>,
    documents: PublicHttpClient,
    handles: PublicHttpClient,
    handle_dns: Arc<dyn HandleDns>,
    next_idx: AtomicUsize,
    cache: HashCache<Did, MiniDoc>,
    /// when a `(did, handle)` verdict, always positive, stops being reused.
    verified_handles: HashCache<(Did, Handle), Instant>,
}

#[derive(Clone)]
pub struct Resolver {
    inner: Arc<ResolverInner>,
}

fn production_handle_sources() -> Vec<HandleSource> {
    vec![
        HandleSource::HttpsWellKnown,
        HandleSource::Xrpc(
            Url::parse(PUBLIC_HANDLE_RESOLVER).expect("public resolver URL is valid"),
        ),
    ]
}

impl Resolver {
    pub fn new(plc_urls: Vec<Url>, identity_cache_size: u64) -> Self {
        Self::with_clients(
            plc_urls,
            production_handle_sources(),
            identity_cache_size,
            public_http(),
            handle_http(),
            Arc::new(SystemHandleDns::new()),
        )
    }

    /// one stub serves both the DID document and `resolveHandle`.
    #[cfg(all(test, feature = "indexer_stream"))]
    pub(crate) fn stubbed_for_test(
        base: Url,
        identity_cache_size: u64,
        public: PublicHttpClient,
    ) -> Self {
        Self::with_clients(
            vec![base.clone()],
            vec![HandleSource::Xrpc(base)],
            identity_cache_size,
            public.clone(),
            public,
            Arc::new(NoHandleDns),
        )
    }

    #[cfg(test)]
    pub(crate) fn with_plc_urls_and_client(
        plc_urls: Vec<Url>,
        identity_cache_size: u64,
        public: PublicHttpClient,
    ) -> Self {
        Self::with_clients(
            plc_urls,
            production_handle_sources(),
            identity_cache_size,
            public.clone(),
            public,
            Arc::new(NoHandleDns),
        )
    }

    fn with_clients(
        plc_urls: Vec<Url>,
        handle_sources: Vec<HandleSource>,
        identity_cache_size: u64,
        documents: PublicHttpClient,
        handles: PublicHttpClient,
        handle_dns: Arc<dyn HandleDns>,
    ) -> Self {
        assert!(
            !plc_urls.is_empty(),
            "at least one PLC source must be provided"
        );
        let capacity = std::cmp::min(1000, (identity_cache_size / 100) as usize);

        Self {
            inner: Arc::new(ResolverInner {
                plc_urls,
                handle_sources,
                documents,
                handles,
                handle_dns,
                next_idx: AtomicUsize::new(0),
                cache: HashCache::with_capacity(capacity, identity_cache_size as usize),
                verified_handles: HashCache::with_capacity(capacity, identity_cache_size as usize),
            }),
        }
    }

    pub async fn invalidate(&self, did: &Did) {
        self.inner.cache.remove_async(did).await;
    }

    pub fn invalidate_sync(&self, did: &Did) {
        self.inner.cache.remove_sync(did);
    }

    pub async fn resolve_did(&self, identifier: &AtIdentifier) -> Result<Did, ResolverError> {
        match identifier {
            AtIdentifier::Did(did) => Ok(did.clone()),
            AtIdentifier::Handle(handle) => {
                self.resolve_handle_with_timeout(handle, HANDLE_VERIFY_TIMEOUT)
                    .await
            }
        }
    }

    pub async fn resolve_doc(&self, did: &Did) -> Result<MiniDoc, ResolverError> {
        let did_static = did.clone();
        if let Some(entry) = self.inner.cache.get_async(&did_static).await {
            return Ok(entry.get().clone());
        }

        let mini = self.resolve_doc_fresh(did).await?;
        let _ = self.inner.cache.put_async(did_static, mini.clone()).await;
        Ok(mini)
    }

    /// resolves the current DID document without reading or replacing the
    /// cached mini document. callers that discard a raced result leave the
    /// cache untouched.
    pub async fn resolve_doc_fresh(&self, did: &Did) -> Result<MiniDoc, ResolverError> {
        Self::decode_mini_doc(did, &self.fetch_document(did).await?)
    }

    pub(crate) async fn cache_doc(&self, did: Did, doc: MiniDoc) {
        let _ = self.inner.cache.entry_async(did).await.put_entry(doc);
    }

    pub async fn resolve_identity_info(
        &self,
        did: &Did,
    ) -> Result<(Url, Option<Handle>), ResolverError> {
        let mini = self.resolve_doc(did).await?;
        Ok((mini.pds, mini.handle))
    }

    pub async fn resolve_signing_key(
        &self,
        did: &Did,
    ) -> Result<PublicKey<'static>, ResolverError> {
        let did = did.clone();
        let mini = self.resolve_doc(&did).await?;
        Ok(mini.key.ok_or(NoSigningKeyError(did)).into_diagnostic()?)
    }

    /// resolves the full DID document as raw [`Data`] without caching, and
    /// extracts the first `alsoKnownAs` handle if present. a document without a
    /// PDS service still answers here: only [`Self::resolve_doc`] needs one.
    pub async fn resolve_raw_doc(
        &self,
        did: &Did,
    ) -> Result<(Data, Option<Handle>), ResolverError> {
        let body = self.fetch_document(did).await?;
        let handle = first_handle(&parse_document(did, &body)?);
        let raw = serde_json::from_slice(&body).map_err(|error| {
            ResolverError::Generic(miette::miette!("failed to parse DID document: {error}"))
        })?;
        Ok((raw, handle))
    }

    async fn fetch_document(&self, did: &Did) -> Result<Vec<u8>, ResolverError> {
        if did.starts_with("did:web:") {
            let url = did_web_url(did)?;
            return self
                .get(&self.inner.documents, &url, MAX_DID_DOC_BYTES)
                .await;
        }
        if !did.starts_with("did:plc:") {
            return Err(ResolverError::Generic(miette::miette!(
                "unsupported DID method: {did}"
            )));
        }

        let sources = &self.inner.plc_urls;
        let start = self.inner.next_idx.fetch_add(1, Ordering::Relaxed) % sources.len();
        for offset in 0..sources.len() {
            let url = plc_url(&sources[(start + offset) % sources.len()], did);
            let result = self
                .get(&self.inner.documents, &url, MAX_DID_DOC_BYTES)
                .await;
            let retriable = matches!(
                result,
                Err(ResolverError::Ratelimited | ResolverError::Transport(_))
            );
            if !retriable || offset + 1 == sources.len() {
                return result;
            }
        }
        unreachable!("PLC source list is non-empty")
    }

    fn decode_mini_doc(did: &Did, body: &[u8]) -> Result<MiniDoc, ResolverError> {
        let doc = parse_document(did, body)?;
        let pds = doc
            .pds_endpoint()
            .ok_or_else(|| ResolverError::MissingPds(did.clone()))?;
        Ok(MiniDoc {
            pds: Url::parse(pds.as_str()).map_err(|error| {
                ResolverError::Generic(miette::miette!("invalid PDS endpoint for {did}: {error}"))
            })?,
            handle: first_handle(&doc),
            // an undecodable key is a missing key: the PDS endpoint still
            // works, and `resolve_signing_key` reports the absence itself.
            key: doc.atproto_public_key().ok().flatten(),
        })
    }

    async fn get(
        &self,
        client: &PublicHttpClient,
        url: &Url,
        max_body_bytes: usize,
    ) -> Result<Vec<u8>, ResolverError> {
        let request = http::Request::get(url.as_str())
            .body(Vec::new())
            .map_err(|error| {
                ResolverError::Generic(miette::miette!("invalid request URL {url}: {error}"))
            })?;
        let response = client
            .send_http_bounded(request, max_body_bytes)
            .await
            .map_err(|error| match error {
                // an oversized body is the source's answer, not an outage.
                PublicHttpError::BodyTooLarge { limit } => ResolverError::Generic(miette::miette!(
                    "identity source {url} response exceeds {limit} bytes"
                )),
                error => ResolverError::Transport(error.to_string().into()),
            })?;
        match response.status() {
            status if status.is_success() => Ok(response.into_body()),
            StatusCode::TOO_MANY_REQUESTS => Err(ResolverError::Ratelimited),
            // an outage at one mirror is not an answer about the identity, so
            // it stays retriable for the next source.
            status if status.is_server_error() => Err(ResolverError::Transport(
                format!("identity source {url} returned {status}").into(),
            )),
            status => Err(ResolverError::Generic(miette::miette!(
                "identity source {url} returned {status}"
            ))),
        }
    }

    async fn resolve_handle_with_timeout(
        &self,
        handle: &Handle,
        timeout: Duration,
    ) -> Result<Did, ResolverError> {
        tokio::time::timeout(timeout, self.resolve_handle_inner(handle))
            .await
            .map_err(|_| ResolverError::Transport("handle resolution timed out".into()))?
    }

    async fn resolve_handle_inner(&self, handle: &Handle) -> Result<Did, ResolverError> {
        if let Some(did) = self.inner.handle_dns.resolve(handle.clone()).await {
            return Ok(did);
        }
        let mut ratelimited = false;
        for source in &self.inner.handle_sources {
            match self.resolve_handle_source(source, handle).await {
                Ok(did) => return Ok(did),
                // a rate limit is our own fault, so it must never be reported
                // as an absent handle mapping.
                Err(ResolverError::Ratelimited) => ratelimited = true,
                Err(_) => {}
            }
        }
        if ratelimited {
            return Err(ResolverError::Ratelimited);
        }
        Err(ResolverError::HandleResolutionExhausted(miette::miette!(
            "failed to resolve handle {handle}"
        )))
    }

    async fn resolve_handle_source(
        &self,
        source: &HandleSource,
        handle: &Handle,
    ) -> Result<Did, ResolverError> {
        match source {
            HandleSource::HttpsWellKnown => {
                let url = Url::parse(&format!(
                    "https://{}/.well-known/atproto-did",
                    handle.as_str()
                ))
                .map_err(|error| {
                    ResolverError::Generic(miette::miette!("invalid handle host {handle}: {error}"))
                })?;
                let body = self
                    .get(&self.inner.handles, &url, MAX_HANDLE_BODY_BYTES)
                    .await?;
                let text = std::str::from_utf8(&body).map_err(|error| {
                    ResolverError::Generic(miette::miette!(
                        "invalid well-known body for {handle}: {error}"
                    ))
                })?;
                let did = text
                    .lines()
                    .find(|line| !line.trim().is_empty())
                    .ok_or_else(|| {
                        ResolverError::Generic(miette::miette!(
                            "empty well-known body for {handle}"
                        ))
                    })?;
                Did::new_owned(did.trim()).map_err(|error| {
                    ResolverError::Generic(miette::miette!(
                        "invalid well-known DID for {handle}: {error}"
                    ))
                })
            }
            HandleSource::Xrpc(base) => {
                let url = xrpc_url(
                    base,
                    "com.atproto.identity.resolveHandle",
                    "handle",
                    handle.as_str(),
                );
                let body = self
                    .get(&self.inner.handles, &url, MAX_HANDLE_BODY_BYTES)
                    .await?;
                serde_json::from_slice::<ResolveHandleOutput>(&body)
                    .map(|output| output.did)
                    .map_err(|error| {
                        ResolverError::Generic(miette::miette!(
                            "invalid resolveHandle response from {url}: {error}"
                        ))
                    })
            }
        }
    }

    /// returns `true` if `handle` resolves back to `did`.
    ///
    /// this is the reverse half of handle validity: the caller supplies the
    /// forward half by taking `handle` from that DID's own document.
    ///
    /// an exhausted, failed, or timed-out network lookup answers `false`, so
    /// callers can use `handle.invalid` (atproto.com/specs/handle#invalid-handles).
    /// only positive verdicts are stored, for [`HANDLE_VERIFY_TTL`]: a failure
    /// cannot distinguish a missing mapping from an outage, and a stored
    /// verdict belongs to one pair, so a renamed account ages out rather than
    /// being invalidated.
    pub async fn verify_handle(&self, did: &Did, handle: &Handle) -> Result<bool, ResolverError> {
        let key = (did.clone(), handle.clone());
        let fresh = self
            .inner
            .verified_handles
            .read_async(&key, |_, expires_at| *expires_at > Instant::now())
            .await;
        if fresh == Some(true) {
            return Ok(true);
        }

        let verdict = self
            .verify_handle_with_timeout(did, handle, HANDLE_VERIFY_TIMEOUT)
            .await?;
        if verdict {
            // concurrent verdicts of one pair can land in either order; both
            // were true, so the loser only re-dates the entry.
            let _ = self
                .inner
                .verified_handles
                .entry_async(key)
                .await
                .put_entry(Instant::now() + HANDLE_VERIFY_TTL);
        }
        Ok(verdict)
    }

    async fn verify_handle_with_timeout(
        &self,
        did: &Did,
        handle: &Handle,
        timeout: Duration,
    ) -> Result<bool, ResolverError> {
        Self::handle_verdict(did, self.resolve_handle_with_timeout(handle, timeout).await)
    }

    fn handle_verdict(
        did: &Did,
        result: Result<Did, ResolverError>,
    ) -> Result<bool, ResolverError> {
        match result {
            Ok(resolved_did) => Ok(resolved_did.as_str() == did.as_str()),
            Err(ResolverError::HandleResolutionExhausted(_) | ResolverError::Transport(_)) => {
                Ok(false)
            }
            Err(error) => Err(error),
        }
    }
}

/// parses a DID document and rejects one that answers for another DID.
fn parse_document(did: &Did, body: &[u8]) -> Result<DidDocument, ResolverError> {
    let doc: DidDocument = serde_json::from_slice(body).map_err(|error| {
        ResolverError::Generic(miette::miette!("failed to parse DID document: {error}"))
    })?;
    if doc.id.as_str() != did.as_str() {
        return Err(ResolverError::Generic(miette::miette!(
            "DID document id {} does not match requested {did}",
            doc.id
        )));
    }
    Ok(doc)
}

fn first_handle(doc: &DidDocument) -> Option<Handle> {
    let mut handles = doc.handles();
    handles
        .is_empty()
        .not()
        .then(|| handles.remove(0).into_static())
}

fn plc_url(base: &Url, did: &Did) -> Url {
    let mut url = base.clone();
    url.set_path(&format!(
        "{}/{}",
        base.path().trim_end_matches('/'),
        did.as_str()
    ));
    url.set_query(None);
    url.set_fragment(None);
    url
}

fn xrpc_url(base: &Url, nsid: &str, key: &str, value: &str) -> Url {
    let mut url = base.clone();
    let path = format!("{}/xrpc/{nsid}", base.path().trim_end_matches('/'));
    url.set_path(&path);
    url.set_query(None);
    url.set_fragment(None);
    url.query_pairs_mut().append_pair(key, value);
    url
}

fn did_web_url(did: &Did) -> Result<Url, ResolverError> {
    let rest = did
        .as_str()
        .strip_prefix("did:web:")
        .ok_or_else(|| ResolverError::Generic(miette::miette!("unsupported DID method: {did}")))?;
    let mut parts = rest.split(':');
    let host = parts
        .next()
        .filter(|host| !host.is_empty())
        .ok_or_else(|| ResolverError::Generic(miette::miette!("invalid did:web: {did}")))?
        // a port is percent-encoded in the method-specific id
        // (w3c-ccg.github.io/did-method-web/#did-method-operations).
        .replace("%3A", ":")
        .replace("%3a", ":");
    let segments = parts.collect::<Vec<_>>();
    let mut url = Url::parse(&format!("https://{host}/")).map_err(|error| {
        ResolverError::Generic(miette::miette!("invalid did:web host in {did}: {error}"))
    })?;
    if segments.is_empty() {
        url.set_path("/.well-known/did.json");
    } else {
        url.set_path(&format!("/{}/did.json", segments.join("/")));
    }
    Ok(url)
}

#[derive(Debug, Diagnostic, Error)]
#[error("no atproto signing key in DID doc for {0}")]
pub struct NoSigningKeyError(Did);

#[cfg(test)]
mod tests {
    use super::*;
    use axum::{Router, http::header, response::IntoResponse};
    use std::sync::Arc;
    use tokio::sync::RwLock;

    struct HandleStub {
        addr: std::net::SocketAddr,
        resolve_calls: Arc<std::sync::atomic::AtomicUsize>,
    }

    struct FixedHandleDns {
        did: Option<Did>,
        calls: Arc<std::sync::atomic::AtomicUsize>,
    }

    impl HandleDns for FixedHandleDns {
        fn resolve(&self, _handle: Handle) -> BoxFuture<'static, Option<Did>> {
            self.calls.fetch_add(1, Ordering::SeqCst);
            let did = self.did.clone();
            Box::pin(async move { did })
        }
    }

    struct PendingHandleDns;

    impl HandleDns for PendingHandleDns {
        fn resolve(&self, _handle: Handle) -> BoxFuture<'static, Option<Did>> {
            Box::pin(std::future::pending())
        }
    }

    /// serves the XRPC `resolveHandle` endpoint over http.
    async fn spawn_handle_stub(did: Did) -> HandleStub {
        use axum::extract::State;
        use std::sync::atomic::AtomicUsize;

        #[derive(Clone)]
        struct Counters {
            did: Did,
            resolve_calls: Arc<AtomicUsize>,
        }

        async fn resolve_handle(State(counters): State<Counters>) -> axum::Json<serde_json::Value> {
            counters
                .resolve_calls
                .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            axum::Json(serde_json::json!({ "did": counters.did.as_str() }))
        }

        async fn not_found() -> (axum::http::StatusCode, &'static str) {
            (axum::http::StatusCode::NOT_FOUND, "no well-known here")
        }

        let counters = Counters {
            did,
            resolve_calls: Arc::new(AtomicUsize::new(0)),
        };
        let app = Router::new()
            .route(
                "/xrpc/com.atproto.identity.resolveHandle",
                axum::routing::get(resolve_handle),
            )
            .fallback(not_found)
            .with_state(counters.clone());
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        tokio::spawn(async move {
            axum::serve(listener, app).await.unwrap();
        });
        HandleStub {
            addr,
            resolve_calls: counters.resolve_calls,
        }
    }

    /// accepts connections and never answers them: a dropped connect is not
    /// reproducible locally, and both end at the same transport budget.
    async fn spawn_silent_listener() -> (std::net::SocketAddr, Arc<std::sync::atomic::AtomicUsize>)
    {
        use std::sync::atomic::AtomicUsize;

        let accepted = Arc::new(AtomicUsize::new(0));
        let counter = accepted.clone();
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        tokio::spawn(async move {
            let mut held = Vec::new();
            while let Ok((stream, _)) = listener.accept().await {
                counter.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
                held.push(stream);
            }
        });
        (addr, accepted)
    }

    /// the well-known step is always https, so only the http resolveHandle step
    /// can be served by a local stub.
    fn stub_resolver(public: PublicHttpClient, handle_dns: Arc<dyn HandleDns>) -> Resolver {
        let base = Url::parse(HANDLE_BASE).unwrap();
        Resolver::with_clients(
            vec![base.clone()],
            vec![HandleSource::HttpsWellKnown, HandleSource::Xrpc(base)],
            10,
            public.clone(),
            public,
            handle_dns,
        )
    }

    /// stub hosts resolve locally, with the handle transport's budget shape.
    fn stub_client(
        base: std::net::SocketAddr,
        hosts: &[(&str, std::net::SocketAddr)],
    ) -> PublicHttpClient {
        let mut builder = reqwest::Client::builder()
            .no_proxy()
            .resolve("public.example", base)
            .timeout(Duration::from_millis(500))
            .connect_timeout(Duration::from_millis(500));
        for &(host, addr) in hosts {
            builder = builder.resolve(host, addr);
        }
        PublicHttpClient::test(builder.build().unwrap())
    }

    const HANDLE_BASE: &str = "http://public.example/";

    #[test]
    fn split_dns_txt_data_is_joined_before_parsing() {
        let txt = TXT::from_bytes(vec![b"did=did:plc:", b"ewvi7nxzyoun6zhxrhs64oiz"]);
        assert_eq!(
            did_from_txt(&txt).unwrap().as_str(),
            "did:plc:ewvi7nxzyoun6zhxrhs64oiz"
        );
        assert!(did_from_txt(&TXT::new(vec!["not-a-did".into()])).is_none());
    }

    #[tokio::test]
    async fn dns_txt_answer_precedes_network_fallbacks() {
        use std::sync::atomic::Ordering;

        let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap();
        let handle = Handle::new_static("fast.test").unwrap();
        let stub = spawn_handle_stub(did.clone()).await;
        let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let public = stub_client(stub.addr, &[("fast.test", stub.addr)]);
        let resolver = stub_resolver(
            public,
            Arc::new(FixedHandleDns {
                did: Some(did.clone()),
                calls: calls.clone(),
            }),
        );

        assert!(resolver.verify_handle(&did, &handle).await.unwrap());
        assert_eq!(calls.load(Ordering::SeqCst), 1);
        assert_eq!(
            stub.resolve_calls.load(Ordering::SeqCst),
            0,
            "a DNS answer must prevent all network fallbacks"
        );
    }

    #[tokio::test]
    async fn the_whole_verification_has_one_budget() {
        let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap();
        let handle = Handle::new_static("slow.test").unwrap();
        let stub = spawn_handle_stub(did.clone()).await;
        let public = stub_client(stub.addr, &[]);
        let resolver = stub_resolver(public, Arc::new(PendingHandleDns));

        let verdict = tokio::time::timeout(
            Duration::from_secs(1),
            resolver.verify_handle_with_timeout(&did, &handle, Duration::from_millis(20)),
        )
        .await
        .expect("the total verification budget must terminate a stalled source")
        .unwrap();

        assert!(!verdict);
    }

    #[tokio::test]
    async fn a_handle_host_that_never_answers_is_bounded() {
        use std::sync::atomic::Ordering;

        let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap();
        let handle = Handle::new_static("slow.test").unwrap();
        let stub = spawn_handle_stub(did.clone()).await;
        let (silent, silent_accepts) = spawn_silent_listener().await;
        let resolver = stub_resolver(
            stub_client(stub.addr, &[("slow.test", silent)]),
            Arc::new(NoHandleDns),
        );

        let started = Instant::now();
        let verdict = tokio::time::timeout(
            Duration::from_secs(5),
            resolver.verify_handle(&did, &handle),
        )
        .await
        .expect("a handle host that never answers must not hold the verdict")
        .unwrap();

        assert!(verdict);
        assert_eq!(
            silent_accepts.load(Ordering::SeqCst),
            1,
            "the account's host is still asked, in spec order"
        );
        assert_eq!(
            stub.resolve_calls.load(Ordering::SeqCst),
            1,
            "the bounded attempt must fall through to the next step"
        );
        assert!(
            started.elapsed() < Duration::from_secs(2),
            "the attempt must be bounded, took {:?}",
            started.elapsed()
        );
    }

    #[tokio::test]
    async fn only_answered_verdicts_are_reused() {
        use std::sync::atomic::Ordering;

        let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap();
        let other = Did::new_static("did:plc:j5hmlfdrwkvtxm7cjmu7j2is").unwrap();
        let handle = Handle::new_static("fast.test").unwrap();
        let stub = spawn_handle_stub(did.clone()).await;
        // fast.test's https step fails the handshake, so resolveHandle answers.
        let resolver = stub_resolver(
            stub_client(stub.addr, &[("fast.test", stub.addr)]),
            Arc::new(NoHandleDns),
        );

        assert!(resolver.verify_handle(&did, &handle).await.unwrap());
        assert_eq!(stub.resolve_calls.load(Ordering::SeqCst), 1);

        assert!(resolver.verify_handle(&did, &handle).await.unwrap());
        assert_eq!(
            stub.resolve_calls.load(Ordering::SeqCst),
            1,
            "a successful verdict is reused"
        );

        assert!(!resolver.verify_handle(&other, &handle).await.unwrap());
        assert_eq!(
            stub.resolve_calls.load(Ordering::SeqCst),
            2,
            "a verdict belongs to one did and handle pair"
        );
        assert!(!resolver.verify_handle(&other, &handle).await.unwrap());
        assert_eq!(
            stub.resolve_calls.load(Ordering::SeqCst),
            3,
            "a negative verdict is never remembered"
        );
    }

    async fn spawn_status_server(status: StatusCode) -> std::net::SocketAddr {
        let app = Router::new().fallback(move || async move { status });
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        tokio::spawn(async move {
            axum::serve(listener, app).await.unwrap();
        });
        addr
    }

    /// resolves handles through one http XRPC source only.
    fn xrpc_handle_resolver(address: std::net::SocketAddr) -> Resolver {
        let base = Url::parse(HANDLE_BASE).unwrap();
        let public = test_public_client(address);
        Resolver::with_clients(
            vec![base.clone()],
            vec![HandleSource::Xrpc(base)],
            10,
            public.clone(),
            public,
            Arc::new(NoHandleDns),
        )
    }

    #[tokio::test]
    async fn a_rate_limited_document_source_stays_rate_limited() {
        let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap();
        let resolver = plc_resolver(spawn_status_server(StatusCode::TOO_MANY_REQUESTS).await);

        assert!(matches!(
            resolver.resolve_doc(&did).await.unwrap_err(),
            ResolverError::Ratelimited
        ));
    }

    #[tokio::test]
    async fn a_rate_limited_handle_source_is_not_an_invalid_handle() {
        let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap();
        let handle = Handle::new_static("fast.test").unwrap();
        let resolver =
            xrpc_handle_resolver(spawn_status_server(StatusCode::TOO_MANY_REQUESTS).await);

        assert!(matches!(
            resolver.verify_handle(&did, &handle).await.unwrap_err(),
            ResolverError::Ratelimited
        ));
    }

    #[tokio::test]
    async fn a_handle_no_source_knows_is_unverified() {
        let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap();
        let handle = Handle::new_static("fast.test").unwrap();
        let resolver = xrpc_handle_resolver(spawn_status_server(StatusCode::NOT_FOUND).await);

        assert!(
            !resolver.verify_handle(&did, &handle).await.unwrap(),
            "an answer of 'no such handle' is an unverified handle, not an error"
        );
    }

    #[tokio::test]
    async fn the_xrpc_source_answers_with_the_did_it_encodes() {
        let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap();
        let handle = Handle::new_static("fast.test").unwrap();
        let stub = spawn_handle_stub(did.clone()).await;
        let resolver = xrpc_handle_resolver(stub.addr);

        assert_eq!(
            resolver
                .resolve_did(&AtIdentifier::Handle(handle))
                .await
                .unwrap(),
            did
        );
    }

    async fn spawn_mutable_doc_server(body: String) -> (std::net::SocketAddr, Arc<RwLock<String>>) {
        let current = Arc::new(RwLock::new(body));
        let served = current.clone();
        let app = Router::new().fallback(move || {
            let served = served.clone();
            async move {
                let body = served.read().await.clone();
                ([(header::CONTENT_TYPE, "application/json")], body).into_response()
            }
        });
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        tokio::spawn(async move {
            axum::serve(listener, app).await.unwrap();
        });
        (addr, current)
    }

    async fn spawn_doc_server(body: String) -> std::net::SocketAddr {
        spawn_mutable_doc_server(body).await.0
    }

    fn test_public_client(address: std::net::SocketAddr) -> PublicHttpClient {
        let client = reqwest::Client::builder()
            .no_proxy()
            .resolve("public.example", address)
            .build()
            .unwrap();
        PublicHttpClient::test(client)
    }

    fn plc_resolver(address: std::net::SocketAddr) -> Resolver {
        Resolver::with_plc_urls_and_client(
            vec![Url::parse("http://public.example/").unwrap()],
            10,
            test_public_client(address),
        )
    }

    fn did_doc_json(did: &Did, extra: serde_json::Value) -> String {
        let mut doc = serde_json::json!({
            "@context": ["https://www.w3.org/ns/did/v1"],
            "id": did.as_str(),
            "alsoKnownAs": ["at://alice.test"],
        });
        let object = doc.as_object_mut().unwrap();
        for (key, value) in extra.as_object().unwrap() {
            object.insert(key.clone(), value.clone());
        }
        doc.to_string()
    }

    fn pds_service(endpoint: &str) -> serde_json::Value {
        serde_json::json!({
            "service": [{
                "id": "#atproto_pds",
                "type": "AtprotoPersonalDataServer",
                "serviceEndpoint": endpoint
            }]
        })
    }

    #[test]
    fn did_web_urls_follow_the_method_spec() {
        let url = |did: &str| {
            did_web_url(&Did::new_owned(did).unwrap())
                .unwrap()
                .to_string()
        };

        assert_eq!(
            url("did:web:example.com"),
            "https://example.com/.well-known/did.json"
        );
        assert_eq!(
            url("did:web:example.com:user:alice"),
            "https://example.com/user/alice/did.json"
        );
        assert_eq!(
            url("did:web:example.com%3A3000"),
            "https://example.com:3000/.well-known/did.json",
            "a port is percent-encoded in the method-specific id"
        );
    }

    #[tokio::test]
    async fn a_document_answering_for_another_did_is_rejected() {
        let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap();
        let other = Did::new_static("did:plc:j5hmlfdrwkvtxm7cjmu7j2is").unwrap();
        let body = did_doc_json(&other, pds_service("https://pds.example"));
        let resolver = plc_resolver(spawn_doc_server(body).await);

        let error = resolver.resolve_doc(&did).await.unwrap_err();
        assert!(
            error.to_string().contains("does not match requested"),
            "unexpected error: {error}"
        );
    }

    #[tokio::test]
    async fn a_document_without_a_pds_still_yields_its_raw_form() {
        let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap();
        let resolver =
            plc_resolver(spawn_doc_server(did_doc_json(&did, serde_json::json!({}))).await);

        assert!(matches!(
            resolver.resolve_doc(&did).await.unwrap_err(),
            ResolverError::MissingPds(_)
        ));

        let (raw, handle) = resolver
            .resolve_raw_doc(&did)
            .await
            .expect("a deactivated account still has a document");
        assert_eq!(serde_json::to_value(raw).unwrap()["id"], did.as_str());
        assert_eq!(handle.unwrap().as_str(), "alice.test");
    }

    #[tokio::test]
    async fn an_undecodable_signing_key_is_a_missing_key() {
        let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap();
        let mut extra = pds_service("https://pds.example");
        extra.as_object_mut().unwrap().insert(
            "verificationMethod".into(),
            serde_json::json!([{
                "id": format!("{}#atproto", did.as_str()),
                "type": "Multikey",
                "controller": did.as_str(),
                "publicKeyMultibase": "zNotAKey"
            }]),
        );
        let resolver = plc_resolver(spawn_doc_server(did_doc_json(&did, extra)).await);

        let mini = resolver
            .resolve_doc(&did)
            .await
            .expect("a broken key must not hide the PDS endpoint");
        assert_eq!(mini.pds.as_str(), "https://pds.example/");
        assert!(mini.key.is_none());
        assert!(
            resolver
                .resolve_signing_key(&did)
                .await
                .unwrap_err()
                .to_string()
                .contains("no atproto signing key")
        );
    }

    #[tokio::test]
    async fn a_failing_mirror_rolls_to_the_next_plc_source() {
        let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap();
        let down = spawn_status_server(StatusCode::BAD_GATEWAY).await;
        let up = spawn_doc_server(did_doc_json(&did, pds_service("https://pds.example"))).await;
        let client = reqwest::Client::builder()
            .no_proxy()
            .resolve("down.example", down)
            .resolve("up.example", up)
            .build()
            .unwrap();
        let resolver = Resolver::with_plc_urls_and_client(
            vec![
                Url::parse("http://down.example/").unwrap(),
                Url::parse("http://up.example/").unwrap(),
            ],
            10,
            PublicHttpClient::test(client),
        );

        assert_eq!(
            resolver.resolve_doc(&did).await.unwrap().pds.as_str(),
            "https://pds.example/"
        );
    }

    #[tokio::test]
    async fn fresh_resolution_bypasses_cache_until_explicitly_installed() {
        let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap();
        let doc = |pds: &str| {
            serde_json::json!({
                "@context": ["https://www.w3.org/ns/did/v1"],
                "id": did.as_str(),
                "alsoKnownAs": ["at://alice.test"],
                "service": [{
                    "id": "#atproto_pds",
                    "type": "AtprotoPersonalDataServer",
                    "serviceEndpoint": pds
                }],
                "verificationMethod": []
            })
            .to_string()
        };
        let (address, current) = spawn_mutable_doc_server(doc("https://pds-a.example")).await;
        let resolver = plc_resolver(address);

        assert_eq!(
            resolver.resolve_doc(&did).await.unwrap().pds.as_str(),
            "https://pds-a.example/"
        );
        *current.write().await = doc("https://pds-b.example");

        let fresh = resolver.resolve_doc_fresh(&did).await.unwrap();
        assert_eq!(fresh.pds.as_str(), "https://pds-b.example/");
        assert_eq!(
            resolver.resolve_doc(&did).await.unwrap().pds.as_str(),
            "https://pds-a.example/"
        );

        resolver.cache_doc(did.clone(), fresh).await;
        assert_eq!(
            resolver.resolve_doc(&did).await.unwrap().pds.as_str(),
            "https://pds-b.example/"
        );
    }

    #[tokio::test]
    async fn raw_did_doc_json_contract_and_size_limit() {
        let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap();
        let body = serde_json::json!({
            "@context": ["https://www.w3.org/ns/did/v1"],
            "id": did.as_str(),
            "alsoKnownAs": ["at://alice.test"],
            "service": [{
                "id": "#atproto_pds",
                "type": "AtprotoPersonalDataServer",
                "serviceEndpoint": "https://pds.example"
            }],
            "verificationMethod": []
        })
        .to_string();
        let address = spawn_doc_server(body).await;
        let resolver = plc_resolver(address);

        let (data, handle) = resolver.resolve_raw_doc(&did).await.unwrap();
        let json = serde_json::to_value(data).unwrap();
        assert_eq!(json["id"], did.as_str());
        assert_eq!(json["service"][0]["serviceEndpoint"], "https://pds.example");
        assert_eq!(handle.unwrap().as_str(), "alice.test");

        let oversized = plc_resolver(spawn_doc_server("x".repeat(MAX_DID_DOC_BYTES + 1)).await);
        let error = oversized.resolve_raw_doc(&did).await.unwrap_err();
        assert!(
            error
                .to_string()
                .contains(&format!("exceeds {MAX_DID_DOC_BYTES} bytes")),
            "unexpected error: {error}"
        );
    }
}