//! JSON-RPC routing and responses between callers and processes.
use crate::{
    Host, NativeFuture, RpcResult, error,
    process::{Process, ProcessInfo},
};
use serde_json::{Value, json};
use tokio::sync::{mpsc, oneshot, watch};
use uuid::Uuid;

fn params_not_object(params: &Value) -> Value {
    let kind = match params {
        Value::String(_) => "a string",
        Value::Array(_) => "an array",
        Value::Number(_) => "a number",
        Value::Null => "null",
        Value::Bool(_) => "a boolean",
        Value::Object(_) => "an object",
    };
    error(
        -32602,
        &format!(
            "params must be a JSON object, got {kind}. Send the object itself, not a JSON-encoded string."
        ),
    )
}

impl Host {
    /// Queue an incoming response or notification for its destination.
    pub(crate) fn send(&self, pid: &str, value: Value) {
        if pid == "console" {
            let _ = self.0.notifications.send(value);
        } else if let Some(tx) = self
            .0
            .state
            .lock()
            .unwrap()
            .processes
            .get(pid)
            .and_then(|p| p.input.as_ref())
        {
            let _ = tx.send(value);
        }
    }
    /// Remap progress to the caller's original request ID.
    pub(crate) fn notify(&self, pid: &str, name: &str, params: Value) {
        let target = self
            .0
            .state
            .lock()
            .unwrap()
            .processes
            .get(pid)
            .and_then(|p| p.answer.as_ref().map(|_| (p.caller.clone(), p.id.clone())));
        if let Some((caller, Some(id))) = target {
            self.call_notify(&caller, &id, name, params);
        }
    }
    /// Deliver a notification on an open core call using the same envelope as plugin progress.
    pub(crate) fn call_notify(&self, caller: &str, id: &Value, name: &str, params: Value) {
        self.send(
            caller,
            json!({
                "jsonrpc":"2.0", "method":"core.notify",
                "params": {"call":id, "name":name, "params":params}
            }),
        );
    }
    /// Validate and route a request, then start its native or WASI process.
    pub(crate) fn route(
        &self,
        caller: String,
        id: Option<Value>,
        method: String,
        params: Value,
        detached: bool,
    ) -> NativeFuture {
        self.route_from(caller, id, method, params, detached, None)
    }
    /// Like `route`, but a WASI process runs the component at `wasm` instead
    /// of the current `plugin.wasm` of its plugin. Activation uses this to
    /// call `rpc.discover` on a candidate file.
    pub(crate) fn route_from(
        &self,
        caller: String,
        id: Option<Value>,
        method: String,
        params: Value,
        detached: bool,
        wasm: Option<std::path::PathBuf>,
    ) -> NativeFuture {
        let host = self.clone();
        Box::pin(async move {
            if !params.is_object() {
                return Err(params_not_object(&params));
            }
            if caller != "console" && caller != "core" {
                let state = host.0.state.lock().unwrap();
                let Some(process) = state.processes.get(&caller) else {
                    return Err(error(-32600, "process ended"));
                };
                if process.info.state != "running" && method != "core.notify" {
                    return Err(error(-32600, "process is stopping"));
                }
            }
            let (plugin, method) = method
                .split_once('.')
                .ok_or_else(|| error(-32601, "method not found"))?;
            if plugin == "core" {
                return host.core(&caller, id, method, params).await;
            }
            if plugin.is_empty()
                || !plugin
                    .bytes()
                    .all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-' || b == b'_')
            {
                return Err(error(-32601, "invalid plugin name"));
            }
            let path = wasm.unwrap_or_else(|| host.0.world.plugin_dir(plugin).join("plugin.wasm"));
            let (native, component, version, admission) = loop {
                if host.0.shutdown.load(std::sync::atomic::Ordering::SeqCst) {
                    return Err(error(-32600, "host is stopping"));
                }
                let native = host.0.state.lock().unwrap().native.get(plugin).cloned();
                let loaded = if native.is_none() {
                    Some(
                        host.0
                            .components
                            .load(path.clone(), host.0.runtime.clone())
                            .await?,
                    )
                } else {
                    None
                };
                let admission = host.0.admission.read().await;
                if let Some(loaded) = loaded {
                    if crate::components::Fingerprint::read(&path).await? != loaded.fingerprint {
                        drop(admission);
                        continue;
                    }
                    break (native, Some(loaded.prepared), loaded.hash, admission);
                }
                break (native, None, "native".into(), admission);
            };
            let pid = format!("{plugin}-{}", Uuid::now_v7());
            let (tx, rx) = oneshot::channel();
            let (input, incoming) = mpsc::unbounded_channel::<Value>();
            let info = ProcessInfo {
                plugin: plugin.into(),
                version,
                method: method.into(),
                params: params.clone(),
                parent: if detached { None } else { Some(caller.clone()) },
                detached,
                start_time: std::time::SystemTime::now()
                    .duration_since(std::time::UNIX_EPOCH)
                    .unwrap()
                    .as_millis(),
                state: "running".into(),
            };
            {
                let mut state = host.0.state.lock().unwrap();
                if host.0.shutdown.load(std::sync::atomic::Ordering::SeqCst) {
                    return Err(error(-32600, "host is stopping"));
                }
                if caller != "console"
                    && caller != "core"
                    && !state
                        .processes
                        .get(&caller)
                        .is_some_and(|p| p.info.state == "running")
                {
                    return Err(error(-32600, "caller stopped or ended"));
                }
                // Older plugins may still try to start their boot service from
                // rpc.retired. The supervisor owns it even while a restart is
                // pending, or after it has given up. Do not create a second owner.
                if state
                    .processes
                    .get(&caller)
                    .is_some_and(|p| p.info.method == "rpc.retired")
                    && state
                        .services
                        .iter()
                        .any(|s| s.method == format!("{plugin}.{method}") && s.params == params)
                {
                    return Ok(Value::Null);
                }
                if caller == "core"
                    && let Some(service) = state
                        .services
                        .iter_mut()
                        .find(|s| Some(&s.request) == id.as_ref())
                {
                    if !service.enabled || service.owner.is_some() {
                        return Err(error(-32600, "service attempt is no longer current"));
                    }
                    service.owner = Some(pid.clone());
                    service.pending = false;
                    service.version = Some(info.version.clone());
                }
                host.0
                    .journal
                    .write(&pid, plugin, "core", &format!("start {method} {params}"));
                state.processes.insert(
                    pid.clone(),
                    Process {
                        info: info.clone(),
                        memory: sausage_vfs::Memory::default(),
                        files: sausage_vfs::Process::default(),
                        store_ended: watch::channel(true).0,
                        input: Some(input),
                        abort: None,
                        caller,
                        id: id.clone(),
                        answer: if id.is_some() { Some(tx) } else { None },
                        ended: watch::channel(false).0,
                        stopped: false,
                        superseded: false,
                    },
                );
            }
            let h = host.clone();
            let p = pid.clone();
            let has_id = id.is_some();
            let (start, ready) = oneshot::channel();
            let task = tokio::spawn(async move {
                let _ = ready.await;
                h.run_process(p, info, component, native, incoming, has_id)
                    .await
            });
            if let Some(p) = host.0.state.lock().unwrap().processes.get_mut(&pid) {
                p.abort = Some(task.abort_handle());
                if p.info.state == "stopping" {
                    task.abort();
                }
            } else {
                task.abort();
            }
            drop(admission);
            let cleanup = host.clone();
            let finished_pid = pid.clone();
            tokio::spawn(async move {
                let status = task.await.unwrap_or_else(|error| {
                    if error.is_cancelled() {
                        "killed".into()
                    } else {
                        format!("host task failed: {error}")
                    }
                });
                cleanup.finish(&finished_pid, &status).await;
            });
            let _ = start.send(());
            if id.is_none() {
                return Ok(Value::Null);
            }
            rx.await
                .unwrap_or_else(|_| Err(error(-32002, "process ended")))
        })
    }
    /// Complete the initial call once and discard late responses.
    pub(crate) fn answer(&self, pid: &str, result: RpcResult) {
        if let Some(tx) = self
            .0
            .state
            .lock()
            .unwrap()
            .processes
            .get_mut(pid)
            .and_then(|p| p.answer.take())
        {
            let _ = tx.send(result);
        }
    }
    /// Classify a stdout line and route requests or the initial response.
    pub(crate) fn output(&self, pid: &str, line: String) {
        if line.trim().is_empty() {
            return;
        }
        let Ok(v) = serde_json::from_str::<Value>(&line) else {
            self.discard(pid);
            return;
        };
        if !v.is_object() || v["jsonrpc"] != "2.0" {
            self.discard(pid);
            return;
        }
        if let Some(method) = v["method"].as_str() {
            if method == "core.notify" || method == "core.unlock" {
                // Deliver progress before a following response removes its destination.
                let params = v.get("params").cloned().unwrap_or(json!({}));
                let result = if !params.is_object() {
                    Err(params_not_object(&params))
                } else if method == "core.unlock" {
                    if let Some(path) = params["path"].as_str() {
                        self.unlock(pid, path);
                        Ok(Value::Null)
                    } else {
                        Err(error(-32602, "path must be a string"))
                    }
                } else if let Some(name) = params["name"].as_str() {
                    self.notify(
                        pid,
                        name,
                        params.get("params").cloned().unwrap_or(json!({})),
                    );
                    Ok(Value::Null)
                } else {
                    Err(error(-32602, "name must be a string"))
                };
                if let Some(id) = v.get("id") {
                    let response = match result {
                        Ok(result) => json!({"jsonrpc":"2.0", "id":id, "result":result}),
                        Err(error) => json!({"jsonrpc":"2.0", "id":id, "error":error}),
                    };
                    self.send(pid, response);
                }
                return;
            }
            let h = self.clone();
            let caller = pid.to_string();
            let method = method.to_string();
            tokio::spawn(async move {
                let id = v.get("id").cloned();
                let result = h
                    .route(
                        caller.clone(),
                        id.clone(),
                        method,
                        v.get("params").cloned().unwrap_or(json!({})),
                        false,
                    )
                    .await;
                if let Some(id) = id {
                    let value = match result {
                        Ok(result) => json!({
                            "jsonrpc": "2.0",
                            "id": id,
                            "result": result
                        }),
                        Err(error) => json!({
                            "jsonrpc": "2.0",
                            "id": id,
                            "error": error
                        }),
                    };
                    h.send(&caller, value);
                }
            });
        } else if v["id"] == pid {
            self.answer(
                pid,
                if let Some(e) = v.get("error").filter(|e| !e.is_null()) {
                    Err(e.clone())
                } else {
                    Ok(v["result"].clone())
                },
            );
        }
    }
    pub(crate) fn discard(&self, pid: &str) {
        // The journal deduplicates this warning for each process.
        if let Some(plugin) = self
            .0
            .state
            .lock()
            .unwrap()
            .processes
            .get(pid)
            .map(|p| p.info.plugin.clone())
        {
            self.0.journal.warning(pid, &plugin);
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    /// PROTOCOL 6: an unlock from stdout completes before output returns.
    #[tokio::test]
    async fn stdout_unlock_is_synchronous() {
        let folder = std::env::temp_dir().join(format!("sausage-unlock-{}", Uuid::now_v7()));
        crate::world::World::create(&folder).unwrap();
        let host = Host::start_with_cache(&folder, &folder.join("cache"))
            .await
            .unwrap();
        host.lock("core", "ordered", true).await.unwrap();
        host.output(
            "core",
            r#"{"jsonrpc":"2.0","method":"core.unlock","params":{"path":"ordered"}}"#.into(),
        );
        let held = host.lock("console", "ordered", false).await.unwrap();
        host.stop().await;
        std::fs::remove_dir_all(folder).unwrap();
        assert_eq!(held, json!({"held":true}));
    }
}