// atlibs run: one run of a module entry (or a composition instance's), as javascript/src/server/cli.mjs.
//   atlibs run <handle>/<lib> | <at://did/com.fedfork.exec.wasm.module/rkey> <entry> ['<json>' | - | @file]
//   atlibs run <name in atlibs.lock.json> <entry> ['<json>']
//   atlibs run <module dir | name under examples/rust> <entry> ['<json>']     (module.json + the cargo wasm build)
//   atlibs run <module.wasm> <module.json> <entry> ['<json>']
//   atlibs run <stage.json | atlibs.lock.json> <module> <entry> ['<json>']
//   atlibs run <composition .json | composition in the lockfile | <handle>/<composition>> <instance> <entry> ['<json>']
//     [--handler <name>=<spec> ...] [--allow-exec]   the composition's declared handlers (handlers.rs)
//     [--params '<json>' | @file]   {inputs, secrets, host}: its parameters (atlibs_host::params::bind)
// '<json>' may be `-` (read stdin) or `@file`: a `stdin` secret stays out of argv. Prints runModule's
// { entry, result, error?, events, calls } as JSON on stdout; a run error is also one line on stderr
// and exit status 1. Ctrl-C ends a run and still prints it.

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

use atlibs_host::compose::{self, Composition};
use atlibs_host::host::{Fetch, Session};
use atlibs_host::run::{self, Call, Ctx, Loader, Module, Options, Runner, run_module};
use atlibs_host::store::Store;
use atlibs_host::transport::Transports;
use atlibs_host::wasm::{Bound, WALL_TIME_MS, Wasm};
use atlibs_host::{Signal, pretty};
use serde_json::{Map, Value, json};

use crate::registry::{self, Lib, Net, Ref};
use crate::{Cli, env};

/// The NSIDs this build's transports serve (cargo features), for loading before a Host exists (refs,
/// compositions): the transports keyed under the state dir this process would use (Host::provides is the
/// same set, from the Host's own transports).
pub fn provides() -> HashSet<String> {
    atlibs_host::transport::provides(&crate::transports(&crate::state_dir()))
}

/// What a ref runs: one module (pinned: the dir whose lockfile alone supplies its hook defaults, Lib::pinned),
/// or a composition (its identity is its file, or its at:// URI).
pub enum Target {
    Module(Module, Option<PathBuf>),
    Composition { id: String, comp: Composition },
}

impl Target {
    pub fn is_composition(&self) -> bool {
        matches!(self, Target::Composition { .. })
    }
}

pub use atlibs_host::registry::module_of;

/// The ref at the front of `words` (registry::resolve_ref; lockfile names in ./atlibs.lock.json) as what
/// runs, and the words after it.
pub async fn target<'a>(net: &Net, words: &'a [String]) -> Result<(Target, &'a [String]), String> {
    let (found, rest) = registry::resolve_ref(net, Path::new("."), words).await?;
    Ok((target_of(found)?, rest))
}

pub fn target_of(found: Ref) -> Result<Target, String> {
    match found {
        Ref::Module(lib) => Ok(Target::Module(module_of(&lib)?, lib.pinned)),
        // composition_libs loaded it (hook defaults added); what is left is the transports this build serves
        Ref::Composition { id, comp, .. } => compose::check(&comp, &provides()).map(|_| Target::Composition { id, comp }),
        Ref::Client(c) => Err(format!("{} is a clientMetadata record: no module to run", c.uri.unwrap_or(c.rkey))),
    }
}

/// The entries a module.json's actions name (`atlibs run <ref>` with no entry, serve's index).
pub fn actions(lib: &Lib) -> Vec<&str> {
    lib.spec["actions"].as_array().into_iter().flatten().filter_map(|a| a["entry"].as_str()).collect()
}

/// What `atlibs run <ref>` with no entry prints: the entries its module.json actions name.
fn entries(found: &Ref) -> String {
    let actions = |lib: &Lib| actions(lib).join(", ");
    match found {
        Ref::Module(lib) => format!("{} entries: {}", lib.key, actions(lib)),
        Ref::Composition { manifest, libs, .. } => manifest["instances"]
            .as_array()
            .into_iter()
            .flatten()
            .map(|inst| format!("{} ({}): {}", inst["name"].as_str().unwrap_or_default(), inst["module"].as_str().unwrap_or_default(), libs.iter().find(|l| inst["module"] == l.key.as_str()).map(actions).unwrap_or_default()))
            .collect::<Vec<_>>()
            .join("\n"),
        Ref::Client(_) => String::new(),
    }
}

pub fn input_of(json: Option<&String>) -> Result<Map<String, Value>, String> {
    let json = json.map_or("{}", String::as_str);
    match serde_json::from_str::<Value>(json) {
        Ok(Value::Object(input)) => Ok(input),
        Ok(_) => Err("input must be a JSON object".into()),
        // never the text itself: it may carry a secret (stdin)
        Err(e) => Err(format!("input is not JSON (line {}, column {})", e.line(), e.column())),
    }
}

/// The CLI input word: '<json>', `-` (stdin) or `@file`, so a secret (a `stdin` field) stays out of argv.
pub fn input_arg(word: Option<&String>) -> Result<Option<String>, String> {
    let read = |what: &str, got: std::io::Result<String>| got.map_err(|e| format!("input {what}: {e}"));
    match word.map(String::as_str) {
        Some("-") => read("-", std::io::read_to_string(std::io::stdin())).map(Some),
        Some(w) if w.starts_with('@') => read(w, std::fs::read_to_string(&w[1..])).map(Some),
        _ => Ok(word.cloned()),
    }
}

/// The signed-in host every run shares: store, HTTP, session, wasm engine.
#[derive(Clone)]
pub struct Host {
    pub store: Store,
    pub fetch: Arc<dyn Fetch>,
    pub session: Option<Arc<Session>>,
    pub wasm: Arc<Wasm>,
    pub budget: Option<Value>,
    /// `--handler` flags: the caller handlers a composition run gets (handlers.rs)
    pub handlers: Vec<(String, crate::handlers::Spec)>,
    /// this build's transports (crate::transports), made once, and the NSIDs they serve (required hooks)
    pub transports: Transports,
    pub provides: Arc<HashSet<String>>,
    /// ATLIBS_HTTP: the origins com.fedfork.exec.http.fetch may reach, with the headers this host adds
    pub http: Option<Arc<run::Http>>,
    /// ATLIBS_HELD: { name: value }, the held values each run starts with (a composition's root's)
    pub held: Option<Map<String, Value>>,
}

/// A JSON env variable, or @file of it.
fn env_json(name: &str) -> Result<Option<Value>, String> {
    let Some(text) = env(name) else { return Ok(None) };
    let text = match text.strip_prefix('@') {
        Some(path) => std::fs::read_to_string(path).map_err(|e| format!("{name} {path}: {e}"))?,
        None => text,
    };
    serde_json::from_str(&text).map(Some).map_err(|e| format!("{name}: {e}"))
}

impl Host {
    pub async fn new(cli: &Cli) -> Result<Self, String> {
        let budget = match env("ATLIBS_BUDGET") {
            Some(b) => Some(serde_json::from_str::<Value>(&b).map_err(|e| format!("ATLIBS_BUDGET: {e}"))?),
            None => None,
        };
        let transports = crate::transports(&cli.state_dir);
        let provides = Arc::new(atlibs_host::transport::provides(&transports));
        // JSON, or @file of it; its own client, following no redirects (an injected key must not travel)
        let http = match env_json("ATLIBS_HTTP")? {
            Some(value) => {
                let client = reqwest::Client::builder().user_agent("atlibs").redirect(reqwest::redirect::Policy::none()).build().map_err(|e| e.to_string())?;
                let mut http = run::Http::from_json(Arc::new(client), &value).map_err(|e| format!("ATLIBS_HTTP: {e}"))?;
                // "iroh://*" in ATLIBS_HTTP: that transport's URLs
                http.schemes = atlibs_host::transport::fetchers(&transports);
                Some(Arc::new(http))
            }
            None => None,
        };
        let held = match env_json("ATLIBS_HELD")? {
            None => None,
            Some(Value::Object(held)) => Some(held),
            Some(_) => return Err("ATLIBS_HELD must be { name: value }".into()),
        };
        Ok(Self { store: cli.store()?.clone(), fetch: cli.fetch.clone(), session: cli.session().await?, wasm: Wasm::new(WALL_TIME_MS), budget, handlers: vec![], transports, provides, http, held })
    }

    /// "no entry <entry> (<its entries>)" when the module (a composition's: `instance`'s) does not export
    /// `entry`; None also when that can't be told here (no such instance, not wasm: the run says why).
    pub fn no_entry(&self, target: &Target, instance: Option<&str>, entry: &str) -> Option<String> {
        let module = match (target, instance) {
            (Target::Module(module, _), _) => module,
            (Target::Composition { comp, .. }, Some(instance)) => comp.modules.get(instance)?,
            _ => return None,
        };
        let entries = self.wasm.entries(&module.bytes).ok()?;
        (!entries.iter().any(|e| e == entry)).then(|| format!("no entry {entry} ({})", entries.join(", ")))
    }

    /// The `call` for one module: hostCall (held to its session's scope when signed in), with attest when the module lists calls (a module that
    /// lists none may call anything, so it gets no signing key), its did:key (sealed under `name`:
    /// the module's identity, or composition#instance), and the transports this CLI was built with.
    pub async fn call_for(&self, module: &Module, name: &str) -> Result<Arc<dyn Call>, String> {
        atlibs_host::registry::module_call(&self.store, self.fetch.clone(), self.session.clone(), &self.transports, module, name).await
    }

    /// A hook default of `module`, as registry::default_lib resolves it (`pinned`: only from
    /// that dir's lockfile), with its own call (call_for: its calls, its attest and did:key, never `module`'s).
    pub fn hook_loader(&self, module: &Module, pinned: Option<PathBuf>) -> Loader {
        let (host, id) = (self.clone(), module.id.clone().unwrap_or_default());
        Arc::new(move |key: String| {
            let (host, id, pinned) = (host.clone(), id.clone(), pinned.clone());
            Box::pin(async move {
                let lib = registry::default_lib(&registry::net_from_env(host.fetch.clone()), &id, pinned.as_deref(), &key).await?;
                let module = module_of(&lib)?;
                let call = host.call_for(&module, &lib.id).await?;
                let runner: Arc<dyn Runner> = Arc::new(Bound { wasm: host.wasm.clone(), bytes: module.bytes.clone(), sha256: module.sha256.clone() });
                Ok((module, runner, call))
            })
        })
    }

    /// One run: a module's entry, or `instance`'s entry in a composition. The signed-in DID is input.did.
    pub async fn run(&self, target: &Target, instance: Option<&str>, entry: &str, input: Map<String, Value>, signal: Signal) -> Result<Value, String> {
        let mut with_did = Map::new();
        if let Some(s) = &self.session {
            with_did.insert("did".into(), json!(s.did()));
        }
        with_did.extend(input);
        let options = Options { signal: Some(signal), budget: self.budget.clone(), provides: Some(self.provides.clone()), http: self.http.clone(), held: self.held.clone(), ..Default::default() };
        let ledger = Some(self.store.ledger("managed_records"));
        match target {
            Target::Module(..) if !self.handlers.is_empty() => Err("--handler supplies a composition's handlers; this ref is one module".into()),
            Target::Module(module, pinned) => {
                // did:keys are keyed by the module's identity (its at:// URI or absolute path), not the name it
                // gives itself: a module must not claim another's key (and ingress host) by naming itself so.
                let call = self.call_for(module, module.id.as_deref().unwrap_or_default()).await?;
                let ctx = Ctx { runner: Arc::new(Bound { wasm: self.wasm.clone(), bytes: module.bytes.clone(), sha256: module.sha256.clone() }), ledger, now: None };
                let options = Options { load: Some(self.hook_loader(module, pinned.clone())), ..options };
                Ok(run_module(&ctx, module, entry, &with_did, call, options).await)
            }
            Target::Composition { id, comp } => {
                let instance = instance.ok_or("a composition runs one of its instances: <ref> <instance> <entry>")?;
                let handlers = crate::handlers::build(self, &comp.manifest, &self.handlers)?;
                // each instance's did:key is sealed under <composition id>#<instance>, so an instance name
                // cannot claim another module's key
                let mut calls = HashMap::new();
                for (name, module) in &comp.modules {
                    calls.insert(name.clone(), self.call_for(module, &format!("{id}#{name}")).await?);
                }
                let comp = Arc::new(Composition { manifest: comp.manifest.clone(), modules: comp.modules.clone() });
                let call_for: compose::CallFor = Arc::new(move |name: &str| calls.get(name).cloned());
                compose::run_compose(comp, self.wasm.clone(), ledger, instance, entry, with_did, call_for, handlers, None, options).await
            }
        }
    }
}

pub async fn main(cli: &Cli, argv: &[String], signal: Signal) -> Result<(), String> {
    let usage = || crate::usage("run");
    let args = crate::Args::parse(argv, &["allow-exec"])?;
    args.only(&["handler", "allow-exec", "params"], "run")?;
    let argv = &args.pos[..];
    if argv.is_empty() {
        return Err(usage());
    }
    let (found, rest) = registry::resolve_ref(&cli.net, Path::new("."), argv).await?;
    let listing = entries(&found);
    let mut target = target_of(found)?;
    let (instance, rest) = if target.is_composition() { (rest.first().map(String::as_str), rest.get(1..).unwrap_or_default()) } else { (None, rest) };
    let (Some(entry), true) = (rest.first(), (1..=2).contains(&rest.len())) else {
        return Err(if rest.is_empty() { format!("{}\n{listing}", usage()) } else { usage() });
    };
    // a composition's parameters, bound for this run (listing its entries needs none)
    if let Target::Composition { comp, .. } = &mut target {
        let given = match input_arg(args.flag("params").map(String::from).as_ref())? {
            Some(text) => serde_json::from_str(&text).map_err(|_| "--params is not JSON".to_string())?,
            None => json!({}),
        };
        comp.manifest = atlibs_host::params::bind(&comp.manifest, &atlibs_host::params::Bind::from_json(&given)?)?;
    } else if args.has("params") {
        return Err("--params binds a composition's parameters; this ref is one module".into());
    }
    let input = input_of(input_arg(rest.get(1))?.as_ref())?;
    let mut host = Host::new(cli).await?;
    host.handlers = crate::handlers::parse(cli, &args.all("handler"), args.has("allow-exec")).await?;
    if let Some(no) = host.no_entry(&target, instance, entry) {
        return Err(no);
    }
    let ran = host.run(&target, instance, entry, input, signal).await?;
    out!("{}", pretty(&ran));
    // the result is printed either way; a run error is the exit status (1) and one stderr line
    match ran.get("error").filter(|e| !e.is_null()) {
        Some(e) => Err(format!("{entry}: {}", e.as_str().map_or_else(|| e.to_string(), String::from))),
        None => Ok(()),
    }
}

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

    #[tokio::test]
    async fn refs_resolve_to_targets_with_their_identity() {
        let pds = Pds::new();
        let uri = pds.publish_module("alice.test", "echo", b"\0asm", None);
        let cli = pds.cli(&tempdir("run"));
        let args: Vec<String> = ["alice.test/echo", "watch", "{}"].iter().map(|s| s.to_string()).collect();
        let (found, rest) = target(&cli.net, &args).await.unwrap();
        let Target::Module(module, _) = found else { panic!("a module") };
        assert_eq!(module.id.as_deref(), Some(uri.as_str()), "did:keys and the ledger key on the at:// URI");
        assert_eq!(rest, &args[1..]);
        let wasm: Vec<String> = vec!["/nope.wasm".into()];
        assert!(target(&cli.net, &wasm).await.err().unwrap().contains("needs a second word"));
        let missing: Vec<String> = vec!["no-such-module".into()];
        assert!(target(&cli.net, &missing).await.err().unwrap().contains("no module no-such-module"));
        assert!(input_of(Some(&"[1]".to_string())).unwrap_err().contains("JSON object"));
        let bad = input_of(Some(&r#"{"stdin":"secret-token""#.to_string())).unwrap_err();
        assert!(bad.starts_with("input is not JSON") && !bad.contains("secret"), "{bad}");
        let file = tempdir("run-input").join("in.json");
        std::fs::write(&file, r#"{"stdin":"tok"}"#).unwrap();
        let read = input_arg(Some(&format!("@{}", file.display()))).unwrap();
        assert_eq!(input_of(read.as_ref()).unwrap()["stdin"], "tok");
        assert_eq!(input_arg(Some(&"{}".to_string())).unwrap().as_deref(), Some("{}"));
    }
}