// Caller handlers for compositions on the CLI (a manifest's `handlers`). Only the command
// line supplies them, never a manifest or lockfile, each one named by the composition:
//
//   --handler <name>=wasm:<ref>#<entry>   another module's entry, run with that module's own grants and
//                                         identity; it may call caps passed in its input (callbacks)
//   --handler <name>=https://host/path    POST the call as JSON, the response JSON is the answer
//                                         (http:// only to loopback); no credentials are sent
//   --handler <name>=exec:<command>       `sh -c <command>`: the call as JSON on stdin, the answer as JSON
//                                         on stdout; needs --allow-exec; runs with an empty environment
//                                         (PATH, HOME, LANG only: no app password, no tokens)
//
// What the http and exec handlers get: { handler, caller, nsid, reason, input } (caps in input stay
// { "$cap": token } and cannot be called from there).

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

use atlibs_host::compose::{Handler, Info};
use atlibs_host::run::{Call, Ctx, Module, Options, Request, Sub, run_module};
use atlibs_host::wasm::Bound;
use atlibs_host::{BoxFut, js_string};
use serde_json::{Map, Value, json};

use crate::run::{Host, Target, target};

const EXEC_OUTPUT_BYTES: usize = 1 << 20;

#[derive(Clone)]
pub enum Spec {
    Wasm { module: Arc<Module>, entry: String },
    Http { url: String },
    Exec { command: String },
}

/// `--handler` values: name=spec. exec needs `allow_exec`; the names are checked against the manifest
/// when the composition runs.
pub async fn parse(cli: &crate::Cli, values: &[&str], allow_exec: bool) -> Result<Vec<(String, Spec)>, String> {
    let mut out: Vec<(String, Spec)> = vec![];
    for value in values {
        let (name, spec) = value.split_once('=').ok_or_else(|| format!("--handler {value}: use <name>=wasm:<ref>#<entry> | https://... | exec:<command>"))?;
        if name.is_empty() || name.len() > 256 || !name.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'_' || b == b'-') {
            return Err(format!("--handler {value}: the name must match /^[A-Za-z0-9_-]{{1,256}}$/"));
        }
        if out.iter().any(|(n, _)| n == name) {
            return Err(format!("--handler {name} given twice"));
        }
        let spec = if let Some(command) = spec.strip_prefix("exec:") {
            if !allow_exec {
                return Err(format!("--handler {name}=exec:... runs a local command: pass --allow-exec to allow it"));
            }
            if command.trim().is_empty() {
                return Err(format!("--handler {name}: exec needs a command"));
            }
            Spec::Exec { command: command.into() }
        } else if let Some(rest) = spec.strip_prefix("wasm:") {
            let (reference, entry) = rest.rsplit_once('#').ok_or_else(|| format!("--handler {name}: wasm:<ref>#<entry>"))?;
            if !atlibs_host::wasm::entry_ok(entry) {
                return Err(format!("--handler {name}: {entry} is not an export name"));
            }
            let words = [reference.to_string()];
            let (found, rest) = target(&cli.net, &words).await.map_err(|e| format!("--handler {name}: {e}"))?;
            let Target::Module(module, _) = found else { return Err(format!("--handler {name}: {reference} is a composition; a handler is one module's entry")) };
            if !rest.is_empty() {
                return Err(format!("--handler {name}: {reference} needs one word"));
            }
            Spec::Wasm { module: Arc::new(module), entry: entry.into() }
        } else if spec.starts_with("https://") || spec.starts_with("http://") {
            let url = reqwest::Url::parse(spec).map_err(|e| format!("--handler {name}: {e}"))?;
            let host_name = url.host_str().unwrap_or_default().trim_matches(['[', ']']);
            let loopback = host_name == "localhost" || host_name.parse::<std::net::IpAddr>().is_ok_and(|ip| ip.is_loopback());
            if url.scheme() == "http" && !loopback {
                return Err(format!("--handler {name}: http:// only to localhost; use https://"));
            }
            Spec::Http { url: url.to_string() }
        } else {
            return Err(format!("--handler {name}={spec}: use wasm:<ref>#<entry>, https://..., or exec:<command>"));
        };
        out.push((name.to_string(), spec));
    }
    Ok(out)
}

/// The handlers a composition run gets: each supplied name must be one the manifest declares.
pub fn build(host: &Host, manifest: &Value, specs: &[(String, Spec)]) -> Result<HashMap<String, Handler>, String> {
    let declared: Vec<String> = manifest["handlers"].as_array().into_iter().flatten().map(|h| js_string(h.get("name"))).collect();
    let mut handlers = HashMap::new();
    for (name, spec) in specs {
        if !declared.contains(name) {
            return Err(format!("--handler {name}: the composition declares no handler {name} (it declares: {})", if declared.is_empty() { "none".into() } else { declared.join(", ") }));
        }
        let (host, key, spec) = (host.clone(), name.clone(), spec.clone());
        let handler: Handler = Arc::new(move |input: Value, info: Info| {
            let (host, name, spec) = (host.clone(), key.clone(), spec.clone());
            Box::pin(async move { answer(&host, &name, &spec, input, info).await })
        });
        handlers.insert(name.clone(), handler);
    }
    Ok(handlers)
}

fn envelope(name: &str, input: &Value, info: &Info) -> Value {
    json!({ "handler": name, "caller": info.caller, "nsid": info.nsid, "reason": info.reason, "input": input })
}

async fn answer(host: &Host, name: &str, spec: &Spec, input: Value, info: Info) -> Result<Value, String> {
    let signal = info.signal.clone();
    let work = async {
        match spec {
            Spec::Wasm { module, entry } => wasm(host, module, entry, input, info).await,
            Spec::Http { url } => http(host, name, url, &envelope(name, &input, &info)).await,
            Spec::Exec { command } => exec(name, command, &envelope(name, &input, &info)).await,
        }
    };
    tokio::select! { out = work => out, reason = signal.aborted() => Err(reason) }
}

/// Cap tokens anywhere in a value.
fn tokens(value: &Value, out: &mut HashSet<String>) {
    match value {
        Value::Object(m) if m.len() == 1 && m.get("$cap").and_then(Value::as_str).is_some_and(|t| t.starts_with("cap.")) => {
            out.insert(m["$cap"].as_str().unwrap_or_default().to_string());
        }
        Value::Object(m) => m.values().for_each(|v| tokens(v, out)),
        Value::Array(a) => a.iter().for_each(|v| tokens(v, out)),
        _ => {}
    }
}

/// The handler module's call: the caps it was handed go back to the composition; anything else is
/// its own host call (its module.json grants, its identity).
struct CapCall {
    info: Info,
    held: HashSet<String>,
    inner: Arc<dyn Call>,
}

impl Call for CapCall {
    fn call(&self, nsid: String, request: Request) -> BoxFut<'_, Result<Option<Value>, String>> {
        if self.held.contains(&nsid) {
            let invoked = self.info.invoke(&nsid, request.input.unwrap_or(Value::Null));
            return Box::pin(async move { invoked.await.map(Some) });
        }
        self.inner.call(nsid, request)
    }
    fn subscribe(&self, nsid: String, request: Request) -> BoxFut<'_, Result<Sub, String>> {
        self.inner.subscribe(nsid, request)
    }
}

async fn wasm(host: &Host, module: &Module, entry: &str, input: Value, info: Info) -> Result<Value, String> {
    let mut run_input = Map::new();
    if let Some(s) = &host.session {
        run_input.insert("did".into(), json!(s.did()));
    }
    match input {
        Value::Object(m) => run_input.extend(m),
        other => drop(run_input.insert("input".into(), other)),
    }
    run_input.insert("caller".into(), json!({ "instance": info.caller, "nsid": info.nsid, "reason": info.reason }));
    let mut held = HashSet::new();
    tokens(&Value::Object(run_input.clone()), &mut held);
    let signal = info.signal.clone();
    let inner = host.call_for(module, module.id.as_deref().unwrap_or_default()).await?;
    // its allowlist stays its own, plus the caps it was handed (holding the token is the grant)
    let mut module = module.clone();
    if let Some(calls) = module.calls.as_mut() {
        calls.extend(held.iter().cloned());
    }
    let module = &module;
    let call: Arc<dyn Call> = Arc::new(CapCall { info, held, inner });
    let ctx = Ctx {
        runner: Arc::new(Bound { wasm: host.wasm.clone(), bytes: module.bytes.clone(), sha256: module.sha256.clone() }),
        ledger: Some(host.store.ledger("managed_records")),
        now: None,
    };
    let ran = run_module(&ctx, module, entry, &run_input, call, Options { signal: Some(signal), budget: host.budget.clone(), provides: Some(host.provides.clone()), http: host.http.clone(), ..Default::default() }).await;
    match ran.get("error") {
        Some(e) if !e.is_null() => Err(js_string(Some(e))),
        _ => Ok(ran.get("result").cloned().unwrap_or(Value::Null)),
    }
}

async fn http(host: &Host, name: &str, url: &str, body: &Value) -> Result<Value, String> {
    let response = host.fetch.fetch("POST", url, vec![("content-type".into(), "application/json".into())], Some(body.to_string().into_bytes())).await?;
    let text = String::from_utf8_lossy(&response.body);
    if !(200..300).contains(&response.status) {
        return Err(format!("handler {name}: {url} answered {}: {}", response.status, text.chars().take(300).collect::<String>()));
    }
    if text.trim().is_empty() {
        return Ok(Value::Null);
    }
    serde_json::from_str(&text).map_err(|_| format!("handler {name}: {url} did not answer JSON"))
}

async fn exec(name: &str, command: &str, body: &Value) -> Result<Value, String> {
    use tokio::io::AsyncWriteExt;
    let mut cmd = tokio::process::Command::new("sh");
    cmd.arg("-c").arg(command).env_clear();
    for key in ["PATH", "HOME", "LANG"] {
        if let Ok(v) = std::env::var(key) {
            cmd.env(key, v);
        }
    }
    let mut child = cmd
        .stdin(std::process::Stdio::piped())
        .stdout(std::process::Stdio::piped())
        .stderr(std::process::Stdio::piped())
        .kill_on_drop(true)
        .spawn()
        .map_err(|e| format!("handler {name}: {e}"))?;
    let mut stdin = child.stdin.take().expect("piped");
    let data = format!("{body}\n");
    tokio::spawn(async move {
        let _ = stdin.write_all(data.as_bytes()).await; // the command may not read it all
    });
    let out = child.wait_with_output().await.map_err(|e| format!("handler {name}: {e}"))?;
    if !out.status.success() {
        let stderr = String::from_utf8_lossy(&out.stderr);
        let tail: String = stderr.trim().chars().rev().take(300).collect::<Vec<_>>().into_iter().rev().collect();
        return Err(format!("handler {name}: {command} exited {}{}", out.status.code().map_or("on a signal".into(), |c| c.to_string()), if tail.is_empty() { String::new() } else { format!(": {tail}") }));
    }
    if out.stdout.len() > EXEC_OUTPUT_BYTES {
        return Err(format!("handler {name}: output exceeds {EXEC_OUTPUT_BYTES} bytes"));
    }
    let text = String::from_utf8_lossy(&out.stdout);
    if text.trim().is_empty() {
        return Ok(Value::Null);
    }
    serde_json::from_str(text.trim()).map_err(|_| format!("handler {name}: {command} did not print JSON"))
}

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

    async fn host_for(dir: &std::path::Path) -> (crate::Cli, Host) {
        let cli = Pds::new().cli(dir);
        let host = Host::new(&cli).await.unwrap();
        (cli, host)
    }

    #[tokio::test]
    async fn specs_are_explicit_and_checked() {
        let dir = tempdir("handlers");
        let (cli, host) = host_for(&dir).await;
        let parse = |v: &'static [&'static str], allow: bool| {
            let cli = &cli;
            async move { parse(cli, v, allow).await }
        };
        assert!(parse(&["confirm=exec:cat"], false).await.err().unwrap().contains("--allow-exec"));
        assert!(matches!(parse(&["confirm=exec:cat"], true).await.unwrap()[0].1, Spec::Exec { .. }));
        assert!(parse(&["confirm=http://example.com/x"], false).await.err().unwrap().contains("only to localhost"));
        assert!(matches!(parse(&["confirm=http://127.0.0.1:9/x"], false).await.unwrap()[0].1, Spec::Http { .. }));
        assert!(matches!(parse(&["confirm=https://h.example/x"], false).await.unwrap()[0].1, Spec::Http { .. }));
        assert!(parse(&["confirm=ftp://x"], false).await.err().unwrap().contains("use wasm:"));
        assert!(parse(&["bad name=exec:x"], true).await.err().unwrap().contains("must match"));
        assert!(parse(&["a=exec:x", "a=exec:y"], true).await.err().unwrap().contains("twice"));
        assert!(parse(&["a=wasm:nowhere"], false).await.err().unwrap().contains("wasm:<ref>#<entry>"));
        assert!(parse(&["a=wasm:no-such-module#go"], false).await.err().unwrap().contains("no module no-such-module"));
        let manifest = json!({ "handlers": [{ "name": "confirm" }] });
        let specs = parse(&["other=exec:cat"], true).await.unwrap();
        assert!(build(&host, &manifest, &specs).err().unwrap().contains("declares no handler other (it declares: confirm)"));
    }

    #[tokio::test]
    async fn exec_gets_the_call_on_stdin_with_an_empty_environment() {
        let body = json!({ "handler": "confirm", "input": { "n": 1 } });
        let echoed = exec("confirm", "cat", &body).await.unwrap();
        assert_eq!(echoed, body);
        // nothing from the host's environment but PATH, HOME and LANG
        let env = exec("confirm", r#"printf '{"vars":"%s"}' "$(env | cut -d= -f1 | sort | tr '\n' ' ')""#, &body).await.unwrap();
        let vars: Vec<&str> = env["vars"].as_str().unwrap().split_whitespace().filter(|v| !["PATH", "HOME", "LANG", "PWD", "SHLVL", "_", "OLDPWD"].contains(v)).collect();
        assert!(vars.is_empty(), "leaked: {vars:?}");
        assert!(exec("confirm", "echo nope >&2; exit 3", &body).await.unwrap_err().ends_with("exited 3: nope"));
        assert!(exec("confirm", "echo not json", &body).await.unwrap_err().contains("did not print JSON"));
        assert_eq!(exec("confirm", "true", &body).await.unwrap(), Value::Null);
    }
}