//! The host `call` (javascript/src/server hostCall + login, didkey didkeyCall) and loadModule/pdsOf
//! from run-module: own PDS, atproto-proxy (signed out: a query straight to the did#id), https service
//! with getServiceAuth, DID-service reads,
//! subscriptions, attest and did:key pseudo-calls. Transports (iroh) are crates on crate::transport.

use std::sync::{Arc, Mutex};

use serde_json::{Map, Value, json};

use crate::crypto::{self, ATTEST_NSID, DidKey};
use crate::run::{Call, Module, Request, Sub};
use crate::store::Store;
use crate::{BoxFut, iso, js_string, transport, ws};

const MODULE_COLLECTION: &str = "com.fedfork.exec.wasm.module";

pub struct Response {
    pub status: u16,
    pub headers: Vec<(String, String)>,
    pub body: Vec<u8>,
}

impl Response {
    fn header(&self, name: &str) -> Option<&str> {
        self.headers
            .iter()
            .find(|(k, _)| k.eq_ignore_ascii_case(name))
            .map(|(_, v)| v.as_str())
    }
}

/// ctx.http.fetch: swappable so tests and the conformance bridge run without a network.
pub trait Fetch: Send + Sync {
    fn fetch(
        &self,
        method: &str,
        url: &str,
        headers: Vec<(String, String)>,
        body: Option<Vec<u8>>,
    ) -> BoxFut<'_, Result<Response, String>>;
}

impl Fetch for reqwest::Client {
    fn fetch(
        &self,
        method: &str,
        url: &str,
        headers: Vec<(String, String)>,
        body: Option<Vec<u8>>,
    ) -> BoxFut<'_, Result<Response, String>> {
        let mut request = self.request(method.parse().unwrap_or(reqwest::Method::GET), url);
        for (k, v) in headers {
            request = request.header(k, v);
        }
        if let Some(body) = body {
            request = request.body(body);
        }
        Box::pin(async move {
            let response = request
                .send()
                .await
                .map_err(|e| format!("fetch failed: {e}"))?;
            let status = response.status().as_u16();
            let headers = response
                .headers()
                .iter()
                .map(|(k, v)| (k.to_string(), v.to_str().unwrap_or_default().to_string()))
                .collect();
            let body = response.bytes().await.map_err(|e| e.to_string())?.to_vec();
            Ok(Response {
                status,
                headers,
                body,
            })
        })
    }
}

pub struct XrpcError {
    pub message: String,
    pub error: Option<String>,
}

/// text/event-stream frames as [{ event, data }] ("message" when a frame names none), data parsed when it is
/// JSON: host-call.js sseEvents.
pub fn sse_events(text: &str) -> Value {
    let text = text.replace("\r\n", "\n");
    let events = text.split("\n\n").filter_map(|frame| {
        let mut event = "message".to_string();
        let mut data = Vec::new();
        for line in frame.lines() {
            if let Some(name) = line.strip_prefix("event:") {
                event = name.trim().to_string();
            } else if let Some(d) = line.strip_prefix("data:") {
                data.push(d.strip_prefix(' ').unwrap_or(d));
            }
        }
        (!data.is_empty()).then(|| {
            let raw = data.join("\n");
            json!({ "event": event, "data": serde_json::from_str::<Value>(&raw).unwrap_or(Value::String(raw)) })
        })
    });
    Value::Array(events.collect())
}

/// The JSON body of an XRPC response, or its error: "<nsid> failed with <error|status>[: message]".
/// A text/event-stream answer comes back as `{ events: sse_events(body) }`, a binary one (a
/// content-type neither JSON nor text: getBlob, a CAR) as `{ $bytes }` (JS json).
pub fn json_response(nsid: &str, response: &Response) -> Result<Value, XrpcError> {
    let ok = (200..300).contains(&response.status);
    let content_type = response.header("content-type").unwrap_or_default().to_ascii_lowercase();
    if ok && content_type.starts_with("text/event-stream") {
        return Ok(json!({ "events": sse_events(&String::from_utf8_lossy(&response.body)) }));
    }
    if ok && !content_type.is_empty() && !content_type.contains("json") && !content_type.starts_with("text/") {
        return Ok(json!({ "$bytes": base64::Engine::encode(&base64::engine::general_purpose::STANDARD_NO_PAD, &response.body) }));
    }
    let body: Value = serde_json::from_slice(&response.body).unwrap_or_else(|_| json!({}));
    if (200..300).contains(&response.status) {
        return Ok(body);
    }
    let error = body
        .get("error")
        .filter(|e| !e.is_null())
        .map(|e| js_string(Some(e)));
    let truthy = |v: &&Value| match v {
        Value::Null => false,
        Value::Bool(b) => *b,
        Value::Number(n) => n.as_f64() != Some(0.0),
        Value::String(s) => !s.is_empty(),
        _ => true,
    };
    let message = body
        .get("message")
        .filter(truthy)
        .map(|m| format!(": {}", js_string(Some(m))))
        .unwrap_or_default();
    // A 429 says when it lifts (ratelimit-reset: unix seconds), so nobody has to probe for it.
    let until = match (
        response.status,
        response.header("ratelimit-reset").filter(|r| !r.is_empty()),
    ) {
        (429, Some(reset)) => format!(
            " ({}; resets {})",
            response.header("ratelimit-policy").unwrap_or("limit"),
            iso(reset.trim().parse::<u64>().unwrap_or(0) * 1000)
        ),
        _ => String::new(),
    };
    Err(XrpcError {
        message: format!(
            "{nsid} failed with {}{message}{until}",
            error.clone().unwrap_or(response.status.to_string())
        ),
        error,
    })
}

pub fn xrpc_url(
    base: &str,
    nsid: &str,
    params: Option<&Map<String, Value>>,
) -> Result<String, String> {
    let mut url = reqwest::Url::parse(base)
        .and_then(|b| b.join(&format!("/xrpc/{nsid}")))
        .map_err(|e| format!("Invalid URL {base}: {e}"))?;
    if let Some(params) = params {
        url.query_pairs_mut()
            .clear()
            .extend_pairs(params.iter().map(|(k, v)| (k, js_string(Some(v)))));
    }
    Ok(url.to_string())
}

/// JS xrpcBody: a request body and its content-type. `{ $bytes }` input goes as those bytes (`encoding`,
/// default application/octet-stream: uploadBlob), any other input as JSON.
pub fn xrpc_body(input: Option<&Value>, encoding: Option<&str>) -> Result<Option<(Vec<u8>, String)>, String> {
    let Some(input) = input else { return Ok(None) };
    match input.as_object().filter(|o| o.len() == 1).and_then(|o| o.get("$bytes")).and_then(Value::as_str) {
        Some(b64) => {
            let bytes = base64::Engine::decode(&base64::engine::general_purpose::STANDARD_NO_PAD, b64.trim_end_matches('=')).map_err(|e| format!("$bytes is not base64: {e}"))?;
            Ok(Some((bytes, encoding.unwrap_or("application/octet-stream").to_string())))
        }
        None => Ok(Some((input.to_string().into_bytes(), "application/json".into()))),
    }
}

async fn get_json(fetch: &dyn Fetch, nsid: &str, url: &str) -> Result<Value, String> {
    json_response(nsid, &fetch.fetch("GET", url, vec![], None).await?).map_err(|e| e.message)
}

/// JS privateHost: loopback, private, link-local, shared, multicast and unspecified hosts
/// (localhost and IPv4 / IPv6 literals). ponytail: literals only, as the JS host; a public name
/// that resolves to one is not caught (needs a resolver hook).
pub fn private_host(host: &url::Host<&str>) -> bool {
    match host {
        url::Host::Domain(d) => {
            let d = d.trim_end_matches('.').to_ascii_lowercase();
            d.is_empty() || d == "localhost" || d.ends_with(".localhost")
        }
        url::Host::Ipv4(ip) => {
            let [a, b, ..] = ip.octets();
            a == 0
                || a == 10
                || a == 127
                || a >= 224
                || (a == 169 && b == 254)
                || (a == 172 && (16..=31).contains(&b))
                || (a == 192 && b == 168)
                || (a == 100 && (64..=127).contains(&b))
        }
        url::Host::Ipv6(ip) => {
            let first = ip.segments()[0];
            first == 0
                || first & 0xfe00 == 0xfc00
                || first & 0xffc0 == 0xfe80
                || first & 0xff00 == 0xff00
        }
    }
}

/// ATLIBS_ALLOW_PRIVATE=1: DID lookups may reach http and private addresses (local dev, tests).
fn allow_private() -> bool {
    std::env::var("ATLIBS_ALLOW_PRIVATE").is_ok_and(|v| v == "1")
}

/// JS publicUrl: a DID document names its own endpoints, and DID reads run for anonymous visitors
/// (ingress), so a lookup must not reach the host's own network: https, and not a private host.
pub fn public_url(url: &str, allow_private: bool) -> Result<String, String> {
    let parsed = url::Url::parse(url).map_err(|_| format!("{url} is not a URL"))?;
    if allow_private {
        return Ok(url.to_string());
    }
    if parsed.scheme() != "https" {
        return Err(format!("{url} is not https"));
    }
    if parsed.host().is_none_or(|h| private_host(&h)) {
        return Err(format!("{url} is a private address"));
    }
    Ok(url.to_string())
}

/// The PDS a did:plc or did:web account lives on, or for did#id that service's endpoint: an https URL
/// off the host's network unless ATLIBS_ALLOW_PRIVATE=1.
pub async fn pds_of(fetch: &dyn Fetch, reference: &str) -> Result<String, String> {
    let (did, id) = reference.split_once('#').unwrap_or((reference, "atproto_pds"));
    let url = if did.starts_with("did:plc:") {
        format!("https://plc.directory/{did}")
    } else if let Some(host) = did.strip_prefix("did:web:") {
        format!(
            "https://{}/.well-known/did.json",
            host.replace("%3A", ":").replace("%3a", ":")
        )
    } else {
        return Err(format!("{did} is not a did:plc or did:web"));
    };
    let response = fetch
        .fetch("GET", &public_url(&url, allow_private())?, vec![], None)
        .await?;
    if !(200..300).contains(&response.status) {
        return Err(format!("Could not resolve {did}"));
    }
    let doc: Value = serde_json::from_slice(&response.body).map_err(|e| e.to_string())?;
    let missing = || if id == "atproto_pds" { format!("{did} has no PDS") } else { format!("{did} has no #{id} service") };
    public_url(&service_in(&doc, did, id).ok_or_else(missing)?, allow_private())
}

fn service_in(doc: &Value, did: &str, id: &str) -> Option<String> {
    let services = doc.get("service")?.as_array()?;
    let found = services.iter().find(|s| {
        s.get("id")
            .and_then(Value::as_str)
            .is_some_and(|x| x == format!("#{id}") || x == format!("{did}#{id}"))
    })?;
    found
        .get("serviceEndpoint")
        .and_then(Value::as_str)
        .map(String::from)
}

/// A blob's CID (raw codec, sha2-256, base32 CIDv1) carries the sha256 its bytes must hash to.
pub fn blob_sha256(cid: Option<&str>) -> Result<String, String> {
    let cid = cid
        .filter(|c| {
            c.starts_with('b')
                && c[1..]
                    .bytes()
                    .all(|b| b"abcdefghijklmnopqrstuvwxyz234567".contains(&b))
        })
        .ok_or("Blob CID must be base32 CIDv1")?;
    let bytes = crypto::unbase32(&cid[1..]);
    if bytes.len() != 36 || bytes[..4] != [0x01, 0x55, 0x12, 0x20] {
        return Err("Blob CID must be a raw sha2-256 CIDv1".into());
    }
    Ok(crypto::hex(&bytes[4..]))
}

/// A published module: its com.fedfork.exec.wasm.module record and blob, from the publisher's PDS.
pub async fn load_module(fetch: &dyn Fetch, uri: &str) -> Result<Module, String> {
    let parts: Vec<&str> = uri
        .strip_prefix("at://")
        .map(|r| r.split('/').collect())
        .unwrap_or_default();
    let [did, collection, rkey] = parts[..] else {
        return Err(format!("{uri} is not a {MODULE_COLLECTION} record"));
    };
    if collection != MODULE_COLLECTION || !did.starts_with("did:") || rkey.is_empty() {
        return Err(format!("{uri} is not a {MODULE_COLLECTION} record"));
    }
    let pds = pds_of(fetch, did).await?;
    let params = |pairs: &[(&str, &str)]| {
        pairs
            .iter()
            .map(|(k, v)| (k.to_string(), json!(v)))
            .collect::<Map<_, _>>()
    };
    let record = fetch
        .fetch(
            "GET",
            &xrpc_url(
                &pds,
                "com.atproto.repo.getRecord",
                Some(&params(&[
                    ("repo", did),
                    ("collection", collection),
                    ("rkey", rkey),
                ])),
            )?,
            vec![],
            None,
        )
        .await?;
    if !(200..300).contains(&record.status) {
        return Err(format!("{uri} not found"));
    }
    // any JSON: a reply that is not an object (or has no value) is an unsupported ABI, not a panic
    let value = serde_json::from_slice::<Value>(&record.body)
        .map_err(|e| e.to_string())?
        .get_mut("value")
        .map(Value::take)
        .unwrap_or(Value::Null);
    if value.get("abi").and_then(Value::as_str) != Some("emdash-wasm-json-v1") {
        return Err(format!(
            "Unsupported module ABI {}",
            js_string(value.get("abi"))
        ));
    }
    let cid = value.pointer("/module/ref/$link").and_then(Value::as_str);
    let sha256 = blob_sha256(cid)?;
    let blob = fetch
        .fetch(
            "GET",
            &xrpc_url(
                &pds,
                "com.atproto.sync.getBlob",
                Some(&params(&[("did", did), ("cid", cid.unwrap_or_default())])),
            )?,
            vec![],
            None,
        )
        .await?;
    if !(200..300).contains(&blob.status) {
        return Err(format!("{uri} blob not found"));
    }
    let mut module = Module::from_manifest(blob.body, &value)?;
    module.sha256 = Some(sha256);
    module.id = Some(uri.to_string());
    Ok(module)
}

/// App-password session: createSession at `service`, then XRPC to the account's own PDS. With a
/// store, the session is the state dir's saved one (sealed setting "session"; createSession is rate
/// limited) and every new one is saved there. Runs share it, in this process and across processes
/// sharing the state dir: when a request comes back ExpiredToken or InvalidToken, one renewal runs at a
/// time here (`renewing`) and under the store's "session" lease across processes; it first re-reads the
/// saved session (another run may have refreshed it already: use that), else refreshes with the newest
/// refresh token, and logs in again only if the refresh is refused. Without a store it lives in memory.
pub struct Session {
    identifier: String,
    password: Option<String>,
    service: String,
    fetch: Arc<dyn Fetch>,
    tokens: Mutex<Value>,
    store: Option<Store>,
    renewing: tokio::sync::Mutex<()>,
    /// an OAuth session (crate::oauth_client): `identifier` is its DID, `tokens` its saved session
    /// (DPoP-bound, refreshed at its authorization server, never re-created)
    oauth: bool,
}

/// Who a session signs in as: an explicit choice, never read from the environment.
/// No config, no session.
#[derive(Clone, Debug)]
pub enum SessionConfig {
    /// The simple / testing case: an app password at `service` (an entryway or PDS). The password may
    /// be None when the state dir holds this identifier's saved session.
    AppPassword { identifier: String, password: Option<String>, service: String },
    /// A user's OAuth session, saved in the state dir by `atlibs login` or a server's /oauth/callback.
    OAuth { did: String },
}

impl SessionConfig {
    pub async fn open(self, fetch: Arc<dyn Fetch>, store: Option<Store>) -> Result<Arc<Session>, String> {
        match self {
            SessionConfig::AppPassword { identifier, password, service } => Session::login(&identifier, password, &service, fetch, store).await,
            SessionConfig::OAuth { did } => Session::oauth(&did, fetch, store.ok_or("an OAuth session lives in a state dir")?).await,
        }
    }
}

/// How long a process may hold the session lease (a refresh, then maybe a createSession) before others take over.
const SESSION_LEASE: std::time::Duration = std::time::Duration::from_secs(30);

impl Session {
    pub async fn login(
        identifier: &str,
        password: Option<String>,
        service: &str,
        fetch: Arc<dyn Fetch>,
        store: Option<Store>,
    ) -> Result<Arc<Self>, String> {
        let session = Arc::new(Self {
            identifier: identifier.into(),
            password,
            service: service.into(),
            fetch,
            tokens: Mutex::new(Value::Null),
            store,
            renewing: tokio::sync::Mutex::new(()),
            oauth: false,
        });
        if !session.adopt_saved()? {
            // Several processes starting on an empty state dir log in once.
            let _lease = session.lease().await?;
            if !session.adopt_saved()? {
                session.create().await?;
            }
        }
        Ok(session)
    }

    /// `did`'s OAuth session, saved in the state dir (oauth_client: `atlibs login`, a server's callback).
    pub async fn oauth(did: &str, fetch: Arc<dyn Fetch>, store: Store) -> Result<Arc<Self>, String> {
        let session = Arc::new(Self {
            identifier: did.into(),
            password: None,
            service: String::new(),
            fetch,
            tokens: Mutex::new(Value::Null),
            store: Some(store),
            renewing: tokio::sync::Mutex::new(()),
            oauth: true,
        });
        if !session.adopt_saved()? {
            return Err(format!("{did} has no OAuth session here: sign in (atlibs login, or the server's /oauth/login)"));
        }
        Ok(session)
    }

    pub fn is_oauth(&self) -> bool {
        self.oauth
    }

    /// The setting (and the lease) this session is saved under.
    fn name(&self) -> String {
        if self.oauth { crate::oauth_client::setting(&self.identifier) } else { "session".into() }
    }

    /// The saved session of this identifier, if the store has one; true when it was taken.
    fn adopt_saved(&self) -> Result<bool, String> {
        let Some(store) = &self.store else { return Ok(false) };
        let saved = store.setting(&self.name())?.filter(|s| {
            (self.oauth || s.get("identifier").and_then(Value::as_str) == Some(&self.identifier))
                && s.get("refreshJwt").is_some_and(|r| r.as_str().is_some_and(|r| !r.is_empty()))
        });
        let found = saved.is_some();
        if let Some(saved) = saved {
            *self.tokens.lock().unwrap() = saved;
        }
        Ok(found)
    }

    async fn lease(&self) -> Result<Option<crate::store::Lease>, String> {
        match &self.store {
            Some(store) => store.lock(&self.name(), SESSION_LEASE).await.map(Some),
            None => Ok(None),
        }
    }

    fn access(&self) -> String {
        js_string(self.tokens.lock().unwrap().get("accessJwt"))
    }

    /// `used` (the access token a request carried) was refused: renew it once for every run sharing it.
    async fn renew(&self, used: &str) -> Result<(), String> {
        let _one = self.renewing.lock().await;
        if self.access() != used {
            return Ok(()); // another run here renewed it while this one waited
        }
        let _lease = self.lease().await?;
        if self.adopt_saved()? && self.access() != used {
            return Ok(()); // another process renewed it
        }
        self.refresh().await
    }

    pub fn did(&self) -> String {
        js_string(self.tokens.lock().unwrap().get("did"))
    }

    pub fn pds(&self) -> String {
        let tokens = self.tokens.lock().unwrap();
        if let Some(pds) = tokens.get("pds").and_then(Value::as_str) {
            return pds.to_string();
        }
        tokens
            .get("didDoc")
            .and_then(|doc| service_in(doc, &js_string(tokens.get("did")), "atproto_pds"))
            .unwrap_or_else(|| self.service.clone())
    }

    fn save(&self, session: Value) -> Result<(), String> {
        if self.oauth {
            *self.tokens.lock().unwrap() = session.clone();
            return self.store.as_ref().map_or(Ok(()), |store| store.set_setting(&self.name(), &session));
        }
        let mut tokens = json!({ "identifier": self.identifier });
        if let Value::Object(fields) = session {
            tokens.as_object_mut().unwrap().extend(fields);
        }
        *self.tokens.lock().unwrap() = tokens.clone();
        match &self.store {
            Some(store) => store.set_setting("session", &tokens),
            None => Ok(()),
        }
    }

    async fn post(
        &self,
        url: &str,
        nsid: &str,
        bearer: Option<String>,
        input: Option<Value>,
    ) -> Result<Value, XrpcError> {
        let mut headers = vec![];
        if let Some(b) = bearer {
            headers.push(("authorization".into(), format!("Bearer {b}")));
        }
        if input.is_some() {
            headers.push(("content-type".into(), "application/json".into()));
        }
        let response = self
            .fetch
            .fetch(
                "POST",
                url,
                headers,
                input.map(|i| i.to_string().into_bytes()),
            )
            .await
            .map_err(|message| XrpcError {
                message,
                error: None,
            })?;
        json_response(nsid, &response)
    }

    async fn create(&self) -> Result<(), String> {
        let nsid = "com.atproto.server.createSession";
        let input = json!({ "identifier": self.identifier, "password": self.password });
        // stderr, so the rate-limited logins are visible (stdout stays the result)
        eprintln!("atlibs: createSession for {}", self.identifier);
        let session = self
            .post(
                &xrpc_url(&self.service, nsid, None)?,
                nsid,
                None,
                Some(input),
            )
            .await
            .map_err(|e| e.message)?;
        self.save(session)
    }

    async fn refresh(&self) -> Result<(), String> {
        if self.oauth {
            let saved = self.tokens.lock().unwrap().clone();
            eprintln!("atlibs: refreshing the OAuth session of {}", self.identifier);
            let fresh = crate::oauth_client::refresh(self.fetch.as_ref(), &saved, self.store.as_ref())
                .await
                .map_err(|e| format!("{}: the OAuth session cannot be refreshed ({e}); sign in again", self.identifier))?;
            return self.save(fresh);
        }
        let nsid = "com.atproto.server.refreshSession";
        let refresh = js_string(self.tokens.lock().unwrap().get("refreshJwt"));
        eprintln!(
            "atlibs: refreshing the saved session for {}",
            self.identifier
        );
        match self
            .post(
                &xrpc_url(&self.pds(), nsid, None)?,
                nsid,
                Some(refresh),
                None,
            )
            .await
        {
            Ok(session) => self.save(session),
            Err(_) => self.create().await,
        }
    }

    pub async fn xrpc(
        &self,
        nsid: &str,
        params: Option<&Map<String, Value>>,
        input: Option<&Value>,
        headers: Vec<(String, String)>,
    ) -> Result<Value, String> {
        let body = input.map(|i| (i.to_string().into_bytes(), "application/json"));
        self.send(nsid, params, body, headers).await
    }

    /// XRPC with a body xrpc_body made (bytes or JSON, with its content-type).
    pub async fn xrpc_with(
        &self,
        nsid: &str,
        params: Option<&Map<String, Value>>,
        body: Option<(Vec<u8>, String)>,
        headers: Vec<(String, String)>,
    ) -> Result<Value, String> {
        let body = body.as_ref().map(|(b, t)| (b.clone(), t.as_str()));
        self.send(nsid, params, body, headers).await
    }

    /// com.atproto.repo.uploadBlob: the bytes as the body (atlibs publish).
    pub async fn upload_blob(&self, bytes: Vec<u8>, mime: &str) -> Result<Value, String> {
        self.send("com.atproto.repo.uploadBlob", None, Some((bytes, mime)), vec![])
            .await
    }

    async fn send(
        &self,
        nsid: &str,
        params: Option<&Map<String, Value>>,
        body: Option<(Vec<u8>, &str)>,
        headers: Vec<(String, String)>,
    ) -> Result<Value, String> {
        if self.oauth {
            return self.send_oauth(nsid, params, body, headers).await;
        }
        for retried in [false, true] {
            let access = self.access();
            let mut all = headers.clone();
            all.push(("authorization".into(), format!("Bearer {access}")));
            if let Some((_, content_type)) = &body {
                all.push(("content-type".into(), content_type.to_string()));
            }
            let url = xrpc_url(&self.pds(), nsid, params)?;
            let response = self
                .fetch
                .fetch(
                    if body.is_some() { "POST" } else { "GET" },
                    &url,
                    all,
                    body.as_ref().map(|(b, _)| b.clone()),
                )
                .await?;
            match json_response(nsid, &response) {
                Ok(body) => return Ok(body),
                Err(e) if !retried && matches!(e.error.as_deref(), Some("ExpiredToken" | "InvalidToken")) => {
                    self.renew(&access).await?
                }
                Err(e) => return Err(e.message),
            }
        }
        unreachable!("the second try returns")
    }

    /// XRPC with the DPoP-bound access token: renewed first when it expires within a minute (one
    /// refresh for every run sharing it), a use_dpop_nonce answer sent again with the new nonce, and a
    /// refused token renewed once.
    async fn send_oauth(
        &self,
        nsid: &str,
        params: Option<&Map<String, Value>>,
        body: Option<(Vec<u8>, &str)>,
        headers: Vec<(String, String)>,
    ) -> Result<Value, String> {
        use crate::oauth_client::{keep_nonce, nonce_refused, proof, token_refused};
        let expiring = || self.tokens.lock().unwrap().get("expiresAt").and_then(Value::as_u64).is_some_and(|at| at < crate::now_ms() + 60_000);
        if expiring() {
            self.renew(&self.access()).await?;
        }
        let (mut renewed, mut nonced) = (false, false);
        loop {
            let access = self.access();
            let key = self.tokens.lock().unwrap().get("dpopKey").and_then(Value::as_str).map(atproto_identity::key::identify_key);
            let key = key.ok_or("the OAuth session has no DPoP key")?.map_err(|e| e.to_string())?;
            let url = xrpc_url(&self.pds(), nsid, params)?;
            let method = if body.is_some() { "POST" } else { "GET" };
            let mut all = headers.clone();
            all.push(("authorization".into(), format!("DPoP {access}")));
            all.push(("dpop".into(), proof(&key, method, &url, Some(&access))?));
            if let Some((_, content_type)) = &body {
                all.push(("content-type".into(), content_type.to_string()));
            }
            let response = self.fetch.fetch(method, &url, all, body.as_ref().map(|(b, _)| b.clone())).await?;
            keep_nonce(&url, &response);
            if nonce_refused(&response) && !nonced {
                nonced = true;
                continue;
            }
            match json_response(nsid, &response) {
                Ok(body) => return Ok(body),
                Err(e) if !renewed && (token_refused(&response) || matches!(e.error.as_deref(), Some("ExpiredToken" | "InvalidToken" | "invalid_token"))) => {
                    renewed = true;
                    self.renew(&access).await?
                }
                Err(e) => return Err(e.message),
            }
        }
    }
}

/// hostCall: the run-module host contract.
pub struct HostCall {
    pub fetch: Arc<dyn Fetch>,
    pub session: Option<Arc<Session>>,
    /// Some(store) enables com.fedfork.exec.attest (needs a session and a module that lists calls).
    pub attest: Option<Store>,
}

impl Call for HostCall {
    fn call(&self, nsid: String, request: Request) -> BoxFut<'_, Result<Option<Value>, String>> {
        Box::pin(async move {
            let Request {
                params,
                input,
                encoding,
                proxy,
                service,
                endpoints,
                ..
            } = request;
            if nsid == ATTEST_NSID {
                let (Some(store), Some(session)) = (&self.attest, &self.session) else {
                    return Err(
                        "attest needs a session and a module record that lists its calls".into(),
                    );
                };
                return crypto::attest(store, Some(&session.did()), input.as_ref()).map(Some);
            }
            // Other host pseudo-calls (a transport this host was built without, ...) must not reach a PDS.
            if nsid.starts_with(transport::PSEUDO) {
                return Err(transport::unavailable(&nsid).unwrap_or_else(|| format!("{nsid} is not available on this host")));
            }
            // did#id with its endpoints: an https one (TransportCall took any of its transports' schemes),
            // with service auth aud did#id
            if let (Some(endpoints), Some(aud)) = (&endpoints, &service) {
                let origin = endpoint_of(endpoints, "https").filter(|o| o.starts_with("https://")).ok_or_else(|| format!("{nsid}: this host reaches none of {aud}'s endpoints"))?;
                let session = self.session.as_ref().ok_or_else(|| format!("{nsid} needs a session (sign in: atlibs login + --as, or --app-password)"))?;
                let token = mint(session, aud, &nsid).await?;
                return bearer_call(self.fetch.as_ref(), &origin, &nsid, params.as_ref(), xrpc_body(input.as_ref(), encoding.as_deref())?, service_token(&token)?)
                    .await
                    .map(Some);
            }
            // A DID service (did: its PDS; did#id: that service) is a public GET, never with credentials;
            // so is a proxied query signed out, sent to the proxy's service directly.
            let open = service.as_deref().filter(|s| s.starts_with("did:")).or(proxy.as_deref().filter(|_| self.session.is_none() && input.is_none()));
            if let Some(did) = open {
                let pds = pds_of(self.fetch.as_ref(), did).await?;
                return get_json(
                    self.fetch.as_ref(),
                    &nsid,
                    &xrpc_url(&pds, &nsid, params.as_ref())?,
                )
                .await
                .map(Some);
            }
            let Some(session) = &self.session else {
                return Err(format!(
                    "{nsid} needs a session (sign in: atlibs login + --as, or --app-password)"
                ));
            };
            if let Some(service) = service {
                let auth_params: Map<String, Value> = [
                    ("aud".to_string(), json!(service_aud(&service))),
                    ("lxm".to_string(), json!(nsid)),
                ]
                .into_iter()
                .collect();
                let token = session
                    .xrpc(
                        "com.atproto.server.getServiceAuth",
                        Some(&auth_params),
                        None,
                        vec![],
                    )
                    .await?;
                return bearer_call(
                    self.fetch.as_ref(),
                    &service,
                    &nsid,
                    params.as_ref(),
                    xrpc_body(input.as_ref(), encoding.as_deref())?,
                    service_token(&token)?,
                )
                .await
                .map(Some);
            }
            let headers = proxy
                .map(|p| vec![("atproto-proxy".to_string(), p)])
                .unwrap_or_default();
            session
                .xrpc_with(&nsid, params.as_ref(), xrpc_body(input.as_ref(), encoding.as_deref())?, headers)
                .await
                .map(Some)
        })
    }

    // A subscribe opens wss://<service, the DID's PDS, or the session PDS>/xrpc/<nsid>?<params>: public
    // unless serviceAuth asks for the session's service-auth token (header or query parameter).
    fn subscribe(&self, nsid: String, mut request: Request) -> BoxFut<'_, Result<Sub, String>> {
        Box::pin(async move {
            if nsid.starts_with(transport::PSEUDO) {
                return Err(transport::unavailable(&nsid).unwrap_or_else(|| format!("{nsid} cannot be subscribed to on this host")));
            }
            if request.proxy.is_some() {
                return Err(
                    "subscribe cannot go through atproto-proxy; call the service directly".into(),
                );
            }
            let base = match request.service.as_deref() {
                Some(did) if did.starts_with("did:") => pds_of(self.fetch.as_ref(), did).await?,
                Some(service) => service.to_string(),
                None => self
                    .session
                    .as_ref()
                    .map(|s| s.pds())
                    .ok_or_else(|| format!("{nsid} needs a service or a session"))?,
            };
            if request.service_auth.is_some() {
                let session = self.session.as_ref().ok_or_else(|| {
                    format!("{nsid} serviceAuth needs a session (sign in: atlibs login + --as, or --app-password)")
                })?;
                let service = request
                    .service
                    .clone()
                    .filter(|s| s.starts_with("https://"))
                    .ok_or_else(|| format!("{nsid} serviceAuth needs an https service"))?;
                let auth_params: Map<String, Value> = [
                    ("aud".to_string(), json!(service_aud(&service))),
                    ("lxm".to_string(), json!(nsid)),
                ]
                .into_iter()
                .collect();
                let token = session
                    .xrpc("com.atproto.server.getServiceAuth", Some(&auth_params), None, vec![])
                    .await?;
                apply_service_auth(&mut request, service_token(&token)?);
            }
            let url = ws::ws_url(&base, &nsid, request.params.as_ref())?;
            Ok(ws::subscribe(
                url,
                nsid,
                request.framing,
                request.keepalive,
                request.headers,
                request.signal,
            ))
        })
    }
}

/// Put a subscribe's service-auth token where its serviceAuth asks: { param } in that query
/// parameter, {} as Authorization (hosts whose WebSocket takes headers).
/// getServiceAuth's token: a non-empty string, never "undefined" sent as one (JS: "no service-auth token").
fn service_token(answer: &Value) -> Result<&str, String> {
    answer.get("token").and_then(Value::as_str).filter(|t| !t.is_empty()).ok_or_else(|| "getServiceAuth answered no service-auth token".into())
}

pub fn apply_service_auth(request: &mut Request, token: &str) {
    let Some(auth) = request.service_auth.take() else {
        return;
    };
    match auth.get("param").and_then(Value::as_str) {
        Some(param) => {
            request
                .params
                .get_or_insert_with(Map::new)
                .insert(param.to_string(), json!(token));
        }
        None => request
            .headers
            .push(("authorization".into(), format!("Bearer {token}"))),
    }
}

pub(crate) async fn bearer_call(
    fetch: &dyn Fetch,
    service: &str,
    nsid: &str,
    params: Option<&Map<String, Value>>,
    body: Option<(Vec<u8>, String)>,
    token: &str,
) -> Result<Value, String> {
    let mut headers = vec![("authorization".to_string(), format!("Bearer {token}"))];
    if let Some((_, content_type)) = &body {
        headers.push(("content-type".into(), content_type.clone()));
    }
    let response = fetch
        .fetch(
            if body.is_some() { "POST" } else { "GET" },
            &xrpc_url(service, nsid, params)?,
            headers,
            body.map(|(b, _)| b),
        )
        .await?;
    json_response(nsid, &response).map_err(|e| e.message)
}

/// The address of the first endpoint of `kind` ([{ kind, address }]).
pub fn endpoint_of(endpoints: &Value, kind: &str) -> Option<String> {
    endpoints.as_array()?.iter().find(|e| e["kind"] == kind).and_then(|e| e["address"].as_str()).map(String::from)
}

/// getServiceAuth { aud, lxm } with the session.
pub async fn mint(session: &Session, aud: &str, lxm: &str) -> Result<Value, String> {
    let params: Map<String, Value> = [("aud".to_string(), json!(aud)), ("lxm".to_string(), json!(lxm))].into_iter().collect();
    session.xrpc("com.atproto.server.getServiceAuth", Some(&params), None, vec![]).await
}

/// JS serviceAud: the did:web aud of a service-auth token for an https service URL,
/// did:web:<host>[%3A<port>].
pub fn service_aud(url: &str) -> String {
    let parsed = reqwest::Url::parse(url).ok();
    let host = parsed.as_ref().and_then(|u| u.host_str()).unwrap_or_default();
    match parsed.as_ref().and_then(|u| u.port()) {
        Some(port) => format!("did:web:{host}%3A{port}"),
        None => format!("did:web:{host}"),
    }
}

fn origin(url: &str) -> Option<String> {
    reqwest::Url::parse(url)
        .ok()
        .map(|u| u.origin().ascii_serialization())
}

/// didkeyCall: the com.fedfork.exec.didkey.* pseudo-calls, and request calls to an https service in
/// `services` carrying a service-auth token the did:key mints; everything else goes to inner.
pub struct DidkeyCall {
    pub key: DidKey,
    pub services: Vec<String>,
    pub fetch: Arc<dyn Fetch>,
    pub inner: Arc<dyn Call>,
    /// One serving instance per did:key: the relay keeps only the newest registration of a DID, so two
    /// runs serving one identity would take it from each other on every reconnect. With a store, a
    /// subscribe to a listed service holds the state dir's lease "atlibs-didkey-serving:<did>" while it
    /// is open (renewed; every process sharing the state dir, JS ones too) and is refused while another
    /// run holds it.
    pub store: Option<Store>,
}

const SERVING_LEASE: std::time::Duration = std::time::Duration::from_secs(30);

/// `sub` with `lease` held, and renewed, until either side ends it.
fn leased(sub: Sub, lease: crate::store::Lease) -> Sub {
    use tokio::sync::mpsc;
    let (out, messages) = mpsc::unbounded_channel();
    let (commands, mut asked) = mpsc::unbounded_channel();
    let Sub { messages: mut from, commands: to } = sub;
    tokio::spawn(async move {
        let mut renew = tokio::time::interval(SERVING_LEASE / 3);
        loop {
            tokio::select! {
                message = from.recv() => match message {
                    Some(m) => if out.send(m).is_err() { break }, // nobody reads it any more
                    None => return, // it ended: the lease drops
                },
                command = asked.recv() => match command {
                    Some(crate::run::SubCmd::Close) | None => break,
                    Some(c) => drop(to.send(c)),
                },
                _ = renew.tick() => {
                    if !lease.renew(SERVING_LEASE).unwrap_or(false) {
                        let _ = out.send(Err("the did:key serving lease was lost".into()));
                        break;
                    }
                }
            }
        }
        let _ = to.send(crate::run::SubCmd::Close);
    });
    Sub { messages, commands }
}

impl Call for DidkeyCall {
    fn call(&self, nsid: String, request: Request) -> BoxFut<'_, Result<Option<Value>, String>> {
        Box::pin(async move {
            if let Some(done) =
                self.key
                    .pseudo_call(&nsid, request.params.as_ref(), request.input.as_ref())
            {
                return done.map(Some);
            }
            let service = request
                .service
                .clone()
                .filter(|s| s.starts_with("https://"));
            if let Some(service) = service.filter(|s| {
                origin(s)
                    .is_some_and(|o| self.services.iter().any(|x| origin(x).as_ref() == Some(&o)))
            }) {
                let token = self.key.service_auth(&service_aud(&service), &nsid, None);
                let call = bearer_call(
                    self.fetch.as_ref(),
                    &service,
                    &nsid,
                    request.params.as_ref(),
                    xrpc_body(request.input.as_ref(), request.encoding.as_deref())?,
                    &token,
                );
                return tokio::select! { out = call => out.map(Some), reason = request.signal.aborted() => Err(reason) };
            }
            self.inner.call(nsid, request).await
        })
    }
    // A subscribe with serviceAuth to a listed service: the did:key mints the token, inner opens the socket.
    fn subscribe(&self, nsid: String, mut request: Request) -> BoxFut<'_, Result<Sub, String>> {
        Box::pin(async move {
            let listed = request
                .service
                .clone()
                .filter(|s| s.starts_with("https://"))
                .filter(|s| {
                    origin(s).is_some_and(|o| {
                        self.services.iter().any(|x| origin(x).as_ref() == Some(&o))
                    })
                });
            if let (Some(service), true) = (listed, request.service_auth.is_some()) {
                let token = self.key.service_auth(&service_aud(&service), &nsid, None);
                apply_service_auth(&mut request, &token);
                if let Some(store) = &self.store {
                    let lease = store.try_lock(&format!("atlibs-didkey-serving:{}", self.key.did), SERVING_LEASE)?.ok_or_else(|| {
                        format!("{} is already serving in another run: one serving instance per did:key (the relay keeps only its newest registration)", self.key.did)
                    })?;
                    return Ok(leased(self.inner.subscribe(nsid, request).await?, lease));
                }
            }
            self.inner.subscribe(nsid, request).await
        })
    }
}

#[cfg(test)]
mod tests {
    #[test]
    fn bytes_go_in_and_come_out_as_dollar_bytes() {
        let body = |input: Value, encoding: Option<&str>| xrpc_body(Some(&input), encoding).unwrap().unwrap();
        assert_eq!(body(json!({ "$bytes": "AAEC/w" }), Some("application/x-git-packed-objects")), (vec![0, 1, 2, 255], "application/x-git-packed-objects".into()));
        assert_eq!(body(json!({ "$bytes": "AAEC/w==" }), None), (vec![0, 1, 2, 255], "application/octet-stream".into()));
        assert_eq!(body(json!({ "$bytes": "AA", "x": 1 }), None).1, "application/json");
        let answer = |t: &str| Response { status: 200, headers: vec![("content-type".into(), t.into())], body: vec![0, 1, 2, 255] };
        assert_eq!(json_response("x", &answer("application/octet-stream")).ok(), Some(json!({ "$bytes": "AAEC/w" })));
        assert_eq!(json_response("x", &answer("text/plain")).ok(), Some(json!({})), "text is not bytes");
    }

    #[test]
    fn json_response_reads_an_event_stream() {
        let body = "event: meta\ndata: {\"jobUri\":\"at://x\"}\n\n: comment\n\ndata: one\ndata: two\n\n";
        let response = super::Response { status: 200, headers: vec![("content-type".into(), "text/event-stream; charset=utf-8".into())], body: body.into() };
        let out = super::json_response("dev.cocore.inference.dispatch", &response).ok().unwrap();
        assert_eq!(out, serde_json::json!({ "events": [{ "event": "meta", "data": { "jobUri": "at://x" } }, { "event": "message", "data": "one\ntwo" }] }));
    }

    use super::*;
    use std::collections::HashMap;

    type Seen = Arc<Mutex<Vec<(String, String, Vec<(String, String)>, Option<String>)>>>;

    /// A fake fetch: answers by URL substring, records every request.
    struct Fake(Vec<(&'static str, u16, Value)>, Seen);

    impl Fetch for Fake {
        fn fetch(
            &self,
            method: &str,
            url: &str,
            headers: Vec<(String, String)>,
            body: Option<Vec<u8>>,
        ) -> BoxFut<'_, Result<Response, String>> {
            self.1.lock().unwrap().push((
                method.into(),
                url.into(),
                headers,
                body.map(|b| String::from_utf8(b).unwrap()),
            ));
            let (_, status, body) = self
                .0
                .iter()
                .find(|(m, ..)| url.contains(m))
                .cloned()
                .unwrap_or(("", 404, json!({ "error": "NotFound" })));
            Box::pin(async move {
                Ok(Response {
                    status,
                    headers: vec![],
                    body: body.to_string().into_bytes(),
                })
            })
        }
    }

    fn did_doc(pds: &str) -> Value {
        json!({ "service": [{ "id": "#atproto_pds", "serviceEndpoint": pds }] })
    }

    #[tokio::test]
    async fn every_host_call_route() {
        let seen: Seen = Arc::default();
        let fake = Arc::new(Fake(
            vec![
                (
                    "createSession",
                    200,
                    json!({ "did": "did:plc:me", "handle": "me.test", "accessJwt": "a1", "refreshJwt": "r1", "didDoc": did_doc("https://pds.me") }),
                ),
                (
                    "plc.directory/did:plc:other",
                    200,
                    did_doc("https://pds.other"),
                ),
                (
                    "pds.other/xrpc/com.atproto.repo.describeRepo",
                    200,
                    json!({ "handle": "other.test" }),
                ),
                ("getServiceAuth", 200, json!({ "token": "svc" })),
                ("svc.example/xrpc/a.b.c", 200, json!({ "ok": 1 })),
                (
                    "pds.me/xrpc/com.atproto.repo.getRecord",
                    200,
                    json!({ "value": {} }),
                ),
                (
                    "api.example/.well-known/did.json",
                    200,
                    json!({ "service": [{ "id": "#svc", "serviceEndpoint": "https://api.example" }] }),
                ),
                ("api.example/xrpc/q.r.s", 200, json!({ "open": 1 })),
                (
                    "pds.me/xrpc/x.y.z",
                    429,
                    json!({ "error": "RateLimitExceeded" }),
                ),
            ],
            seen.clone(),
        ));
        let store = Store::open(&std::env::temp_dir().join(format!("atlibs-host-session-{}", crate::store::uuid())), "atlibs-server").unwrap();
        let session = Session::login(
            "me.test",
            Some("pw".into()),
            "https://bsky.social",
            fake.clone(),
            Some(store.clone()),
        )
        .await
        .unwrap();
        assert_eq!(
            (session.did(), session.pds()),
            ("did:plc:me".into(), "https://pds.me".into())
        );
        assert_eq!(
            store.setting("session").unwrap().unwrap()["identifier"],
            "me.test"
        );
        let call = HostCall {
            fetch: fake.clone(),
            session: Some(session),
            attest: None,
        };
        let req = |params: Value, service: Option<&str>, proxy: Option<&str>| Request {
            params: params.as_object().cloned(),
            service: service.map(String::from),
            proxy: proxy.map(String::from),
            ..Default::default()
        };

        let read = call
            .call(
                "com.atproto.repo.describeRepo".into(),
                req(
                    json!({ "repo": "did:plc:other" }),
                    Some("did:plc:other"),
                    None,
                ),
            )
            .await
            .unwrap();
        assert_eq!(read, Some(json!({ "handle": "other.test" })));
        assert!(
            seen.lock().unwrap().last().unwrap().2.is_empty(),
            "a DID read carries no token"
        );

        assert_eq!(
            call.call(
                "a.b.c".into(),
                req(json!({}), Some("https://svc.example"), None)
            )
            .await
            .unwrap(),
            Some(json!({ "ok": 1 }))
        );
        let log = seen.lock().unwrap().clone();
        assert!(
            log[log.len() - 2]
                .1
                .contains("getServiceAuth?aud=did%3Aweb%3Asvc.example&lxm=a.b.c")
        );
        assert_eq!(
            log.last().unwrap().2,
            vec![("authorization".to_string(), "Bearer svc".to_string())]
        );

        // did#id with endpoints (no transport took one): its https endpoint, service auth aud did#id
        let endpoints = Request {
            endpoints: Some(json!([{ "kind": "iroh", "address": "endpointabc234" }, { "kind": "https", "address": "https://svc.example" }])),
            ..req(json!({}), Some("did:plc:bidder#pdr_temp_market"), None)
        };
        assert_eq!(call.call("a.b.c".into(), endpoints).await.unwrap(), Some(json!({ "ok": 1 })));
        let log = seen.lock().unwrap().clone();
        assert!(log[log.len() - 2].1.contains("getServiceAuth?aud=did%3Aplc%3Abidder%23pdr_temp_market&lxm=a.b.c"), "{}", log[log.len() - 2].1);
        assert_eq!(log.last().unwrap().2, vec![("authorization".to_string(), "Bearer svc".to_string())]);
        let unreachable = Request { endpoints: Some(json!([{ "kind": "iroh", "address": "endpointabc234" }])), ..req(json!({}), Some("did:plc:bidder#pdr_temp_market"), None) };
        assert_eq!(call.call("a.b.c".into(), unreachable).await.unwrap_err(), "a.b.c: this host reaches none of did:plc:bidder#pdr_temp_market's endpoints");

        call.call(
            "com.atproto.repo.getRecord".into(),
            req(
                json!({ "repo": "x" }),
                None,
                Some("did:web:api.example#svc"),
            ),
        )
        .await
        .unwrap();
        let last = seen.lock().unwrap().last().unwrap().clone();
        assert!(
            last.2
                .contains(&("atproto-proxy".into(), "did:web:api.example#svc".into()))
                && last
                    .2
                    .contains(&("authorization".into(), "Bearer a1".into()))
        );

        assert_eq!(
            call.call("x.y.z".into(), req(json!({}), None, None))
                .await
                .unwrap_err(),
            "x.y.z failed with RateLimitExceeded"
        );
        assert_eq!(
            call.call(ATTEST_NSID.into(), Request::default())
                .await
                .unwrap_err(),
            "attest needs a session and a module record that lists its calls"
        );
        let signed_out = HostCall {
            fetch: fake,
            session: None,
            attest: None,
        };
        assert_eq!(
            signed_out
                .call("x.y.z".into(), Request::default())
                .await
                .unwrap_err(),
            "x.y.z needs a session (sign in: atlibs login + --as, or --app-password)"
        );
        // Signed out, a proxied query (and any did#id service) goes straight to that service, no credentials.
        for request in [req(json!({ "a": 1 }), None, Some("did:web:api.example#svc")), req(json!({ "a": 1 }), Some("did:web:api.example#svc"), None)] {
            assert_eq!(signed_out.call("q.r.s".into(), request).await.unwrap(), Some(json!({ "open": 1 })));
            let last = seen.lock().unwrap().last().unwrap().clone();
            assert_eq!((last.1.as_str(), last.2.len()), ("https://api.example/xrpc/q.r.s?a=1", 0));
        }
        let procedure = Request { input: Some(json!({})), ..req(json!({}), None, Some("did:web:api.example#svc")) };
        assert!(signed_out.call("q.r.s".into(), procedure).await.unwrap_err().ends_with("needs a session (sign in: atlibs login + --as, or --app-password)"));
        assert_eq!(
            signed_out.call("q.r.s".into(), req(json!({}), Some("did:web:api.example#nope"), None)).await.unwrap_err(),
            "did:web:api.example has no #nope service"
        );
        assert_eq!(
            signed_out
                .subscribe("com.fedfork.exec.attest".into(), Request::default())
                .await
                .err()
                .unwrap(),
            "com.fedfork.exec.attest cannot be subscribed to on this host"
        );
        assert_eq!(
            signed_out
                .subscribe("a.b.c".into(), Request::default())
                .await
                .err()
                .unwrap(),
            "a.b.c needs a service or a session"
        );
    }

    #[tokio::test]
    async fn expired_tokens_refresh_once_and_saved_sessions_skip_create() {
        struct Expiring(Mutex<u32>, Seen);
        impl Fetch for Expiring {
            fn fetch(
                &self,
                method: &str,
                url: &str,
                headers: Vec<(String, String)>,
                _: Option<Vec<u8>>,
            ) -> BoxFut<'_, Result<Response, String>> {
                self.1
                    .lock()
                    .unwrap()
                    .push((method.into(), url.into(), headers.clone(), None));
                let mut n = self.0.lock().unwrap();
                *n += 1;
                let (status, body) = if url.contains("refreshSession") {
                    (
                        200,
                        json!({ "did": "did:plc:me", "accessJwt": "a2", "refreshJwt": "r2" }),
                    )
                } else if headers.iter().any(|(_, v)| v == "Bearer a1") {
                    (400, json!({ "error": "ExpiredToken" }))
                } else {
                    (200, json!({ "ok": true }))
                };
                Box::pin(async move {
                    Ok(Response {
                        status,
                        headers: vec![],
                        body: body.to_string().into_bytes(),
                    })
                })
            }
        }
        let seen: Seen = Arc::default();
        let fetch = Arc::new(Expiring(Mutex::new(0), seen.clone()));
        let store = Store::open(&std::env::temp_dir().join(format!("atlibs-host-session-{}", crate::store::uuid())), "atlibs-server").unwrap();
        store.set_setting("session", &json!({ "identifier": "me.test", "did": "did:plc:me", "accessJwt": "a1", "refreshJwt": "r1" })).unwrap();
        let session = Session::login("me.test", None, "https://pds.me", fetch, Some(store)).await.unwrap();
        assert_eq!(
            session.xrpc("a.b.c", None, None, vec![]).await.unwrap(),
            json!({ "ok": true })
        );
        let urls: Vec<String> = seen.lock().unwrap().iter().map(|r| r.1.clone()).collect();
        assert_eq!(
            urls,
            [
                "https://pds.me/xrpc/a.b.c",
                "https://pds.me/xrpc/com.atproto.server.refreshSession",
                "https://pds.me/xrpc/a.b.c"
            ]
        );
    }

    /// A PDS that takes each refresh token once (rotation), so a second refresh with r1 would fail and
    /// fall back to createSession; a1 is expired, and refused as invalid once rotated.
    #[derive(Default)]
    struct Rotating {
        refreshes: Mutex<u32>,
        creates: Mutex<u32>,
    }
    impl Fetch for Rotating {
        fn fetch(&self, _: &str, url: &str, headers: Vec<(String, String)>, _: Option<Vec<u8>>) -> BoxFut<'_, Result<Response, String>> {
            let bearer = headers.iter().find(|(k, _)| k == "authorization").map(|(_, v)| v.clone()).unwrap_or_default();
            let (status, body) = if url.contains("refreshSession") {
                let mut n = self.refreshes.lock().unwrap();
                *n += 1;
                if bearer == "Bearer r1" && *n == 1 {
                    (200, json!({ "did": "did:plc:me", "accessJwt": "a2", "refreshJwt": "r2" }))
                } else {
                    (400, json!({ "error": "ExpiredToken", "message": "refresh token already used" }))
                }
            } else if url.contains("createSession") {
                *self.creates.lock().unwrap() += 1;
                (200, json!({ "did": "did:plc:me", "accessJwt": "a9", "refreshJwt": "r9" }))
            } else if bearer == "Bearer a1" && *self.refreshes.lock().unwrap() > 0 {
                (401, json!({ "error": "InvalidToken" }))
            } else if bearer == "Bearer a1" {
                (400, json!({ "error": "ExpiredToken" }))
            } else {
                (200, json!({ "ok": true }))
            };
            Box::pin(async move {
                tokio::time::sleep(std::time::Duration::from_millis(10)).await; // let runs overlap
                Ok(Response { status, headers: vec![], body: body.to_string().into_bytes() })
            })
        }
    }

    #[tokio::test]
    async fn concurrent_runs_share_one_refresh_in_process_and_across_processes() {
        let dir = std::env::temp_dir().join(format!("atlibs-host-session-{}", crate::store::uuid()));
        let saved = json!({ "identifier": "me.test", "did": "did:plc:me", "accessJwt": "a1", "refreshJwt": "r1" });
        let open = || Store::open(&dir, "atlibs-server").unwrap(); // its own sqlite connection, as another process has
        open().set_setting("session", &saved).unwrap();
        let login = |fetch: Arc<Rotating>| Session::login("me.test", Some("pw".into()), "https://pds.me", fetch, Some(open()));
        let ok = json!({ "ok": true });

        // 8 runs on one session: one refresh, everyone retries with it
        let pds = Arc::new(Rotating::default());
        let session = login(pds.clone()).await.unwrap();
        let ran = futures_util::future::join_all((0..8).map(|_| session.xrpc("a.b.c", None, None, vec![]))).await;
        assert!(ran.iter().all(|r| r.as_ref() == Ok(&ok)), "{ran:?}");
        assert_eq!((*pds.refreshes.lock().unwrap(), *pds.creates.lock().unwrap()), (1, 0));

        // two "processes" (sessions on their own connections) with the same saved, expired session: one refresh
        open().set_setting("session", &saved).unwrap();
        let pds = Arc::new(Rotating::default());
        let (a, b) = (login(pds.clone()).await.unwrap(), login(pds.clone()).await.unwrap());
        let ran = futures_util::future::join_all((0..8).map(|i| if i % 2 == 0 { a.xrpc("a.b.c", None, None, vec![]) } else { b.xrpc("a.b.c", None, None, vec![]) })).await;
        assert!(ran.iter().all(|r| r.as_ref() == Ok(&ok)), "{ran:?}");
        assert_eq!((*pds.refreshes.lock().unwrap(), *pds.creates.lock().unwrap()), (1, 0));

        // a run still holding the rotated-away token (401 InvalidToken) re-reads the saved session: no refresh
        open().set_setting("session", &saved).unwrap();
        let pds = Arc::new(Rotating::default());
        let (a, stale) = (login(pds.clone()).await.unwrap(), login(pds.clone()).await.unwrap());
        assert_eq!(a.xrpc("a.b.c", None, None, vec![]).await, Ok(ok.clone()));
        assert_eq!(stale.xrpc("a.b.c", None, None, vec![]).await, Ok(ok.clone()));
        assert_eq!((*pds.refreshes.lock().unwrap(), *pds.creates.lock().unwrap()), (1, 0));
        assert_eq!(open().setting("session").unwrap().unwrap()["refreshJwt"], "r2");

        // an empty state dir: concurrent first logins create one session
        let empty = std::env::temp_dir().join(format!("atlibs-host-session-{}", crate::store::uuid()));
        let pds = Arc::new(Rotating::default());
        let first = |pds: Arc<Rotating>| Session::login("me.test", Some("pw".into()), "https://pds.me", pds, Some(Store::open(&empty, "atlibs-server").unwrap()));
        let (x, y) = futures_util::future::join(first(pds.clone()), first(pds.clone())).await;
        assert!(x.is_ok() && y.is_ok());
        assert_eq!(*pds.creates.lock().unwrap(), 1);
        let _ = std::fs::remove_dir_all(&dir);
        let _ = std::fs::remove_dir_all(&empty);
    }

    /// An authorization server + PDS for OAuth sessions: refresh tokens used once (rotating), DPoP nonces
    /// required (use_dpop_nonce first), access tokens per DID (a token used for another repo is refused).
    #[derive(Default)]
    struct OAuthPds {
        refreshes: Mutex<u32>,
        writes: Mutex<Vec<(String, String)>>, // (token's DID, repo written)
    }
    impl Fetch for OAuthPds {
        fn fetch(&self, _: &str, url: &str, headers: Vec<(String, String)>, body: Option<Vec<u8>>) -> BoxFut<'_, Result<Response, String>> {
            let header = |n: &str| headers.iter().find(|(k, _)| k == n).map(|(_, v)| v.clone()).unwrap_or_default();
            let proof: Value = header("dpop").split('.').nth(1).and_then(|p| crate::store::unbase64(p).ok()).and_then(|b| serde_json::from_slice(&b).ok()).unwrap_or_default();
            let nonce = vec![("dpop-nonce".to_string(), "n1".to_string())];
            let body = String::from_utf8(body.unwrap_or_default()).unwrap_or_default();
            let (status, extra, out) = if proof["nonce"] != "n1" {
                if url.contains("/token") { (400, vec![], json!({ "error": "use_dpop_nonce" })) } else { (401, vec![("www-authenticate".to_string(), "DPoP error=\"use_dpop_nonce\"".to_string())], json!({ "error": "use_dpop_nonce" })) }
            } else if url.contains("/token") {
                let form: HashMap<String, String> = url::form_urlencoded::parse(body.as_bytes()).into_owned().collect();
                let did = form["refresh_token"].split('|').next().unwrap_or_default().to_string();
                let mut n = self.refreshes.lock().unwrap();
                *n += 1;
                if form["refresh_token"].ends_with("|r1") && form.get("client_id").is_some_and(|c| c.starts_with("http://localhost?")) {
                    (200, vec![], json!({ "access_token": format!("{did}|a2"), "refresh_token": format!("{did}|r2"), "token_type": "DPoP", "expires_in": 3600, "sub": did, "scope": "atproto" }))
                } else {
                    (400, vec![], json!({ "error": "invalid_grant" }))
                }
            } else {
                let token = header("authorization").strip_prefix("DPoP ").unwrap_or_default().to_string();
                let (did, which) = token.split_once('|').unwrap_or_default();
                let repo = serde_json::from_str::<Value>(&body).ok().and_then(|b| b["repo"].as_str().map(String::from)).unwrap_or_default();
                if which != "a2" || proof["ath"].as_str().is_none() {
                    (401, vec![("www-authenticate".to_string(), "DPoP error=\"invalid_token\"".to_string())], json!({ "error": "invalid_token" }))
                } else if repo != did {
                    (403, vec![], json!({ "error": "Forbidden", "message": "not your repo" }))
                } else {
                    self.writes.lock().unwrap().push((did.to_string(), repo));
                    (200, vec![], json!({ "uri": format!("at://{did}/x/1") }))
                }
            };
            Box::pin(async move {
                tokio::time::sleep(std::time::Duration::from_millis(10)).await;
                Ok(Response { status, headers: [nonce, extra].concat(), body: out.to_string().into_bytes() })
            })
        }
    }

    #[tokio::test]
    async fn oauth_sessions_refresh_once_per_user_and_never_cross() {
        let dir = std::env::temp_dir().join(format!("atlibs-host-oauth-{}", crate::store::uuid()));
        let open = || Store::open(&dir, "atlibs-server").unwrap();
        let save = |did: &str| {
            let key = atproto_identity::key::generate_key(atproto_identity::key::KeyType::P256Private).unwrap();
            let saved = json!({ "did": did, "pds": "https://pds.test", "issuer": "https://pds.test", "tokenEndpoint": "https://pds.test/oauth/token",
                "dpopKey": key.to_string(), "clientId": "http://localhost?scope=atproto", "authMethod": "none",
                "accessJwt": format!("{did}|a1"), "refreshJwt": format!("{did}|r1"), "expiresAt": crate::now_ms() + 1000 });
            open().set_setting(&crate::oauth_client::setting(did), &saved).unwrap();
        };
        let write = |s: &Arc<Session>, repo: &str| {
            let (s, input) = (s.clone(), json!({ "repo": repo }));
            async move { s.xrpc("com.atproto.repo.createRecord", None, Some(&input), vec![]).await }
        };
        assert!(Session::oauth("did:plc:nobody", Arc::new(OAuthPds::default()), open()).await.is_err(), "no saved session, no session");

        // N=8 runs for one user, its token expiring, across two "processes" (sessions on their own connections): one refresh
        save("did:plc:alice");
        let pds = Arc::new(OAuthPds::default());
        let (a, b) = (Session::oauth("did:plc:alice", pds.clone(), open()).await.unwrap(), Session::oauth("did:plc:alice", pds.clone(), open()).await.unwrap());
        assert!(a.is_oauth() && a.did() == "did:plc:alice" && a.pds() == "https://pds.test");
        let ran = futures_util::future::join_all((0..8).map(|i| write(if i % 2 == 0 { &a } else { &b }, "did:plc:alice"))).await;
        assert!(ran.iter().all(Result::is_ok), "{ran:?}");
        assert_eq!(*pds.refreshes.lock().unwrap(), 1);
        assert_eq!(open().setting("oauth:did:plc:alice").unwrap().unwrap()["refreshJwt"], "did:plc:alice|r2");

        // 3 users x 3 runs at once: one refresh each, every write in the user's own repo with their own token
        let pds = Arc::new(OAuthPds::default());
        let mut runs = vec![];
        for user in ["did:plc:u1", "did:plc:u2", "did:plc:u3"] {
            save(user);
            let s = Session::oauth(user, pds.clone(), open()).await.unwrap();
            runs.extend((0..3).map(|_| write(&s, user)));
        }
        let ran = futures_util::future::join_all(runs).await;
        assert!(ran.iter().all(Result::is_ok), "{ran:?}");
        assert_eq!(*pds.refreshes.lock().unwrap(), 3);
        assert!(pds.writes.lock().unwrap().iter().all(|(did, repo)| did == repo));
        // a user's session cannot write another's repo (the PDS refuses: the token is theirs alone)
        let u1 = Session::oauth("did:plc:u1", pds.clone(), open()).await.unwrap();
        assert!(write(&u1, "did:plc:u2").await.unwrap_err().contains("not your repo"));
        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn blob_cids() {
        let cid = format!(
            "b{}",
            crypto::base32(&[&[0x01, 0x55, 0x12, 0x20][..], &[7; 32]].concat())
        );
        assert_eq!(blob_sha256(Some(&cid)).unwrap(), "07".repeat(32));
        assert_eq!(
            blob_sha256(Some(
                "bafyreie5737gdxlw5i64vzichcalba3z2v5n6icifvx5xytvske7mr3hpm"
            ))
            .unwrap_err(),
            "Blob CID must be a raw sha2-256 CIDv1"
        );
    }

    #[tokio::test]
    async fn pds_of_is_https_off_the_host_network() {
        let resolve = |endpoint: &str| {
            let fake = Fake(
                vec![("plc.directory/did:plc:a", 200, did_doc(endpoint))],
                Arc::default(),
            );
            async move { pds_of(&fake, "did:plc:a").await }
        };
        assert_eq!(resolve("https://pds.example").await.unwrap(), "https://pds.example");
        for bad in [
            "http://pds.example",
            "https://127.0.0.1",
            "https://localhost:3000",
            "https://169.254.169.254/latest",
            "https://10.1.2.3",
            "https://172.20.0.1",
            "https://192.168.1.1",
            "https://100.64.0.1",
            "https://0.0.0.0",
            "https://[::1]",
            "https://[fd00::1]",
            "https://[fe80::1]",
            "https://[::ffff:127.0.0.1]",
            "https://2130706433",
            "file:///etc/passwd",
            "nope",
        ] {
            assert!(resolve(bad).await.is_err(), "{bad}");
        }
        let seen: Seen = Arc::default();
        let fake = Fake(vec![], seen.clone());
        for did in ["did:web:localhost%3A8080", "did:web:127.0.0.1", "did:web:[::1]"] {
            assert!(pds_of(&fake, did).await.is_err(), "{did}");
        }
        assert!(seen.lock().unwrap().is_empty(), "nothing private was fetched");
        assert_eq!(public_url("http://127.0.0.1:2583", true).unwrap(), "http://127.0.0.1:2583");
    }

    #[tokio::test]
    async fn load_module_carries_didkey_and_refuses_odd_records() {
        const URI: &str = "at://did:plc:pub/com.fedfork.exec.wasm.module/m";
        let cid = "bafkreie5737gdxlw5i64vzichcalba3z2v5n6icifvx5xytvske7mr3hpm";
        let load = |record: Value| {
            let fake = Fake(
                vec![
                    ("plc.directory/did:plc:pub", 200, did_doc("https://pds.example")),
                    ("getRecord", 200, record),
                    ("getBlob", 200, json!("wasm")),
                ],
                Arc::default(),
            );
            async move { load_module(&fake, URI).await }
        };
        let record = |extra: Value| {
            let mut value = json!({ "abi": "emdash-wasm-json-v1", "module": { "ref": { "$link": cid } }, "calls": ["a.b.c"] });
            value.as_object_mut().unwrap().extend(extra.as_object().unwrap().clone());
            json!({ "value": value })
        };
        let module = load(record(json!({ "didkey": { "services": ["https://xrpc.fedproxy.com"] } })))
            .await
            .unwrap();
        assert_eq!(module.id.as_deref(), Some(URI));
        assert_eq!(module.didkey, Some(json!({ "services": ["https://xrpc.fedproxy.com"] })));
        assert_eq!(
            load(record(json!({ "calls": "a.b.c" }))).await.err().unwrap(),
            "module calls must be a list of strings"
        );
        assert_eq!(
            load(record(json!({ "anyRecords": [1] }))).await.err().unwrap(),
            "module anyRecords must be a list of strings"
        );
        // a reply that is not an object (or has no value) is an error, not a panic
        for odd in [json!([1, 2]), json!("x"), json!(null), json!({})] {
            assert!(load(odd).await.err().unwrap().starts_with("Unsupported module ABI"));
        }
    }

    #[test]
    fn service_aud_encodes_the_port() {
        assert_eq!(service_aud("https://svc.example/x"), "did:web:svc.example");
        assert_eq!(service_aud("https://svc.example:443/"), "did:web:svc.example");
        assert_eq!(service_aud("https://svc.example:8443/xrpc"), "did:web:svc.example%3A8443");
    }

    #[test]
    fn a_missing_service_auth_token_is_refused_not_sent() {
        assert_eq!(service_token(&json!({ "token": "jwt" })).unwrap(), "jwt");
        for answer in [json!({}), json!({ "token": "" }), json!({ "token": null }), json!({ "token": 5 })] {
            assert_eq!(service_token(&answer).unwrap_err(), "getServiceAuth answered no service-auth token", "{answer}");
        }
    }

    #[test]
    fn service_auth_goes_in_the_param_it_names_or_the_authorization_header() {
        let mut query = Request {
            params: json!({ "did": "d" }).as_object().cloned(),
            service_auth: Some(json!({ "param": "service_auth" })),
            ..Default::default()
        };
        apply_service_auth(&mut query, "jwt1");
        assert_eq!(Value::Object(query.params.unwrap()), json!({ "did": "d", "service_auth": "jwt1" }));
        assert!(query.service_auth.is_none() && query.headers.is_empty());
        let mut header = Request { service_auth: Some(json!({})), ..Default::default() };
        apply_service_auth(&mut header, "jwt1");
        assert_eq!(header.headers, vec![("authorization".to_string(), "Bearer jwt1".to_string())]);
        let mut none = Request::default();
        apply_service_auth(&mut none, "jwt1");
        assert!(none.headers.is_empty() && none.params.is_none());
    }

    #[tokio::test]
    async fn didkey_call_signs_a_subscribe_service_auth_for_listed_services_only() {
        struct Record(Mutex<Vec<Request>>);
        impl Call for Record {
            fn call(&self, _: String, _: Request) -> BoxFut<'_, Result<Option<Value>, String>> {
                Box::pin(async { Ok(None) })
            }
            fn subscribe(&self, _: String, request: Request) -> BoxFut<'_, Result<Sub, String>> {
                self.0.lock().unwrap().push(request);
                Box::pin(async { Err("recorded".into()) })
            }
        }
        struct NoFetch;
        impl Fetch for NoFetch {
            fn fetch(
                &self,
                _: &str,
                _: &str,
                _: Vec<(String, String)>,
                _: Option<Vec<u8>>,
            ) -> BoxFut<'_, Result<Response, String>> {
                Box::pin(async { Err("no network".into()) })
            }
        }
        let inner = Arc::new(Record(Mutex::new(vec![])));
        let key = DidKey::new(None).unwrap();
        let did = key.did.clone();
        let call = DidkeyCall { key, services: vec!["https://xrpc.fedproxy.com".into()], fetch: Arc::new(NoFetch), inner: inner.clone(), store: None };
        let req = |service: &str, auth: Value| Request { service: Some(service.into()), subscribe: true, service_auth: Some(auth), ..Default::default() };
        let nsid = "com.fedproxy.temp.xrpc.subscribe";
        for r in [req("https://xrpc.fedproxy.com", json!({ "param": "service_auth" })), req("https://xrpc.fedproxy.com", json!({})), req("https://other.example", json!({}))] {
            let _ = call.subscribe(nsid.into(), r).await;
        }
        let seen = inner.0.lock().unwrap();
        let token = seen[0].params.as_ref().unwrap()["service_auth"].as_str().unwrap().to_string();
        let claims: Value = serde_json::from_slice(&base64::Engine::decode(&base64::engine::general_purpose::URL_SAFE_NO_PAD, token.split('.').nth(1).unwrap()).unwrap()).unwrap();
        assert_eq!((claims["iss"].as_str(), claims["aud"].as_str(), claims["lxm"].as_str()), (Some(did.as_str()), Some("did:web:xrpc.fedproxy.com"), Some(nsid)));
        assert!(seen[0].service_auth.is_none());
        assert!(seen[1].headers[0].1.starts_with("Bearer ey"));
        assert_eq!(seen[2].service_auth, Some(json!({})), "not listed: left for the host's session token");
    }

    /// Two "processes" (stores on their own connections to one state dir) serving one did:key: the
    /// second relay subscribe is refused while the first is open, and goes through once it closes.
    #[tokio::test]
    async fn one_serving_instance_per_did_key_across_processes() {
        struct Open;
        impl Call for Open {
            fn call(&self, _: String, _: Request) -> BoxFut<'_, Result<Option<Value>, String>> {
                Box::pin(async { Ok(None) })
            }
            fn subscribe(&self, _: String, _: Request) -> BoxFut<'_, Result<Sub, String>> {
                let (tx, messages) = tokio::sync::mpsc::unbounded_channel();
                let (commands, mut rx) = tokio::sync::mpsc::unbounded_channel();
                tokio::spawn(async move {
                    let _ = tx.send(Ok(json!("registered")));
                    while let Some(c) = rx.recv().await {
                        if matches!(c, crate::run::SubCmd::Close) {
                            break; // closing drops tx: the stream ends
                        }
                    }
                });
                Box::pin(async move { Ok(Sub { messages, commands }) })
            }
        }
        let dir = std::env::temp_dir().join(format!("atlibs-host-serving-{}", crate::store::uuid()));
        let key = DidKey::new(None).unwrap();
        let serving = |store: Store| DidkeyCall { key: DidKey::new(Some(&key.jwk)).unwrap(), services: vec!["https://relay.example".into()], fetch: Arc::new(NoNet), inner: Arc::new(Open), store: Some(store) };
        let (a, b) = (serving(Store::open(&dir, "atlibs-server").unwrap()), serving(Store::open(&dir, "atlibs-server").unwrap()));
        let relay = || Request { service: Some("https://relay.example".into()), subscribe: true, service_auth: Some(json!({})), ..Default::default() };
        let nsid = "com.fedproxy.temp.xrpc.subscribe".to_string();
        let mut first = a.subscribe(nsid.clone(), relay()).await.unwrap();
        assert_eq!(first.messages.recv().await, Some(Ok(json!("registered"))));
        let refused = b.subscribe(nsid.clone(), relay()).await.err().unwrap();
        assert!(refused.contains("already serving in another run"), "{refused}");
        // not the relay: no lease needed
        assert!(b.subscribe(nsid.clone(), Request { service: Some("https://other.example".into()), subscribe: true, ..Default::default() }).await.is_ok());
        first.commands.send(crate::run::SubCmd::Close).unwrap();
        assert_eq!(first.messages.recv().await, None, "closed");
        let mut second = b.subscribe(nsid, relay()).await.unwrap();
        assert_eq!(second.messages.recv().await, Some(Ok(json!("registered"))));
        let _ = std::fs::remove_dir_all(&dir);
    }

    struct NoNet;
    impl Fetch for NoNet {
        fn fetch(&self, _: &str, _: &str, _: Vec<(String, String)>, _: Option<Vec<u8>>) -> BoxFut<'_, Result<Response, String>> {
            Box::pin(async { Err("no network".into()) })
        }
    }
}