//! Published modules and compositions, by their records: what `atlibs run`/`add` resolve, as a library
//! for hosts that embed atlibs (a CI runner, a server). The CLI adds lockfiles and local builds on top.

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

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

use crate::compose::{self, Composition};
use crate::crypto::ingress_host;
use crate::host::{DidkeyCall, HostCall, Session};
use crate::oauth::ScopeCall;
use crate::run::Call;
use crate::store::Store;
use crate::transport::{TransportCall, Transports};
use crate::host::{Fetch, blob_sha256, json_response, public_url, xrpc_url};
use crate::BoxFut;
use crate::run::{Module, module_key_ok, module_pin};
use crate::wasm::sha256_hex;

pub const MODULE: &str = "com.fedfork.exec.wasm.module";
pub const CLIENT: &str = "com.fedfork.exec.oauth.clientMetadata";
pub const COMPOSITION: &str = "com.fedfork.exec.composition";
pub const ABI: &str = "emdash-wasm-json-v1";

/// HTTP plus where DIDs and handles resolve (env, so tests and mirrors can point elsewhere).
#[derive(Clone)]
pub struct Net {
    pub fetch: Arc<dyn Fetch>,
    pub plc: String,
    pub resolver: String,
    pub allow_private: bool,
}

pub fn split_uri(uri: &str) -> Option<(&str, &str, &str)> {
    let mut parts = uri.strip_prefix("at://")?.split('/');
    match (parts.next(), parts.next(), parts.next(), parts.next()) {
        (Some(did), Some(c), Some(r), None) if did.starts_with("did:") && !c.is_empty() && !r.is_empty() => Some((did, c, r)),
        _ => None,
    }
}

fn params(pairs: &[(&str, &str)]) -> Map<String, Value> {
    pairs.iter().map(|(k, v)| (k.to_string(), json!(v))).collect()
}

impl Net {
    /// plc.directory and public.api.bsky.app, public addresses only.
    pub fn new(fetch: Arc<dyn Fetch>) -> Self {
        Self { fetch, plc: "https://plc.directory".into(), resolver: "https://public.api.bsky.app".into(), allow_private: false }
    }

    async fn get(&self, nsid: &str, url: &str) -> Result<Value, String> {
        let url = public_url(url, self.allow_private)?;
        json_response(nsid, &self.fetch.fetch("GET", &url, vec![], None).await?).map_err(|e| e.message)
    }

    pub async fn did_doc(&self, did: &str) -> Result<Value, String> {
        let url = if did.starts_with("did:plc:") {
            format!("{}/{did}", self.plc.trim_end_matches('/'))
        } 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"));
        };
        self.get(did, &url).await.map_err(|e| format!("resolving {did}: {e}"))
    }

    pub async fn pds(&self, did: &str) -> Result<String, String> {
        let doc = self.did_doc(did).await?;
        let pds = doc["service"]
            .as_array()
            .into_iter()
            .flatten()
            .find(|s| s["id"] == "#atproto_pds" || s["id"] == format!("{did}#atproto_pds"))
            .and_then(|s| s["serviceEndpoint"].as_str())
            .ok_or_else(|| format!("{did} has no PDS"))?;
        public_url(pds, self.allow_private)
    }

    /// handle -> DID, checked both ways: the DID document must claim the handle.
    // ponytail: resolveHandle at ATLIBS_HANDLE_RESOLVER, no DNS TXT of our own; the alsoKnownAs check
    // keeps a lying resolver from substituting another account.
    pub async fn resolve_handle(&self, handle: &str) -> Result<String, String> {
        let nsid = "com.atproto.identity.resolveHandle";
        let found = self.get(nsid, &xrpc_url(&self.resolver, nsid, Some(&params(&[("handle", handle)])))?).await.map_err(|e| format!("resolving {handle}: {e}"))?;
        let did = found["did"].as_str().ok_or_else(|| format!("{handle} did not resolve"))?.to_string();
        let doc = self.did_doc(&did).await?;
        let claimed = doc["alsoKnownAs"].as_array().into_iter().flatten().any(|a| a.as_str() == Some(&format!("at://{handle}")));
        if !claimed {
            return Err(format!("{handle} resolves to {did}, whose DID document does not claim it"));
        }
        Ok(did)
    }

    /// DID -> its handle, for display: the DID document's first alsoKnownAs at:// handle, when that handle
    /// resolves back to the DID (resolve_handle). None otherwise; integrity never rests on it.
    pub async fn handle_of(&self, did: &str) -> Option<String> {
        let doc = self.did_doc(did).await.ok()?;
        let handle = doc["alsoKnownAs"].as_array()?.iter().find_map(|a| a.as_str()?.strip_prefix("at://").filter(|h| !h.starts_with("did:")).map(String::from))?;
        (self.resolve_handle(&handle).await.ok()? == did).then_some(handle)
    }

    /// (cid, value) of a record; None when the PDS has no such record.
    pub async fn record(&self, uri: &str) -> Result<Option<(String, Value)>, String> {
        let (did, collection, rkey) = split_uri(uri).ok_or_else(|| format!("{uri} is not an at:// record URI"))?;
        let url = xrpc_url(&self.pds(did).await?, "com.atproto.repo.getRecord", Some(&params(&[("repo", did), ("collection", collection), ("rkey", rkey)])))?;
        let response = self.fetch.fetch("GET", &public_url(&url, self.allow_private)?, vec![], None).await?;
        match json_response("com.atproto.repo.getRecord", &response) {
            Ok(mut body) => Ok(Some((body["cid"].as_str().unwrap_or_default().to_string(), body["value"].take()))),
            Err(e) if response.status == 404 || e.error.as_deref() == Some("RecordNotFound") => Ok(None),
            Err(e) => Err(format!("{uri}: {}", e.message)),
        }
    }

    pub async fn blob(&self, did: &str, cid: &str) -> Result<Vec<u8>, String> {
        let url = xrpc_url(&self.pds(did).await?, "com.atproto.sync.getBlob", Some(&params(&[("did", did), ("cid", cid)])))?;
        let response = self.fetch.fetch("GET", &public_url(&url, self.allow_private)?, vec![], None).await?;
        if !(200..300).contains(&response.status) {
            return Err(format!("blob {cid} of {did} not found ({})", response.status));
        }
        Ok(response.body)
    }
}

/// A session's com.fedfork.exec.oauth.clientMetadata record.
#[derive(Clone)]
pub struct Client {
    pub rkey: String,
    pub record: Value,
    pub uri: Option<String>,
    pub cid: Option<String>,
}

/// One module: bytes, its module.json fields, and its session's record.
#[derive(Clone)]
pub struct Lib {
    pub key: String,
    /// what was asked for (<handle>/<lib>), kept in the lockfile
    pub reference: Option<String>,
    pub uri: Option<String>,
    pub cid: Option<String>,
    pub wasm: Vec<u8>,
    pub sha256: String,
    /// module.json fields; published ones carry their uri and the session's rkey
    pub spec: Value,
    pub client: Option<Client>,
    /// the run identity: its at:// URI, else the wasm's absolute path
    pub id: String,
    /// Pinned (vendored in a lockfile, or a ref @pin): the directory whose atlibs.lock.json alone may
    /// supply its hook defaults (default_lib).
    pub pinned: Option<PathBuf>,
}

impl Lib {
    /// What its pins match: run::module_pin of its wasm and its record's grants.
    pub fn pin(&self) -> String {
        module_pin(&self.wasm, &self.spec)
    }
}

/// The rkey publish also writes a module's record at, `<rkey>.<first 32 hex of its pin>`, so a republish leaves
/// every pinned version in place. None for a pin that is not 32 to 64 hex (a record CID).
pub fn version_rkey(rkey: &str, pin: &str) -> Option<String> {
    ((32..=64).contains(&pin.len()) && pin.bytes().all(|b| b.is_ascii_hexdigit())).then(|| format!("{rkey}.{}", pin[..32].to_ascii_lowercase()))
}

/// The rkey publish also writes a composition's record at, `<rkey>.<its record CID>`: what a CID pin (a binding,
/// `@<cid>`) loads, left in place by a republish. None for a CID that is not 1 to 100 letters and digits.
pub fn composition_version(rkey: &str, cid: &str) -> Option<String> {
    ((1..=100).contains(&cid.len()) && cid.bytes().all(|b| b.is_ascii_alphanumeric())).then(|| format!("{rkey}.{cid}"))
}

/// A composition record: (its CID, it). Pinned by `cid`, the record at composition_version, which must be that CID.
pub async fn composition(net: &Net, uri: &str, cid: Option<&str>) -> Result<(String, Value), String> {
    let (did, _, rkey) = split_uri(uri).filter(|(_, c, _)| *c == COMPOSITION).ok_or_else(|| format!("{uri} is not a {COMPOSITION} record"))?;
    let at = match cid {
        Some(c) => format!("at://{did}/{COMPOSITION}/{}", composition_version(rkey, c).ok_or_else(|| format!("{uri}: {c} is not a record CID"))?),
        None => uri.to_string(),
    };
    let (found, value) = net.record(&at).await?.ok_or_else(|| if cid.is_some() { format!("{uri}: version {} not published ({at} not found)", cid.unwrap_or_default()) } else { format!("{uri} not found") })?;
    if let Some(pinned) = cid.filter(|c| *c != found) {
        return Err(format!("{uri}: pinned to {pinned}, but it is record {found}"));
    }
    Ok((found, value))
}

/// A published module: its record, its blob (sha256 checked against the CID), its session record. With a
/// pin (check_pin) the record at version_rkey, whose identity is still `uri`.
pub async fn fetch_lib(net: &Net, uri: &str, reference: Option<&str>, pin: Option<&str>) -> Result<Lib, String> {
    let (did, _, rkey) = split_uri(uri).filter(|(_, c, _)| *c == MODULE).ok_or_else(|| format!("{uri} is not a {MODULE} record"))?;
    let at = pin.and_then(|p| version_rkey(rkey, p)).map_or_else(|| uri.to_string(), |v| format!("at://{did}/{MODULE}/{v}"));
    let (cid, value) = net.record(&at).await?.ok_or_else(|| if at == uri { format!("{uri} not found") } else { format!("{uri}: pin {} not published ({at} not found)", pin.unwrap_or_default()) })?;
    if value["abi"] != ABI {
        return Err(format!("{uri}: unsupported module ABI {}", value["abi"]));
    }
    let blob_cid = value.pointer("/module/ref/$link").and_then(Value::as_str);
    let expected = blob_sha256(blob_cid)?;
    let wasm = net.blob(did, blob_cid.unwrap_or_default()).await?;
    let sha256 = sha256_hex(&wasm);
    if sha256 != expected {
        return Err(format!("{uri}: blob does not match its CID"));
    }
    let mut spec = spec_of(&value);
    spec.insert("uri".into(), json!(uri));
    let mut client = None;
    if let Some(session_uri) = value.pointer("/session/uri").and_then(Value::as_str) {
        let (_, _, session) = split_uri(session_uri).filter(|(_, c, _)| *c == CLIENT).ok_or_else(|| format!("{uri}: session is not a {CLIENT} record"))?;
        let (session_cid, record) = net.record(session_uri).await?.ok_or_else(|| format!("{uri}: session record {session_uri} not found"))?;
        if Some(session_cid.as_str()) != value.pointer("/session/cid").and_then(Value::as_str) {
            eprintln!("warning: {uri}: session record changed since the module was published");
        }
        spec.insert("session".into(), json!(session));
        client = Some(Client { rkey: session.into(), record, uri: Some(session_uri.into()), cid: Some(session_cid) });
    }
    let lib = Lib { key: rkey.into(), reference: reference.map(String::from), uri: Some(uri.into()), cid: Some(cid), wasm, sha256, spec: Value::Object(spec), client, id: uri.into(), pinned: None };
    check_pin(&lib, pin)?;
    Ok(lib)
}

/// The module.json fields a host reads, and nothing else: a record (or module.json) cannot choose its
/// own identity or ledger owner (`id`, `owner`, `uri`); those come from where it was loaded.
pub const SPEC_FIELDS: [&str; 12] = ["name", "description", "actions", "config", "calls", "anyRecords", "didkey", "session", "hooks", "implements", "anyHooks", "http"];

pub fn spec_of(value: &Value) -> Map<String, Value> {
    SPEC_FIELDS.iter().filter_map(|k| value.get(*k).map(|v| (k.to_string(), v.clone()))).collect()
}

/// The at:// URI a remote ref names (a module, else a composition of that rkey) and its pin.
pub async fn resolve(net: &Net, reference: &str) -> Result<(String, Option<String>), String> {
    if reference.starts_with("at://") {
        return Ok((reference.to_string(), None));
    }
    let (path, pin) = match reference.split_once('@') {
        Some((p, pin)) => (p, Some(pin.to_string())),
        None => (reference, None),
    };
    let (handle, lib) = path.split_once('/').ok_or_else(|| format!("{reference} is not <handle>/<lib>"))?;
    let did = if handle.starts_with("did:") { handle.to_string() } else { net.resolve_handle(handle).await? };
    for collection in [MODULE, COMPOSITION] {
        let uri = format!("at://{did}/{collection}/{lib}");
        if net.record(&uri).await?.is_some() {
            return Ok((uri, pin));
        }
    }
    Err(format!("{handle} ({did}) publishes no module or composition named {lib}"))
}

/// How a record is shown to people: `<handle>/<rkey>` (Net::handle_of), else `<did>/<rkey>`; the at:// URI
/// for a collection a `<handle>/<rkey>` does not name (resolve: modules and compositions).
pub async fn named(net: &Net, uri: &str) -> String {
    match split_uri(uri) {
        Some((did, collection, rkey)) if collection == MODULE || collection == COMPOSITION => format!("{}/{rkey}", net.handle_of(did).await.unwrap_or_else(|| did.into())),
        _ => uri.into(),
    }
}

/// `@pin`: the record CID, or at least 32 hex characters of its pin (module_pin: the wasm and the record's
/// grants, so a record whose calls or hooks changed no longer matches).
pub fn check_pin(lib: &Lib, pin: Option<&str>) -> Result<(), String> {
    let hex = |p: &str| p.bytes().all(|b| b.is_ascii_hexdigit());
    match pin {
        None => Ok(()),
        Some(p) if lib.cid.as_deref() == Some(p) => Ok(()),
        Some(p) if hex(p) && p.len() < 32 => Err(format!("{}: pin {p} is too short: give the record CID, or at least 32 hex characters of the pin", lib.key)),
        Some(p) if hex(p) && p.len() <= 64 && lib.pin().starts_with(&p.to_ascii_lowercase()) => Ok(()),
        Some(p) => Err(format!("{}: pinned to {p}, but it is pin {} (record {})", lib.key, lib.pin(), lib.cid.as_deref().unwrap_or("-"))),
    }
}

/// A lib as a runner takes it: its module, with its identity (id, key) and its session's scope.
pub fn module_of(lib: &Lib) -> Result<Module, String> {
    let mut module = Module::from_manifest(lib.wasm.clone(), &lib.spec)?;
    module.sha256 = Some(lib.sha256.clone());
    module.id = Some(lib.id.clone());
    module.key = Some(lib.key.clone());
    module.scope = lib.client.as_ref().and_then(|c| Some((c.rkey.clone(), c.record["clientMetadata"]["scope"].as_str()?.to_string())));
    Ok(module)
}

/// A published composition's instance module: an at:// URI, or a key of its publisher's; at its pin when pinned.
pub async fn instance_lib(net: &Net, reference: &str, publisher: &str, pin: Option<&str>) -> Result<Lib, String> {
    if reference.starts_with("at://") {
        return fetch_lib(net, reference, None, pin).await;
    }
    if !module_key_ok(reference) {
        return Err(format!("{reference}: a published composition names its publisher's modules by key, or by at:// URI"));
    }
    fetch_lib(net, &format!("at://{publisher}/{MODULE}/{reference}"), None, pin).await
}

/// Where a composition's module refs come from: (ref, its instance's pin) -> lib.
pub type LibOf = Arc<dyn Fn(String, Option<String>) -> BoxFut<'static, Result<Lib, String>> + Send + Sync>;

/// compose::load with `lib_of` finding each module (pins checked, hook defaults loaded as siblings), and
/// each ref it loaded with its lib.
pub async fn load_with(manifest: &Value, lib_of: LibOf) -> Result<(Composition, Vec<(String, Lib)>), String> {
    let found: Arc<Mutex<Vec<(String, Lib)>>> = Arc::default();
    let loader: compose::ModuleLoader = {
        let found = found.clone();
        Arc::new(move |reference: String, pin: Option<String>| {
            let (found, lib_of) = (found.clone(), lib_of.clone());
            Box::pin(async move {
                let lib = lib_of(reference.clone(), pin).await?;
                let module = module_of(&lib)?;
                found.lock().unwrap().push((reference, lib));
                Ok(module)
            })
        })
    };
    let comp = compose::load(manifest, &loader).await?;
    let found = found.lock().unwrap().clone();
    Ok((comp, found))
}

/// The composition record at `uri` (at its version `cid` when given: composition), loaded with its modules from
/// their records. -> (its CID, it, each ref with its lib).
pub async fn published(net: &Net, uri: &str, cid: Option<&str>) -> Result<(String, Composition, Vec<(String, Lib)>), String> {
    let (found, value) = composition(net, uri, cid).await?;
    let did = split_uri(uri).map(|(d, _, _)| d).unwrap_or_default();
    let (net, publisher) = (net.clone(), did.to_string());
    let lib_of: LibOf = Arc::new(move |reference: String, pin: Option<String>| {
        let (net, publisher) = (net.clone(), publisher.clone());
        Box::pin(async move { instance_lib(&net, &reference, &publisher, pin.as_deref()).await })
    });
    let (comp, libs) = load_with(&value, lib_of).await?;
    Ok((found, comp, libs))
}

/// The call one module runs on: XRPC with `session` (held to the module's own session scope when signed
/// in), attest when it lists calls (one that lists none may call anything, so gets no signing key), its
/// did:key (sealed under `name`: its identity, or composition#instance; an OAuth user's under "<did> <name>"),
/// and `transports`.
pub async fn module_call(store: &Store, fetch: Arc<dyn Fetch>, session: Option<Arc<Session>>, transports: &Transports, module: &Module, name: &str) -> Result<Arc<dyn Call>, String> {
    let lists_calls = module.calls.as_ref().is_some_and(|c| !c.is_empty());
    let attest = (session.is_some() && lists_calls).then(|| store.clone());
    let mut call: Arc<dyn Call> = Arc::new(HostCall { fetch: fetch.clone(), session: session.clone(), attest });
    if session.is_some() {
        let key = module.key.clone().unwrap_or_else(|| name.to_string());
        call = Arc::new(ScopeCall { module: key, session: module.scope.clone(), inner: call });
    }
    if let (Some(didkey), true) = (&module.didkey, lists_calls) {
        let name = match &session {
            Some(s) if s.is_oauth() => format!("{} {name}", s.did()),
            _ => name.to_string(),
        };
        let key = store.didkey(&name).await?;
        let services: Vec<String> = didkey.get("services").and_then(Value::as_array).map(|s| s.iter().filter_map(|x| x.as_str().map(String::from)).collect()).unwrap_or_default();
        for service in &services {
            eprintln!("{}: https://{}/", key.did, ingress_host(&key.did, service));
        }
        call = Arc::new(DidkeyCall { key, services, fetch, inner: call, store: Some(store.clone()) });
    }
    Ok(Arc::new(TransportCall { transports: transports.clone(), listed: lists_calls, inner: call }))
}