// atlibs publish: modules, clientMetadata records, web tiles and compositions, as the ATPROTO_HANDLE
// account (app password; one createSession, then the saved session refreshes).
//
//   atlibs publish [<name>...] [--dry-run]           only what changed since the last publish (cache_root, else the
//                                                   live records): each module as `publish module`, each record as
//                                                   `publish records`; a status table
//   atlibs publish module <dir|name>... [--no-build] refuses uncommitted sources (--allow-dirty); BUILD (reproducible), every source file as a
//                                                   com.fedfork.exec.source.file record (by sha256, reused),
//                                                   a source manifest (git origin + commit), the blob, then
//                                                   the com.fedfork.exec.wasm.module record (rkey = the
//                                                   directory name); writes records/wasm/<name>.json
//   atlibs publish records [file...]                each record (default records/*.json) by its $type, rkey =
//                                                   the file name
//   atlibs publish tile <dir> <name>                a web tile (https://dasl.ing/tiles.html): ing.dasl.masl
//   atlibs publish composition <file> [--rkey k]    com.fedfork.exec.composition; module keys become this
//                                                   account's at://did/com.fedfork.exec.wasm.module/<key>
//   atlibs publish delete <at-uri|collection/rkey>  remove this account's records (an old module, its manifest)
// Every record is checked against the lexicons first: the CLI's own and the repo's lexicons/ (lexicon.rs).

use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::process::Command;
use std::sync::Arc;

use atlibs_host::crypto::{base32, unbase32};
use atlibs_host::host::{Session, blob_sha256};
use atlibs_host::registry::{composition_version, version_rkey};
use atlibs_host::run::pin_of_sha256;
use atlibs_host::{compose, iso_now, now_ms};
use serde_json::{Map, Value, json};
use sha2::{Digest, Sha256};

use crate::lexicon::Lexicons;
use crate::registry::{ABI, CLIENT, COMPOSITION, Lib, MODULE, find_up, local_lib, module_dir, spec_of, write_json};
use crate::{Args, Cli};

const FILE: &str = "com.fedfork.exec.source.file";
const MANIFEST: &str = "com.fedfork.exec.source.manifest";
const MASL: &str = "ing.dasl.masl";
// The reference build of a cargo workspace's wasm modules, in the container (registry::container_build, -w /src):
// local paths are remapped out of the binaries so the sha256 matches a rebuild from the published sources. The
// manifest's build.command, so `atlibs verify` runs exactly this; it assumes nothing of the repo but its workspace.
const BUILD: &str = r#"RUSTFLAGS="--cfg getrandom_backend=\"unsupported\" --remap-path-prefix=${CARGO_HOME:-$HOME/.cargo}=/cargo --remap-path-prefix=$PWD=/wasm" cargo build --release --locked --target wasm32-unknown-unknown"#;

pub struct Publisher {
    pub session: Arc<Session>,
    pub did: String,
    lexicons: Lexicons,
}

fn git(root: &Path, args: &[&str]) -> Result<String, String> {
    let out = Command::new("git").arg("-C").arg(root).args(args).output().map_err(|e| format!("git: {e}"))?;
    if !out.status.success() {
        return Err(format!("git {}: {}", args.join(" "), String::from_utf8_lossy(&out.stderr).trim()));
    }
    Ok(String::from_utf8_lossy(&out.stdout).trim().to_string())
}

/// The repository a path is in (where lexicons/, records/ and target/ are), else the path itself.
pub fn root_of(path: &Path) -> PathBuf {
    git(path, &["rev-parse", "--show-toplevel"]).map(PathBuf::from).unwrap_or_else(|_| path.to_path_buf())
}

impl Publisher {
    pub async fn new(cli: &Cli, root: &Path) -> Result<Self, String> {
        let session = cli.require_session().await?;
        let lexicons = Lexicons::load(find_up(root, "lexicons").as_deref())?;
        Ok(Self { did: session.did(), session, lexicons })
    }

    /// {uri, cid, value} of one of this account's records; None when there is none.
    pub async fn get(&self, collection: &str, rkey: &str) -> Result<Option<Value>, String> {
        let params: Map<String, Value> = [("repo", &self.did), ("collection", &collection.to_string()), ("rkey", &rkey.to_string())].into_iter().map(|(k, v)| (k.to_string(), json!(v))).collect();
        match self.session.xrpc("com.atproto.repo.getRecord", Some(&params), None, vec![]).await {
            Ok(found) => Ok(Some(json!({ "uri": found["uri"], "cid": found["cid"], "value": found["value"] }))),
            Err(e) if e.contains("RecordNotFound") || e.contains("Could not locate record") => Ok(None),
            Err(e) => Err(e),
        }
    }

    /// createRecord or putRecord, after the lexicon check; {uri, cid}. A module is written at its version first,
    /// a composition after (its key is the CID the write gives): what pins load, left in place by republishes.
    pub async fn write(&self, method: &str, collection: &str, rkey: &str, record: &Value) -> Result<Value, String> {
        self.lexicons.validate(record).map_err(|e| format!("{collection}/{rkey} is not valid: {e}"))?;
        if collection == MODULE {
            self.put_version(collection, &module_version(rkey, record)?, record).await?;
        }
        let written = self.raw(method, collection, rkey, record).await?;
        if collection == COMPOSITION {
            self.put_version(collection, &composition_version(rkey, written["cid"].as_str().unwrap_or_default()).ok_or("the PDS gave no CID")?, record).await?;
        }
        Ok(written)
    }

    async fn raw(&self, method: &str, collection: &str, rkey: &str, record: &Value) -> Result<Value, String> {
        let input = json!({ "repo": self.did, "collection": collection, "rkey": rkey, "record": record });
        let out = self.session.xrpc(&format!("com.atproto.repo.{method}"), None, Some(&input), vec![]).await?;
        Ok(json!({ "uri": out["uri"], "cid": out["cid"] }))
    }

    /// A record at its version key, as is; a composition's must come out the CID the key names.
    async fn put_version(&self, collection: &str, version: &str, record: &Value) -> Result<Value, String> {
        let written = self.raw("putRecord", collection, version, record).await?;
        let cid = written["cid"].as_str().unwrap_or_default();
        if collection == COMPOSITION && !version.ends_with(&format!(".{cid}")) {
            return Err(format!("{collection}/{version}: written as record {cid}, not the CID its key names"));
        }
        Ok(written)
    }
}

/// A module record's version key (registry::version_rkey): its pin from the blob CID's sha256 and its grants.
fn module_version(rkey: &str, record: &Value) -> Result<String, String> {
    let sha = blob_sha256(record.pointer("/module/ref/$link").and_then(Value::as_str))?;
    version_rkey(rkey, &pin_of_sha256(&sha, record)).ok_or_else(|| format!("{rkey}: no pin"))
}

/// The com.fedfork.exec.wasm.module record for a module.json: optional fields are left out when it
/// has none (a policy module has no session).
pub fn module_record(spec: &Value, blob: &Value, did: &str, session_cid: &str, source: &Value, created_at: &str) -> Value {
    let mut record = json!({ "$type": MODULE, "name": spec["name"], "module": blob, "abi": ABI, "source": source, "actions": spec.get("actions").cloned().unwrap_or(json!([])), "createdAt": created_at });
    if let Some(d) = spec.get("description") {
        record["description"] = d.clone();
    }
    if let Some(s) = spec["session"].as_str() {
        record["session"] = json!({ "uri": format!("at://{did}/{CLIENT}/{s}"), "cid": session_cid });
    }
    for key in ["config", "calls", "anyRecords", "didkey", "hooks", "implements", "anyHooks", "http"] {
        if let Some(v) = spec.get(key) {
            record[key] = v.clone();
        }
    }
    record
}

fn source_file(path: &str) -> bool {
    let name = path.rsplit('/').next().unwrap_or(path);
    !path.ends_with(".wasm") && !["module.json", "test.mjs", "harness.mjs", ".gitignore"].contains(&name)
}

/// rkey -> {uri, cid, value} of every record of this account in a collection (listRecords, paged).
async fn list(publisher: &Publisher, collection: &str) -> Result<HashMap<String, Value>, String> {
    let mut out = HashMap::new();
    let mut cursor: Option<String> = None;
    loop {
        let mut params: Map<String, Value> = [("repo", json!(publisher.did)), ("collection", json!(collection)), ("limit", json!("100"))].into_iter().map(|(k, v)| (k.to_string(), v)).collect();
        if let Some(c) = &cursor {
            params.insert("cursor".into(), json!(c));
        }
        let page = publisher.session.xrpc("com.atproto.repo.listRecords", Some(&params), None, vec![]).await?;
        let records = page["records"].as_array().cloned().unwrap_or_default();
        for r in &records {
            if let Some(rkey) = r["uri"].as_str().and_then(|u| u.rsplit('/').next()) {
                out.insert(rkey.to_string(), json!({ "uri": r["uri"], "cid": r["cid"], "value": r["value"] }));
            }
        }
        cursor = page["cursor"].as_str().map(String::from);
        if records.is_empty() || cursor.is_none() {
            return Ok(out);
        }
    }
}

/// Every workspace source file as a com.fedfork.exec.source.file record (rkey = sha256): the ones
/// this account already has are reused, the rest created. The same list for every module.
async fn source_files(publisher: &Publisher, root: &Path, sources: &[String], now: &str) -> Result<Vec<Value>, String> {
    let known = list(publisher, FILE).await?;
    let mut files = vec![];
    for path in source_paths(root, sources)? {
        let path = path.as_str();
        let bytes = std::fs::read(root.join(path)).map_err(|e| format!("{path}: {e}"))?;
        let sha = atlibs_host::wasm::sha256_hex(&bytes);
        let reference = match known.get(&sha) {
            Some(r) => json!({ "uri": r["uri"], "cid": r["cid"] }),
            None => {
                let text = String::from_utf8(bytes.clone()).map_err(|_| format!("{path} is not UTF-8 text"))?;
                publisher.write("createRecord", FILE, &sha, &json!({ "$type": FILE, "sha256": sha, "size": bytes.len(), "text": text, "createdAt": now })).await?
            }
        };
        files.push(json!({ "path": path, "file": reference }));
    }
    Ok(files)
}

/// What the build reads, as git pathspecs relative to `root` (the repo, which must be the cargo workspace root):
/// Cargo.toml, Cargo.lock and .cargo there, every workspace member (`cargo metadata`; path dependencies under the
/// root are members too) and the module directories, so a repo with no workspace still has its modules checked.
/// The manifest carries these files, so anyone can rebuild.
fn sources(root: &Path, dirs: &[PathBuf]) -> Vec<String> {
    let mut out: Vec<String> = ["Cargo.toml", "Cargo.lock", ".cargo"].map(String::from).to_vec();
    let root = root.canonicalize().unwrap_or_else(|_| root.to_path_buf());
    let metadata = Command::new("cargo").args(["metadata", "--no-deps", "--format-version", "1", "--manifest-path"]).arg(root.join("Cargo.toml")).output();
    let members = metadata.ok().filter(|o| o.status.success()).and_then(|o| serde_json::from_slice::<Value>(&o.stdout).ok());
    let manifests = members.iter().flat_map(|m| m["packages"].as_array().cloned().unwrap_or_default()).filter_map(|p| p["manifest_path"].as_str().map(PathBuf::from));
    for dir in manifests.filter_map(|m| m.parent().map(Path::to_path_buf)).chain(dirs.iter().filter_map(|d| d.canonicalize().ok())) {
        if let Ok(rel) = dir.strip_prefix(&root) {
            // a package at the root reads anything in the repo
            out.push(if rel.as_os_str().is_empty() { ".".into() } else { rel.to_string_lossy().into_owned() });
        }
    }
    out.sort();
    out.dedup();
    out
}

/// The tracked source files of `sources`, as the manifest lists them.
fn source_paths(root: &Path, sources: &[String]) -> Result<Vec<String>, String> {
    let args: Vec<&str> = ["ls-files", "--"].into_iter().chain(sources.iter().map(String::as_str)).collect();
    Ok(git(root, &args)?.lines().filter(|p| source_file(p)).map(String::from).collect())
}

/// BUILD in the reference container (registry::container_build) on a copy of exactly the source files the
/// manifest lists, as `verify` rebuilds them (same files, same /src, same image): every module's wasm is then
/// copied to the repo's target/wasm32-unknown-unknown/release.
fn build(root: &Path, sources: &[String]) -> Result<(), String> {
    // the build is --locked: without a lockfile it fails in the container, after pulling the image
    if !root.join("Cargo.lock").is_file() {
        return Err(format!("no Cargo.lock in {}: build once (cargo wasm) and commit it", root.display()));
    }
    let work = std::env::temp_dir().join(format!("atlibs-build-{}", atlibs_host::store::uuid()));
    let built = (|| {
        for path in &source_paths(root, sources)? {
            let to = work.join(path);
            std::fs::create_dir_all(to.parent().expect("a file")).map_err(|e| format!("{}: {e}", to.display()))?;
            std::fs::copy(root.join(path), &to).map_err(|e| format!("{path}: {e}"))?;
        }
        crate::registry::container_build(&crate::registry::engine()?, &work, &rustc_version(), BUILD)?;
        let release = Path::new("target/wasm32-unknown-unknown/release");
        std::fs::create_dir_all(root.join(release)).map_err(|e| e.to_string())?;
        for entry in std::fs::read_dir(work.join(release)).map_err(|e| format!("{}: {e}", release.display()))? {
            let from = entry.map_err(|e| e.to_string())?.path();
            if from.extension().is_some_and(|x| x == "wasm") {
                std::fs::copy(&from, root.join(release).join(from.file_name().expect("a file"))).map_err(|e| format!("{}: {e}", from.display()))?;
            }
        }
        Ok(())
    })();
    let _ = std::fs::remove_dir_all(&work);
    built
}

/// `rustc --version`: the build image's version and the manifest's build.version.
fn rustc_version() -> String {
    Command::new("rustc").arg("--version").output().map(|o| String::from_utf8_lossy(&o.stdout).trim().to_string()).unwrap_or_default()
}

/// Uncommitted changes in the sources: the manifest names a commit, so what is built and uploaded must be it.
fn dirty(root: &Path, sources: &[String]) -> Result<String, String> {
    let args: Vec<&str> = ["status", "--porcelain", "--"].into_iter().chain(sources.iter().map(String::as_str)).collect();
    git(root, &args)
}

/// The CID of a lib's session (clientMetadata) record on this account; "" for a lib without one.
async fn session_cid(publisher: &Publisher, lib: &Lib) -> Result<String, String> {
    let Some(s) = lib.spec["session"].as_str() else { return Ok(String::new()) };
    let found = publisher.get(CLIENT, s).await?.ok_or_else(|| format!("{}: no {CLIENT} record {s} on {}: `atlibs publish records` first", lib.key, publisher.did))?;
    Ok(found["cid"].as_str().unwrap_or_default().to_string())
}

/// Publishes each (lib, its session CID): a source manifest (origin = HEAD), the blob, the module record,
/// with records/wasm-style snapshots under `out`. The shared source file records first (an error there
/// fails them all), then one {uri, cid, sha256, source} or error per lib.
async fn upload(publisher: &Publisher, root: &Path, sources: &[String], libs: &[(&Lib, String)], out: &Path) -> Result<Vec<Result<Value, String>>, String> {
    let now = iso_now();
    let files = source_files(publisher, root, sources, &now).await?;
    let rustc = rustc_version();
    let repository = std::env::var("REPOSITORY").ok().filter(|r| !r.is_empty()).map_or_else(|| git(root, &["remote", "get-url", "origin"]), Ok)?;
    let commit = git(root, &["rev-parse", "HEAD"])?;
    let mut results = vec![];
    for (lib, session_cid) in libs {
        let result: Result<Value, String> = async {
            let key = &lib.key;
            let manifest = json!({
                "$type": MANIFEST,
                "name": lib.spec["name"],
                "description": lib.spec["description"],
                "language": "rust",
                "files": files,
                "build": { "tool": "cargo", "version": rustc, "command": BUILD, "artifact": format!("target/wasm32-unknown-unknown/release/{}.wasm", key.replace('-', "_")) },
                "output": { "sha256": lib.sha256, "size": lib.wasm.len() },
                "origin": { "repository": repository, "commit": commit },
                "createdAt": now,
            });
            let source = publisher.write("createRecord", MANIFEST, &format!("{key}.{}", now_ms()), &manifest).await?;
            write_json(&out.join(format!("{key}.manifest.json")), &manifest)?;
            let blob = publisher.session.upload_blob(lib.wasm.clone(), "application/wasm").await?["blob"].clone();
            let file = out.join(format!("{key}.json"));
            let created_at = std::fs::read(&file).ok().and_then(|b| serde_json::from_slice::<Value>(&b).ok()).and_then(|r| r["createdAt"].as_str().map(String::from)).unwrap_or_else(|| now.clone());
            let record = module_record(&lib.spec, &blob, &publisher.did, session_cid, &source, &created_at);
            let written = publisher.write("putRecord", MODULE, key, &record).await?;
            write_json(&file, &record)?;
            Ok(json!({ "uri": written["uri"], "cid": written["cid"], "sha256": lib.sha256, "source": source["uri"] }))
        }
        .await;
        results.push(result);
    }
    Ok(results)
}

async fn module(cli: &Cli, args: &Args) -> Result<(), String> {
    args.only(&["no-build", "out", "allow-dirty"], "publish module")?;
    if args.pos.is_empty() {
        return Err(crate::usage("publish module"));
    }
    let cwd = std::env::current_dir().map_err(|e| e.to_string())?;
    let dirs = args.pos.iter().map(|name| module_dir(&cwd, name).ok_or_else(|| format!("no module directory {name} (module.json)"))).collect::<Result<Vec<_>, _>>()?;
    let root = root_of(&dirs[0]);
    let sources = sources(&root, &dirs);
    let dirty = dirty(&root, &sources)?;
    if !dirty.is_empty() && !args.has("allow-dirty") {
        return Err(format!("uncommitted changes in the sources the manifest pins to HEAD; commit them first (or --allow-dirty):\n{dirty}"));
    }
    if !args.has("no-build") {
        build(&root, &sources)?;
    }
    let libs = dirs.iter().map(|d| local_lib(d)).collect::<Result<Vec<_>, _>>()?;
    let publisher = Publisher::new(cli, &root).await?;
    // Each session's clientMetadata record, resolved before anything is uploaded.
    let mut with_sessions = vec![];
    for lib in &libs {
        with_sessions.push((lib, session_cid(&publisher, lib).await?));
    }
    let out = args.flag("out").map(PathBuf::from).unwrap_or_else(|| root.join("records/wasm"));
    for result in upload(&publisher, &root, &sources, &with_sessions, &out).await? {
        out!("{}", result?);
    }
    Ok(())
}

/// atlibs publish delete <at://uri | collection/rkey> ...: this account's records only.
async fn delete(cli: &Cli, args: &Args) -> Result<(), String> {
    args.only(&[], "publish delete")?;
    if args.pos.is_empty() {
        return Err(crate::usage("publish delete"));
    }
    let session = cli.require_session().await?;
    let did = session.did();
    for target in &args.pos {
        let (repo, collection, rkey) = match crate::registry::split_uri(target) {
            Some(parts) => parts,
            None => match target.split_once('/') {
                Some((c, r)) if !c.is_empty() && !r.is_empty() && !r.contains('/') => (did.as_str(), c, r),
                _ => return Err(format!("{target} is not at://did/collection/rkey or collection/rkey")),
            },
        };
        if repo != did {
            return Err(format!("{target} is not in {did}'s repo"));
        }
        session.xrpc("com.atproto.repo.deleteRecord", None, Some(&json!({ "repo": did, "collection": collection, "rkey": rkey })), vec![]).await?;
        out!("{}", json!({ "deleted": format!("at://{did}/{collection}/{rkey}") }));
    }
    Ok(())
}

/// atlibs publish versions <name> ... [--dry-run]: each named module and composition record of this account,
/// copied as is to its version key when it has none (published before pins were kept): getRecord only, no build,
/// so no pin changes. A JSON line per record: exists, would-write or written; none when the name has neither.
async fn versions(cli: &Cli, args: &Args) -> Result<(), String> {
    args.only(&["dry-run"], "publish versions")?;
    if args.pos.is_empty() {
        return Err(crate::usage("publish versions"));
    }
    let publisher = Publisher::new(cli, &root_of(&std::env::current_dir().map_err(|e| e.to_string())?)).await?;
    for name in &args.pos {
        let mut found = false;
        for collection in [MODULE, COMPOSITION] {
            let Some(live) = publisher.get(collection, name).await? else { continue };
            found = true;
            let cid = live["cid"].as_str().unwrap_or_default();
            let version = match collection {
                MODULE => module_version(name, &live["value"])?,
                _ => composition_version(name, cid).ok_or_else(|| format!("{name}: no CID"))?,
            };
            let status = match publisher.get(collection, &version).await? {
                Some(_) => "exists",
                None if args.has("dry-run") => "would-write",
                None => {
                    publisher.put_version(collection, &version, &live["value"]).await?;
                    "written"
                }
            };
            out!("{}", json!({ "status": status, "uri": format!("at://{}/{collection}/{version}", publisher.did), "cid": cid }));
        }
        if !found {
            out!("{}", json!({ "status": "none", "name": name }));
        }
    }
    Ok(())
}

/// The *.json files in a directory, sorted.
fn json_files(dir: &Path) -> Result<Vec<PathBuf>, String> {
    let mut all: Vec<PathBuf> = std::fs::read_dir(dir).map_err(|e| format!("{}: {e}", dir.display()))?.filter_map(|e| e.ok().map(|e| e.path())).filter(|p| p.extension().is_some_and(|e| e == "json")).collect();
    all.sort();
    Ok(all)
}

async fn records(cli: &Cli, args: &Args) -> Result<(), String> {
    args.only(&[], "publish records")?;
    let cwd = std::env::current_dir().map_err(|e| e.to_string())?;
    let root = root_of(&cwd);
    let files: Vec<PathBuf> = if args.pos.is_empty() {
        json_files(&root.join("records"))?
    } else {
        args.pos.iter().map(PathBuf::from).collect()
    };
    let publisher = Publisher::new(cli, &root).await?;
    for file in files {
        let record: Value = serde_json::from_slice(&std::fs::read(&file).map_err(|e| format!("{}: {e}", file.display()))?).map_err(|e| format!("{}: {e}", file.display()))?;
        let collection = record["$type"].as_str().ok_or_else(|| format!("{}: no $type", file.display()))?.to_string();
        let rkey = file.file_stem().and_then(|s| s.to_str()).unwrap_or_default();
        out!("{}", publisher.write("putRecord", &collection, rkey, &record).await?);
    }
    Ok(())
}

async fn composition(cli: &Cli, args: &Args) -> Result<(), String> {
    args.only(&["rkey"], "publish composition")?;
    let [file] = &args.pos[..] else { return Err(crate::usage("publish composition")) };
    let path = Path::new(file);
    let mut manifest: Value = serde_json::from_slice(&std::fs::read(path).map_err(|e| format!("{file}: {e}"))?).map_err(|e| format!("{file}: {e}"))?;
    let stem = path.file_stem().and_then(|s| s.to_str()).unwrap_or_default();
    let rkey = args.flag("rkey").map(String::from).unwrap_or_else(|| {
        if stem == "composition" { path.canonicalize().ok().and_then(|p| p.parent()?.file_name()?.to_str().map(String::from)).unwrap_or_default() } else { stem.to_string() }
    });
    let publisher = Publisher::new(cli, &root_of(path.parent().filter(|p| !p.as_os_str().is_empty()).unwrap_or(Path::new(".")))).await?;
    for inst in manifest["instances"].as_array_mut().into_iter().flatten() {
        let Some(m) = inst["module"].as_str().filter(|m| !m.starts_with("at://")).map(String::from) else { continue };
        if !crate::registry::is_remote(&m) {
            inst["module"] = json!(format!("at://{}/{MODULE}/{m}", publisher.did));
            continue;
        }
        // <handle>/<lib>[@pin]: the record keeps the DID's at:// URI (a handle can move) and, pinned, the full pin
        let (uri, pin) = crate::registry::resolve(&cli.net, &m).await?;
        if crate::registry::split_uri(&uri).map(|(_, c, _)| c) != Some(MODULE) {
            return Err(format!("instance {}: {m} is not a {MODULE}", inst["name"]));
        }
        if pin.is_some() {
            let lib = crate::registry::fetch_lib(&cli.net, &uri, Some(&m), pin.as_deref()).await?;
            inst["pin"] = json!(lib.pin());
        }
        inst["module"] = json!(uri);
    }
    manifest["$type"] = json!(COMPOSITION);
    if manifest.get("createdAt").is_none() {
        manifest["createdAt"] = json!(iso_now());
    }
    compose::validate(&manifest)?;
    out!("{}", publisher.write("putRecord", COMPOSITION, &rkey, &manifest).await?);
    Ok(())
}

// ---- atlibs publish [<name> ...]: only what changed -----------------------------------------

/// $XDG_CACHE_HOME/atlibs, default ~/.cache/atlibs: what `atlibs publish` last published, per account DID
/// (a handle change keeps it): <did>/<lib>.json for a module, <did>/<collection>/<rkey>.json for a record.
pub fn cache_root() -> PathBuf {
    let home = || PathBuf::from(std::env::var("HOME").unwrap_or_default()).join(".cache");
    crate::env("XDG_CACHE_HOME").map(PathBuf::from).filter(|p| p.is_absolute()).unwrap_or_else(home).join("atlibs")
}

/// A module record's content besides its blob, source, session CID and dates: the module.json fields as
/// the record carries them (actions default [], the session by rkey), as a DRISL CID.
fn module_content(value: &Value) -> Result<String, String> {
    let mut spec = spec_of(value);
    spec.entry("actions").or_insert(json!([]));
    if let Some(rkey) = spec.get("session").and_then(|s| s["uri"].as_str()).and_then(|u| u.rsplit('/').next()).map(String::from) {
        spec.insert("session".into(), json!(rkey));
    }
    drisl_cid(&Value::Object(spec))
}

/// A live module record (listRecords) as a cache entry: its pin from the blob CID's sha256.
fn live_module(found: &Value) -> Option<Value> {
    let v = &found["value"];
    let sha = blob_sha256(v.pointer("/module/ref/$link").and_then(Value::as_str)).ok()?;
    Some(json!({ "pin": pin_of_sha256(&sha, v), "content": module_content(v).ok()?, "session": v.pointer("/session/cid"), "uri": found["uri"], "cid": found["cid"] }))
}

struct Row {
    status: &'static str,
    name: String,
    entry: Value,
    pin: String,
    note: String,
}

fn read_cache(path: &Path) -> Option<Value> {
    serde_json::from_slice(&std::fs::read(path).ok()?).ok()
}

/// After a live check: the cache file holds what is live (nothing when nothing is). A note when the live
/// record is not the one the cache had (published elsewhere, or deleted): drift.
fn refresh(path: &Path, cached: Option<Value>, live: Option<Value>) -> Result<(Option<Value>, String), String> {
    let note = match (&cached, &live) {
        (Some(c), l) if l.as_ref().map(|l| &l["cid"]) != Some(&c["cid"]) => format!("drift: cached {} is live {}", c["cid"].as_str().unwrap_or("-"), l.as_ref().and_then(|l| l["cid"].as_str()).unwrap_or("nothing")),
        _ => String::new(),
    };
    match &live {
        Some(l) => write_json(path, l)?,
        None => drop(std::fs::remove_file(path)),
    }
    Ok((live, note))
}

/// `atlibs publish [<name> ...]`: every module (examples/rust/* or modules/*, or --dir) and records/*.json, else the
/// named ones (and their session records); each compared with the cache (no network) or, with no cache
/// entry or --check-live, the live record (one listRecords per collection); only what changed is published.
async fn changed(cli: &Cli, args: &Args, cache_root: &Path) -> Result<Vec<Row>, String> {
    args.only(&["dir", "dry-run", "check-live", "allow-dirty", "no-build", "out"], "publish")?;
    let (dry_run, check_live) = (args.has("dry-run"), args.has("check-live"));
    let cwd = std::env::current_dir().map_err(|e| e.to_string())?;
    let base = args.flag("dir").map(|d| cwd.join(d)).unwrap_or_else(|| cwd.clone());
    let root = root_of(&base);
    let mut dirs = vec![];
    // rkey -> record, sorted
    let mut records = std::collections::BTreeMap::new();
    let mut add_record = |file: &Path| -> Result<(), String> {
        let record: Value = serde_json::from_slice(&std::fs::read(file).map_err(|e| format!("{}: {e}", file.display()))?).map_err(|e| format!("{}: {e}", file.display()))?;
        records.insert(file.file_stem().and_then(|s| s.to_str()).unwrap_or_default().to_string(), record);
        Ok(())
    };
    if args.pos.is_empty() {
        let examples = if args.has("dir") { Some(base.clone()) } else { find_up(&base, "examples/rust").or_else(|| find_up(&base, "modules")) };
        let examples = examples.ok_or_else(|| format!("no examples/rust or modules in or above {}: --dir <modules>", base.display()))?;
        let mut found: Vec<PathBuf> = std::fs::read_dir(&examples).map_err(|e| format!("{}: {e}", examples.display()))?.filter_map(|e| e.ok().map(|e| e.path())).filter(|d| d.join("module.json").is_file()).collect();
        found.sort();
        dirs = found;
        if root.join("records").is_dir() {
            for file in json_files(&root.join("records"))? {
                add_record(&file)?;
            }
        }
    } else {
        for name in &args.pos {
            match module_dir(&base, name) {
                Some(d) => dirs.push(d),
                None if root.join(format!("records/{name}.json")).is_file() => add_record(&root.join(format!("records/{name}.json")))?,
                None => return Err(format!("no module {name} (a directory with module.json, or examples/rust/{name}) and no records/{name}.json")),
            }
        }
    }
    let sources = sources(&root, &dirs);
    let dirty = dirty(&root, &sources)?;
    if !args.has("no-build") && !dirs.is_empty() {
        build(&root, &sources)?;
    }
    let libs: Vec<(String, Result<Lib, String>)> = dirs.iter().map(|d| (d.file_name().and_then(|n| n.to_str()).unwrap_or_default().to_string(), local_lib(d))).collect();
    for lib in libs.iter().filter_map(|(_, l)| l.as_ref().ok()) {
        if let Some(c) = &lib.client {
            records.entry(c.rkey.clone()).or_insert_with(|| c.record.clone());
        }
    }
    let publisher = Publisher::new(cli, &root).await?;
    let cache = cache_root.join(&publisher.did);
    let record_cache = |collection: &str, rkey: &str| cache.join(collection).join(format!("{rkey}.json"));
    let module_cache = |key: &str| cache.join(format!("{key}.json"));

    // One listRecords per collection that has an item to check live.
    let mut wanted: Vec<&str> = records.iter().filter_map(|(rkey, r)| r["$type"].as_str().filter(|c| check_live || !record_cache(c, rkey).is_file())).collect();
    if libs.iter().any(|(_, l)| l.as_ref().is_ok_and(|l| check_live || !module_cache(&l.key).is_file())) {
        wanted.push(MODULE);
    }
    wanted.sort();
    wanted.dedup();
    let mut live: HashMap<&str, HashMap<String, Value>> = HashMap::new();
    for collection in wanted {
        live.insert(collection, list(&publisher, collection).await?);
    }

    let mut rows = vec![];
    // clientMetadata rkey -> its CID now; None: about to change (dry run) or failed
    let mut sessions: HashMap<String, Option<String>> = HashMap::new();
    for (rkey, record) in &records {
        let Some(collection) = record["$type"].as_str() else {
            rows.push(Row { status: "failed", name: format!("records/{rkey}"), entry: json!({}), pin: String::new(), note: "no $type".into() });
            continue;
        };
        let content = match drisl_cid(record) {
            Ok(c) => c,
            Err(e) => {
                rows.push(Row { status: "failed", name: format!("records/{rkey}"), entry: json!({}), pin: String::new(), note: e });
                continue;
            }
        };
        let path = record_cache(collection, rkey);
        let (cached, note) = match live.get(collection) {
            Some(listed) => refresh(&path, read_cache(&path), listed.get(rkey).and_then(|f| Some(json!({ "content": drisl_cid(&f["value"]).ok()?, "uri": f["uri"], "cid": f["cid"] }))))?,
            None => (read_cache(&path), String::new()),
        };
        let current = |cid: Option<&str>| (collection == CLIENT).then(|| (rkey.clone(), cid.map(String::from)));
        match cached.filter(|c| c["content"] == content) {
            Some(c) => {
                sessions.extend(current(c["cid"].as_str()));
                rows.push(Row { status: if note.is_empty() { "unchanged" } else { "drift" }, name: format!("records/{rkey}"), entry: c, pin: String::new(), note });
            }
            None if dry_run => {
                sessions.extend(current(None));
                rows.push(Row { status: "would-publish", name: format!("records/{rkey}"), entry: json!({}), pin: String::new(), note });
            }
            None => match publisher.write("putRecord", collection, rkey, record).await {
                Ok(w) => {
                    let entry = json!({ "content": content, "uri": w["uri"], "cid": w["cid"], "publishedAt": iso_now() });
                    write_json(&path, &entry)?;
                    sessions.extend(current(w["cid"].as_str()));
                    rows.push(Row { status: "published", name: format!("records/{rkey}"), entry, pin: String::new(), note });
                }
                Err(e) => {
                    sessions.extend(current(None));
                    rows.push(Row { status: "failed", name: format!("records/{rkey}"), entry: json!({}), pin: String::new(), note: e });
                }
            },
        }
    }

    let mut pending = vec![];
    for (name, lib) in &libs {
        let lib = match lib {
            Ok(l) => l,
            Err(e) => {
                rows.push(Row { status: "failed", name: name.clone(), entry: json!({}), pin: String::new(), note: e.clone() });
                continue;
            }
        };
        let (pin, path) = (lib.pin(), module_cache(&lib.key));
        let content = match module_content(&lib.spec) {
            Ok(c) => c,
            Err(e) => {
                rows.push(Row { status: "failed", name: lib.key.clone(), entry: json!({}), pin, note: e });
                continue;
            }
        };
        let (cached, mut note) = match live.get(MODULE) {
            Some(listed) => refresh(&path, read_cache(&path), listed.get(&lib.key).and_then(live_module))?,
            None => (read_cache(&path), String::new()),
        };
        // its session record changed (or is about to): the module record names that record's CID
        let session_moved = |c: &Value| match lib.spec["session"].as_str().and_then(|s| sessions.get(s)) {
            Some(Some(cid)) => c["session"].as_str() != Some(cid.as_str()),
            Some(None) => true,
            None => false,
        };
        if let Some(c) = cached.as_ref().filter(|c| c["pin"] == pin.as_str() && c["content"] == content.as_str() && !session_moved(c)) {
            rows.push(Row { status: if note.is_empty() { "unchanged" } else { "drift" }, name: lib.key.clone(), entry: c.clone(), pin, note });
            continue;
        }
        if cached.as_ref().is_some_and(|c| c["pin"] == pin.as_str() && c["content"] == content.as_str()) {
            note = [note, "session record changed".into()].join(" ").trim().to_string();
        }
        if !dirty.is_empty() && !args.has("allow-dirty") {
            rows.push(Row { status: "skipped:dirty", name: lib.key.clone(), entry: json!({}), pin, note });
        } else if dry_run {
            rows.push(Row { status: "would-publish", name: lib.key.clone(), entry: json!({}), pin, note });
        } else {
            match session_cid(&publisher, lib).await {
                Ok(s) => {
                    // filled in once uploaded
                    rows.push(Row { status: "failed", name: lib.key.clone(), entry: json!({}), pin: pin.clone(), note: String::new() });
                    pending.push((rows.len() - 1, lib, s, content, pin, path, note));
                }
                Err(e) => rows.push(Row { status: "failed", name: lib.key.clone(), entry: json!({}), pin, note: e }),
            }
        }
    }
    if !pending.is_empty() {
        let commit = git(&root, &["rev-parse", "HEAD"])?;
        let out = args.flag("out").map(PathBuf::from).unwrap_or_else(|| root.join("records/wasm"));
        let pairs: Vec<(&Lib, String)> = pending.iter().map(|(_, lib, s, ..)| (*lib, s.clone())).collect();
        let results = match upload(&publisher, &root, &sources, &pairs, &out).await {
            Ok(results) => results,
            Err(e) => pending.iter().map(|_| Err(e.clone())).collect(),
        };
        for ((i, lib, session, content, pin, path, note), result) in pending.into_iter().zip(results) {
            rows[i] = match result {
                Ok(w) => {
                    let entry = json!({ "pin": pin, "content": content, "session": (!session.is_empty()).then_some(session), "uri": w["uri"], "cid": w["cid"], "commit": commit, "publishedAt": iso_now() });
                    write_json(&path, &entry)?;
                    Row { status: "published", name: lib.key.clone(), entry, pin, note }
                }
                Err(e) => Row { status: "failed", name: lib.key.clone(), entry: json!({}), pin, note: e },
            };
        }
    }

    let width = rows.iter().map(|r| r.name.len()).max().unwrap_or(0).max(4);
    // the ref as `atlibs add` takes it, by the handle --as named; records/wasm/<name>.json keeps the uri
    let who = crate::login::handle().unwrap_or_else(|| publisher.did.clone());
    out!("{:<13}  {:<width$}  {:<64}  {:<64}  ref / note", "status", "name", "pin", "cid");
    for r in &rows {
        let pin = if r.pin.is_empty() { "-" } else { &r.pin };
        let shown = r.entry["uri"].as_str().and_then(crate::registry::split_uri).map(|(_, _, rkey)| format!("{who}/{rkey}")).unwrap_or_default();
        let what = [shown.as_str(), &r.note].iter().filter(|s| !s.is_empty()).copied().collect::<Vec<_>>().join("  ");
        out!("{:<13}  {:<width$}  {pin:<64}  {:<64}  {what}", r.status, r.name, r.entry["cid"].as_str().unwrap_or("-"));
    }
    let count = |status: &str| rows.iter().filter(|r| r.status == status).count();
    eprintln!("{} published, {} unchanged, {} would publish, {} skipped:dirty, {} drift, {} failed (cache {})", count("published"), count("unchanged"), count("would-publish"), count("skipped:dirty"), count("drift"), count("failed"), cache.display());
    if count("skipped:dirty") > 0 {
        eprintln!("skipped:dirty: commit the sources first (the manifest pins HEAD), or --allow-dirty:\n{dirty}");
    }
    Ok(rows)
}

// ---- web tiles -----------------------------------------------------------------------------

fn head(out: &mut Vec<u8>, major: u8, n: u64) {
    let m = major << 5;
    match n {
        0..24 => out.push(m | n as u8),
        24..0x100 => out.extend([m | 24, n as u8]),
        0x100..0x10000 => out.extend([m | 25].into_iter().chain((n as u16).to_be_bytes())),
        0x10000..0x1_0000_0000 => out.extend([m | 26].into_iter().chain((n as u32).to_be_bytes())),
        _ => out.extend([m | 27].into_iter().chain(n.to_be_bytes())),
    }
}

/// DRISL (deterministic dag-cbor): map keys by length then bytes, { $link } a CID (tag 42).
// ponytail: integers only, no floats, which a tile manifest does not have.
pub fn drisl(value: &Value, out: &mut Vec<u8>) -> Result<(), String> {
    match value {
        Value::Null => out.push(0xf6),
        Value::Bool(b) => out.push(if *b { 0xf5 } else { 0xf4 }),
        Value::Number(n) => match n.as_i64() {
            Some(i) if i >= 0 => head(out, 0, i as u64),
            Some(i) => head(out, 1, (-1 - i) as u64),
            None => return Err(format!("{n} is not an integer")),
        },
        Value::String(s) => {
            head(out, 3, s.len() as u64);
            out.extend(s.as_bytes());
        }
        Value::Array(items) => {
            head(out, 4, items.len() as u64);
            for item in items {
                drisl(item, out)?;
            }
        }
        Value::Object(map) if map.len() == 1 && map.contains_key("$link") => {
            let cid = [&[0u8][..], &unbase32(map["$link"].as_str().unwrap_or_default().get(1..).unwrap_or_default())].concat();
            out.extend([0xd8, 42]);
            head(out, 2, cid.len() as u64);
            out.extend(cid);
        }
        Value::Object(map) => {
            let mut keys: Vec<&String> = map.keys().collect();
            keys.sort_by(|a, b| a.len().cmp(&b.len()).then(a.as_bytes().cmp(b.as_bytes())));
            head(out, 5, keys.len() as u64);
            for key in keys {
                head(out, 3, key.len() as u64);
                out.extend(key.as_bytes());
                drisl(&map[key], out)?;
            }
        }
    }
    Ok(())
}

pub fn drisl_cid(value: &Value) -> Result<String, String> {
    let mut bytes = vec![];
    drisl(value, &mut bytes)?;
    Ok(format!("b{}", base32(&[&[0x01, 0x71, 0x12, 0x20][..], &Sha256::digest(&bytes)].concat())))
}

fn tid() -> String {
    const S32: &[u8] = b"234567abcdefghijklmnopqrstuvwxyz";
    let rand = u64::from(atlibs_host::store::random_bytes::<2>()[0]) | (u64::from(atlibs_host::store::random_bytes::<1>()[0] & 3) << 8);
    let mut n = (now_ms() * 1000) << 10 | rand;
    let mut out = [0u8; 13];
    for slot in out.iter_mut().rev() {
        *slot = S32[(n & 31) as usize];
        n >>= 5;
    }
    String::from_utf8(out.to_vec()).expect("ascii")
}

fn tile_files(dir: &Path, prefix: &str, out: &mut Vec<String>) -> Result<(), String> {
    let mut names: Vec<String> = std::fs::read_dir(dir.join(prefix.trim_start_matches('/'))).map_err(|e| e.to_string())?.filter_map(|e| e.ok()?.file_name().into_string().ok()).filter(|n| !n.starts_with('.')).collect();
    names.sort();
    for name in names {
        let path = format!("{prefix}/{name}");
        // metadata() follows symlinks, as the examples' atlibs -> javascript/src links need
        if std::fs::metadata(dir.join(path.trim_start_matches('/'))).map_err(|e| format!("{path}: {e}"))?.is_dir() {
            tile_files(dir, &path, out)?;
        } else {
            out.push(path);
        }
    }
    Ok(())
}

fn content_type(path: &str) -> &'static str {
    match path.rsplit('.').next().unwrap_or_default() {
        "html" => "text/html",
        "js" | "mjs" => "text/javascript",
        "css" => "text/css",
        "json" => "application/json",
        "svg" => "image/svg+xml",
        "png" => "image/png",
        "wasm" => "application/wasm",
        _ => "application/octet-stream",
    }
}

async fn tile(cli: &Cli, args: &Args) -> Result<(), String> {
    args.only(&["out"], "publish tile")?;
    let [dir, name] = &args.pos[..] else { return Err(crate::usage("publish tile")) };
    let dir = Path::new(dir);
    let root = root_of(&std::env::current_dir().map_err(|e| e.to_string())?);
    let publisher = Publisher::new(cli, &root).await?;
    let mut masl: Value = serde_json::from_slice(&std::fs::read(dir.join("manifest.json")).map_err(|e| format!("{}/manifest.json: {e}", dir.display()))?).map_err(|e| e.to_string())?;
    let mut paths = vec![];
    tile_files(dir, "", &mut paths)?;
    let mut resources = Map::new();
    for path in paths.iter().filter(|p| *p != "/manifest.json") {
        // the spec: upload as octet-stream and keep the PDS's blob ref; content-type is what the loader serves
        let bytes = std::fs::read(dir.join(path.trim_start_matches('/'))).map_err(|e| format!("{path}: {e}"))?;
        let blob = publisher.session.upload_blob(bytes, "application/octet-stream").await?["blob"].clone();
        resources.insert(if path == "/index.html" { "/".into() } else { path.clone() }, json!({ "src": blob, "content-type": content_type(path) }));
    }
    if !resources.contains_key("/") {
        return Err(format!("{} has no index.html", dir.display()));
    }
    masl["resources"] = Value::Object(resources);
    let params: Map<String, Value> = [("repo", json!(publisher.did)), ("collection", json!(MASL)), ("limit", json!("100"))].into_iter().map(|(k, v)| (k.to_string(), v)).collect();
    let listed = publisher.session.xrpc("com.atproto.repo.listRecords", Some(&params), None, vec![]).await?;
    let rkey = listed["records"].as_array().into_iter().flatten().find(|r| r["value"]["tile"]["name"] == masl["name"]).and_then(|r| r["uri"].as_str()?.rsplit('/').next().map(String::from)).unwrap_or_else(tid);
    let record = json!({ "$type": MASL, "cid": drisl_cid(&masl)?, "tile": masl, "createdAt": iso_now() });
    let file = args.flag("out").map(PathBuf::from).unwrap_or_else(|| root.join("records/tiles")).join(format!("{name}.json"));
    write_json(&file, &record)?;
    let written = publisher.write("putRecord", MASL, &rkey, &record).await?;
    out!("{}", json!({ "uri": written["uri"], "cid": written["cid"], "file": file }));
    Ok(())
}

pub async fn main(cli: &Cli, argv: &[String]) -> Result<(), String> {
    let usage = || crate::usage("publish");
    let what = argv.first().map(String::as_str).unwrap_or_default();
    if !["module", "records", "tile", "composition", "delete", "versions"].contains(&what) {
        let rows = changed(cli, &Args::parse(argv, &["dry-run", "check-live", "no-build", "allow-dirty"])?, &cache_root()).await?;
        return match rows.iter().filter(|r| r.status == "failed").count() {
            0 => Ok(()),
            n => Err(format!("{n} failed")),
        };
    }
    let rest = &argv[1..];
    let args = Args::parse(rest, &["no-build", "allow-dirty", "dry-run"])?;
    match what {
        "module" => module(cli, &args).await,
        "records" => records(cli, &args).await,
        "tile" => tile(cli, &args).await,
        "composition" => composition(cli, &args).await,
        "delete" => delete(cli, &args).await,
        "versions" => versions(cli, &args).await,
        _ => Err(usage()),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::fake::Pds;
    use crate::registry::tests::tempdir;

    const REPO: &str = concat!(env!("CARGO_MANIFEST_DIR"), "/../..");

    #[test]
    fn module_records_for_every_example_are_valid() {
        let lex = Lexicons::load(None).unwrap();
        let blob = json!({ "$type": "blob", "ref": { "$link": "bafkreie5737gdxlw5i64vzichcalba3z2v5n6icifvx5xytvske7mr3hpm" }, "mimeType": "application/wasm", "size": 4 });
        let source = json!({ "uri": "at://did:plc:pub/com.fedfork.exec.source.manifest/m", "cid": "bafyreie5737gdxlw5i64vzichcalba3z2v5n6icifvx5xytvske7mr3hpm" });
        let spec = |name: &str| serde_json::from_slice::<Value>(&std::fs::read(Path::new(REPO).join(format!("examples/rust/{name}/module.json"))).unwrap()).unwrap();
        let record = |name: &str| module_record(&spec(name), &blob, "did:plc:pub", "bafysession", &source, "2026-11-08T00:00:00.000Z");
        let compose = record("compose");
        assert!(compose.get("session").is_none());
        assert_eq!(compose["calls"], json!([]));
        let didkey = module_record(&json!({ "name": "relay", "didkey": { "services": ["https://xrpc.fedproxy.com"] } }), &blob, "did:plc:pub", "", &source, "2026-11-08T00:00:00.000Z");
        assert_eq!(didkey["didkey"], json!({ "services": ["https://xrpc.fedproxy.com"] }));
        lex.validate(&didkey).unwrap();
        assert_eq!(record("statusphere")["session"], json!({ "uri": "at://did:plc:pub/com.fedfork.exec.oauth.clientMetadata/statusphere", "cid": "bafysession" }));
        for entry in std::fs::read_dir(Path::new(REPO).join("examples/rust")).unwrap() {
            let name = entry.unwrap().file_name().into_string().unwrap();
            if Path::new(REPO).join(format!("examples/rust/{name}/module.json")).is_file() {
                lex.validate(&record(&name)).unwrap_or_else(|e| panic!("{name}: {e}"));
            }
        }
    }

    #[test]
    fn drisl_cids_match_other_implementations() {
        assert_eq!(drisl_cid(&json!({})).unwrap(), "bafyreigbtj4x7ip5legnfznufuopl4sg4knzc2cof6duas4b3q2fy6swua");
        // at://did:plc:izttpdp3l6vss5crelt5kcux/ing.dasl.masl/3meggnybzqs2n, published by atile (@atcute/cbor)
        let hello = json!({
            "name": "The Hello Tile",
            "icons": [{ "src": "/icon.png" }],
            "resources": {
                "/": { "src": { "ref": { "$link": "bafkreicknles2uzv7sruakcbjtn5pnoc4rqmgkvnavuq4raxexjvcn7osq" }, "size": 20, "$type": "blob", "mimeType": "application/octet-stream" }, "content-type": "text/html" },
                "/icon.png": { "src": { "ref": { "$link": "bafkreif5dsnqtcqbxmycphlfbqoebbexcijhpj3faqtlkcsrfeap7g5qcm" }, "size": 2136, "$type": "blob", "mimeType": "image/png" }, "content-type": "image/png" },
                "/banner.jpg": { "src": { "ref": { "$link": "bafkreifhqqoqp7bch26ogcsnkgfui24pva23iw5hapfcfeg2t3f7rsxsyi" }, "size": 187308, "$type": "blob", "mimeType": "image/jpeg" }, "content-type": "image/jpeg" }
            },
            "description": "An extremely simple tile that just demonstrates how easy it is to get something published and going.",
            "screenshots": [{ "src": "/banner.jpg" }]
        });
        assert_eq!(drisl_cid(&hello).unwrap(), "bafyreibbbvc5iydonff5p7px7qyhlrbs7uuqjvlqav4pj53qdsdnc7e7aa");
        assert_eq!(tid().len(), 13);
    }

    #[tokio::test]
    async fn publish_composition_and_a_tile_to_a_fake_pds_with_one_session() {
        let pds = Pds::new();
        let dir = tempdir("publish");
        let cli = pds.cli(&dir);
        sign_in_as_bob(&dir);
        let comp = Path::new(REPO).join("examples/rust/compose/composition.json");
        composition(&cli, &Args::parse(&[comp.display().to_string()], &[]).unwrap()).await.unwrap();
        let published = pds.get("at://did:plc:bobtest/com.fedfork.exec.composition/compose").unwrap();
        assert_eq!(published["instances"][0]["module"], "at://did:plc:bobtest/com.fedfork.exec.wasm.module/compose");
        // and at its version, what a binding's CID loads
        let uri = "at://did:plc:bobtest/com.fedfork.exec.composition/compose";
        let (cid, _) = cli.net.record(uri).await.unwrap().unwrap();
        assert_eq!(pds.get(&format!("{uri}.{cid}")), Some(published.clone()));
        // published modules by <handle>/<lib> or <did>/<lib>: stored by at:// URI, a pinned one with its full pin
        let echo = pds.publish_module("alice.test", "echo", b"\0asm-echo", None);
        let pin = crate::registry::fetch_lib(&cli.net, &echo, None, None).await.unwrap().pin();
        let refs = dir.join("refs.composition.json");
        write_json(&refs, &json!({ "name": "refs", "instances": [{ "name": "a", "module": "alice.test/echo" }, { "name": "b", "module": format!("did:plc:alicetest/echo@{}", &pin[..32]) }] })).unwrap();
        composition(&cli, &Args::parse(&[refs.display().to_string()], &[]).unwrap()).await.unwrap();
        let published = pds.get("at://did:plc:bobtest/com.fedfork.exec.composition/refs.composition").unwrap();
        assert_eq!(published["instances"], json!([{ "name": "a", "module": echo }, { "name": "b", "module": echo, "pin": pin }]));

        let tile_dir = dir.join("tile");
        std::fs::create_dir_all(tile_dir.join("js")).unwrap();
        std::fs::write(tile_dir.join("manifest.json"), r#"{"name":"t"}"#).unwrap();
        std::fs::write(tile_dir.join("index.html"), "<p>hi</p>").unwrap();
        std::fs::write(tile_dir.join("js/a.js"), "1").unwrap();
        let p = Publisher::new(&cli, Path::new(REPO)).await.unwrap();
        assert_eq!(p.did, "did:plc:bobtest");
        let out = p.write("putRecord", CLIENT, "x", &json!({ "$type": CLIENT, "name": "x" })).await;
        assert!(out.unwrap_err().contains("is not valid"), "lexicon check before writing");
        let bare = Publisher::new(&cli, &dir).await.unwrap();
        let out = bare.write("putRecord", CLIENT, "x", &json!({ "$type": CLIENT, "name": "x" })).await;
        assert!(out.unwrap_err().contains("is not valid"), "the built-in lexicons, in a repo with no lexicons/");
        let sessions = pds.log().iter().filter(|l| l.ends_with("createSession")).count();
        assert_eq!(sessions, 1, "the saved session is reused");
        tile(&cli, &Args::parse(&[tile_dir.display().to_string(), "t".into(), "--out".into(), dir.display().to_string()], &[]).unwrap()).await.unwrap();
        let written: Value = serde_json::from_slice(&std::fs::read(dir.join("t.json")).unwrap()).unwrap();
        assert_eq!(written["cid"].as_str().unwrap(), drisl_cid(&written["tile"]).unwrap());
        assert!(written["tile"]["resources"]["/"]["src"]["ref"]["$link"].is_string());
        assert_eq!(written["tile"]["resources"]["/js/a.js"]["content-type"], "text/javascript");

        // a module, unbuilt (`cargo wasm` output), its records written under --out instead of records/wasm
        if !Path::new(REPO).join("target/wasm32-unknown-unknown/release/statusphere.wasm").exists() {
            return eprintln!("skipping publish module: run cargo wasm");
        }
        let out = dir.join("out");
        let argv = |more: &[&str]| Args::parse(&[&[Path::new(REPO).join("examples/rust/statusphere").to_str().unwrap(), "--no-build", "--allow-dirty", "--out", out.to_str().unwrap()][..], more].concat().iter().map(|s| s.to_string()).collect::<Vec<_>>(), &["no-build", "allow-dirty"]).unwrap();
        assert!(module(&cli, &argv(&[])).await.unwrap_err().contains("atlibs publish records"), "the session record comes first");
        pds.put("bob.test", CLIENT, "statusphere", json!({ "$type": CLIENT }));
        module(&cli, &argv(&[])).await.unwrap();
        let record = pds.get("at://did:plc:bobtest/com.fedfork.exec.wasm.module/statusphere").unwrap();
        assert_eq!(record["abi"], ABI);
        assert!(record["source"]["uri"].as_str().unwrap().contains("com.fedfork.exec.source.manifest/statusphere."));
        let manifest: Value = serde_json::from_slice(&std::fs::read(out.join("statusphere.manifest.json")).unwrap()).unwrap();
        assert!(manifest["files"].as_array().unwrap().iter().any(|f| f["path"] == "examples/rust/statusphere/src/lib.rs"));
        assert!(!manifest["files"].as_array().unwrap().iter().any(|f| f["path"] == "examples/rust/statusphere/module.json"));
        let paths: Vec<&str> = manifest["files"].as_array().unwrap().iter().filter_map(|f| f["path"].as_str()).collect();
        assert!(["Cargo.lock", ".cargo/config.toml", "rust/src/abi/src/lib.rs", "examples/rust/compose/src/lib.rs"].iter().all(|p| paths.contains(p)), "the workspace: {paths:?}");
        assert!(!paths.iter().any(|p| p.starts_with("docs/") || p.starts_with("rust/host/")), "nothing the build does not read: {paths:?}");
        assert_eq!(manifest["build"]["command"], BUILD);
        let lib = crate::registry::fetch_lib(&cli.net, record_uri(&record, "did:plc:bobtest").as_str(), None, None).await.unwrap();
        assert_eq!(lib.sha256, manifest["output"]["sha256"].as_str().unwrap(), "what add fetches is what was built");
        let version = format!("at://did:plc:bobtest/com.fedfork.exec.wasm.module/statusphere.{}", &lib.pin()[..32]);
        assert_eq!(pds.get(&version), Some(record.clone()), "and at its version, what pins load");
        let created = pds.log().iter().filter(|l| l.ends_with("createRecord")).count();
        module(&cli, &argv(&[])).await.unwrap();
        assert_eq!(pds.log().iter().filter(|l| l.ends_with("createRecord")).count(), created + 1, "republishing reuses every source file record: only a new manifest");
        delete(&cli, &Args::parse(&["com.fedfork.exec.wasm.module/statusphere".into()], &[]).unwrap()).await.unwrap();
        assert!(pds.get("at://did:plc:bobtest/com.fedfork.exec.wasm.module/statusphere").is_none());
        assert!(delete(&cli, &Args::parse(&["at://did:plc:other/c/r".into()], &[]).unwrap()).await.unwrap_err().contains("not in did:plc:bobtest"));
    }

    /// `atlibs publish` on a two-module workspace (a git repo): first everything, then nothing (the
    /// cache, no network), then only what changed; dry runs, drift and dirty sources.
    #[tokio::test]
    async fn publish_only_what_changed() {
        const MODULE_URI: &str = "at://did:plc:bobtest/com.fedfork.exec.wasm.module/";
        let pds = Pds::new();
        let dir = tempdir("changed");
        let cli = pds.cli(&dir);
        sign_in_as_bob(&dir);
        let ws = dir.join("ws");
        let git = |args: &[&str]| assert!(Command::new("git").args(["-c", "user.name=t", "-c", "user.email=t@t", "-C"]).arg(&ws).args(args).output().unwrap().status.success(), "git {args:?}");
        let file = |rel: &str, text: &str| {
            std::fs::create_dir_all(ws.join(rel).parent().unwrap()).unwrap();
            std::fs::write(ws.join(rel), text).unwrap();
        };
        file("examples/rust/a/module.json", r#"{"name":"a","calls":[]}"#);
        file("examples/rust/a/src/lib.rs", "// a");
        file("examples/rust/b/module.json", r#"{"name":"b","session":"b","calls":["com.atproto.repo.getRecord"]}"#);
        file("examples/rust/b/src/lib.rs", "// b");
        file("records/b.json", r#"{"$type":"com.fedfork.exec.oauth.clientMetadata","name":"b","clientMetadata":{"client_id":"https://b.test/c.json","redirect_uris":["https://b.test/"],"scope":"atproto"},"createdAt":"2026-11-10T00:00:00.000Z"}"#);
        file("target/wasm32-unknown-unknown/release/a.wasm", "wasm a");
        file("target/wasm32-unknown-unknown/release/b.wasm", "wasm b");
        file(".gitignore", "target\n");
        std::fs::create_dir_all(&ws).unwrap();
        git(&["init", "-q"]);
        git(&["remote", "add", "origin", "https://example.test/ws"]);
        git(&["add", "."]);
        git(&["commit", "-qm", "one"]);
        let cache = dir.join("cache");
        let examples = ws.join("examples/rust").display().to_string();
        let run = |more: &[&str]| {
            let argv: Vec<String> = [&["--dir", examples.as_str(), "--no-build"][..], more].concat().iter().map(|s| s.to_string()).collect();
            let (cli, cache) = (&cli, &cache);
            async move { changed(cli, &Args::parse(&argv, &["dry-run", "check-live", "no-build", "allow-dirty"]).unwrap(), cache).await.unwrap() }
        };
        let statuses = |rows: &[Row]| rows.iter().map(|r| format!("{} {}", r.name, r.status)).collect::<Vec<_>>();
        let writes = || pds.log().iter().filter(|l| l.contains("putRecord") || l.contains("uploadBlob")).count();

        assert_eq!(statuses(&run(&["--dry-run"]).await), ["records/b would-publish", "a would-publish", "b would-publish"]);
        assert_eq!(writes(), 0, "a dry run publishes nothing");
        assert_eq!(statuses(&run(&[]).await), ["records/b published", "a published", "b published"]);
        let did_cache = cache.join("did:plc:bobtest");
        let a: Value = serde_json::from_slice(&std::fs::read(did_cache.join("a.json")).unwrap()).unwrap();
        assert_eq!(a["uri"], "at://did:plc:bobtest/com.fedfork.exec.wasm.module/a");
        assert_eq!(a["commit"].as_str().unwrap().len(), 40);
        assert!(did_cache.join("com.fedfork.exec.oauth.clientMetadata/b.json").is_file(), "the cache is keyed by DID");
        let b = pds.get("at://did:plc:bobtest/com.fedfork.exec.wasm.module/b").unwrap();
        assert_eq!(b["session"]["cid"], serde_json::from_slice::<Value>(&std::fs::read(did_cache.join("com.fedfork.exec.oauth.clientMetadata/b.json")).unwrap()).unwrap()["cid"]);

        let seen = pds.log().len();
        assert_eq!(statuses(&run(&[]).await), ["records/b unchanged", "a unchanged", "b unchanged"]);
        assert_eq!(pds.log().len(), seen, "a cache hit needs no network");

        // a changed grant: only a; its pin is the one `atlibs add` checks
        file("examples/rust/a/module.json", r#"{"name":"a","calls":["com.atproto.repo.getRecord"]}"#);
        git(&["commit", "-qam", "two"]);
        let rows = run(&[]).await;
        assert_eq!(statuses(&rows), ["records/b unchanged", "a published", "b unchanged"]);
        let lib = crate::registry::fetch_lib(&cli.net, "at://did:plc:bobtest/com.fedfork.exec.wasm.module/a", None, None).await.unwrap();
        assert_eq!(lib.pin(), rows[1].pin);

        // drift: the live record is not the one this cache published (same content: reported, not republished)
        let mut live = pds.get("at://did:plc:bobtest/com.fedfork.exec.wasm.module/a").unwrap();
        live["createdAt"] = json!("2020-01-01T00:00:00.000Z");
        let (_, cid) = pds.put("bob.test", MODULE, "a", live);
        let before = writes();
        let rows = run(&["--check-live"]).await;
        assert_eq!(statuses(&rows), ["records/b unchanged", "a drift", "b unchanged"]);
        assert!(rows[1].note.contains(&cid));
        assert_eq!(writes(), before);
        assert_eq!(statuses(&run(&[]).await)[1], "a unchanged", "the cache now holds the live record");
        // a missing cache entry: one listRecords for modules, then unchanged
        std::fs::remove_file(did_cache.join("b.json")).unwrap();
        assert_eq!(statuses(&run(&[]).await)[2], "b unchanged");

        // b and a composition published before versions were kept: `publish versions` copies them, getRecord only
        let b_uri = "at://did:plc:bobtest/com.fedfork.exec.wasm.module/b";
        let b_version = format!("{MODULE_URI}b.{}", &crate::registry::fetch_lib(&cli.net, b_uri, None, None).await.unwrap().pin()[..32]);
        pds.delete(&b_version);
        let (c_uri, c_cid) = pds.put("bob.test", COMPOSITION, "c", json!({ "$type": COMPOSITION, "instances": [{ "name": "x", "module": "b" }], "createdAt": "x" }));
        let copy = |more: &[&str]| {
            let argv: Vec<String> = [&["b", "c", "nope"][..], more].concat().iter().map(|s| s.to_string()).collect();
            let cli = &cli;
            async move { versions(cli, &Args::parse(&argv, &["dry-run"]).unwrap()).await.unwrap() }
        };
        let (before, lists) = (writes(), pds.log().iter().filter(|l| l.contains("listRecords")).count());
        copy(&["--dry-run"]).await;
        assert_eq!(writes(), before, "a dry run writes nothing");
        copy(&[]).await;
        assert_eq!((pds.get(&b_version), pds.get(&format!("{c_uri}.{c_cid}"))), (pds.get(b_uri), pds.get(&c_uri)));
        copy(&[]).await;
        assert_eq!(writes(), before + 2, "once");
        assert_eq!(pds.log().iter().filter(|l| l.contains("listRecords")).count(), lists, "getRecord only");

        // a session record change republishes its module (the record names the session's CID)
        file("records/b.json", r#"{"$type":"com.fedfork.exec.oauth.clientMetadata","name":"b2","clientMetadata":{"client_id":"https://b.test/c.json","redirect_uris":["https://b.test/"],"scope":"atproto"},"createdAt":"2026-11-10T00:00:00.000Z"}"#);
        assert_eq!(statuses(&run(&["b"]).await), ["records/b published", "b published"]);

        // dirty sources: a changed module is skipped, not published from edits HEAD does not have
        file("examples/rust/a/module.json", r#"{"name":"a","calls":[]}"#);
        let before = writes();
        assert_eq!(statuses(&run(&["a"]).await), ["a skipped:dirty"]);
        assert_eq!(writes(), before);
        assert_eq!(statuses(&run(&["a", "--allow-dirty", "--dry-run"]).await), ["a would-publish"]);
    }

    #[test]
    fn the_cache_is_under_xdg_cache_home_when_it_is_absolute() {
        // SAFETY: no other test reads XDG_CACHE_HOME
        unsafe { std::env::set_var("XDG_CACHE_HOME", "/x/cache") }
        assert_eq!(cache_root(), Path::new("/x/cache/atlibs"));
        unsafe { std::env::set_var("XDG_CACHE_HOME", "relative") }
        assert!(cache_root().ends_with(".cache/atlibs"));
    }

    /// The session flags (login::take_flags, once per process: every test signs in as bob.test).
    fn sign_in_as_bob(dir: &Path) {
        let password = dir.join("password");
        std::fs::write(&password, "app-password\n").unwrap();
        crate::login::take_flags(&["--app-password".into(), "bob.test".into(), "--password-file".into(), password.display().to_string(), "--pds".into(), crate::fake::PDS.into()]).unwrap();
    }

    fn record_uri(_record: &Value, did: &str) -> String {
        format!("at://{did}/com.fedfork.exec.wasm.module/statusphere")
    }
}