//! Process metadata, process lifetime, and attached children.
use crate::{Host, RpcResult};
use serde_json::{Value, json};
use tokio::sync::{mpsc, oneshot, watch};
#[derive(Clone, serde::Serialize)]
/// A snapshot of process metadata for inspection and the VFS.
pub struct ProcessInfo {
    /// The plugin that supplied the process code.
    pub plugin: String,
    /// The SHA-256 of its component, or native for a native plugin.
    pub version: String,
    /// The method without its plugin prefix.
    pub method: String,
    /// The object supplied by the caller.
    pub params: Value,
    /// The attached parent, or no parent for a detached process.
    pub parent: Option<String>,
    /// Whether this process survives the end of its caller.
    pub detached: bool,
    /// The start time in milliseconds since the Unix epoch.
    pub start_time: u128,
    /// The current lifecycle state.
    pub state: String,
}
pub(crate) struct Process {
    pub(crate) info: ProcessInfo,
    pub(crate) memory: sausage_vfs::Memory,
    pub(crate) files: sausage_vfs::Process,
    pub(crate) store_ended: watch::Sender<bool>,
    pub(crate) input: Option<mpsc::UnboundedSender<Value>>,
    pub(crate) abort: Option<tokio::task::AbortHandle>,
    pub(crate) caller: String,
    pub(crate) id: Option<Value>,
    pub(crate) answer: Option<oneshot::Sender<RpcResult>>,
    pub(crate) ended: watch::Sender<bool>,
    pub(crate) stopped: bool,
    pub(crate) superseded: bool,
}
impl Host {
    /// Called only after the execution task ends. Wait for the store and file operations before releasing locks.
    pub(crate) async fn finish(&self, pid: &str, status: &str) {
        let (files, mut store_ended) = {
            let mut state = self.0.state.lock().unwrap();
            let Some(p) = state.processes.get_mut(pid) else {
                return;
            };
            p.info.state = "stopping".into();
            (p.files.clone(), p.store_ended.subscribe())
        };
        let _ = store_ended.wait_for(|ended| *ended).await;
        files.idle().await;
        let mut state = self.0.state.lock().unwrap();
        let process = state.processes.remove(pid);
        if let Some(p) = process {
            self.0.journal.write(pid, &p.info.plugin, "core", status);
            let logs = p
                .answer
                .as_ref()
                .map(|_| self.0.journal.query_later(json!({"pid":pid,"limit":20})));
            self.0.journal.forget(pid);
            Self::release_locks(&mut state, pid);
            self.service_ended(&mut state, pid, status, p.stopped, p.superseded);
            let status = status.to_string();
            tokio::spawn(async move {
                if let Some(answer) = p.answer {
                    let lines = logs.unwrap().await.unwrap_or_else(|_| json!([]));
                    let failure = json!({
                        "code": -32002,
                        "message": "process ended without an answer",
                        "data": { "status": status, "lines": lines }
                    });
                    let _ = answer.send(Err(failure));
                }
                p.ended.send_replace(true);
            });
            let children: Vec<_> = state
                .processes
                .iter()
                .filter(|(_, p)| p.info.parent.as_deref() == Some(pid))
                .map(|(id, _)| id.clone())
                .collect();
            drop(state);
            for child in children {
                self.begin_stop(&child);
            }
        }
    }
    /// Close stdin and start the same five-second deadline for each attached child.
    pub(crate) fn begin_stop(&self, pid: &str) -> Option<watch::Receiver<bool>> {
        self.stop_process(pid, false)
    }
    /// Grace expiry is retirement, not an operator stopping the service.
    pub(crate) fn begin_retire(&self, pid: &str) -> Option<watch::Receiver<bool>> {
        self.stop_process(pid, true)
    }
    fn stop_process(&self, pid: &str, retirement: bool) -> Option<watch::Receiver<bool>> {
        let mut state = self.0.state.lock().unwrap();
        let p = state.processes.get_mut(pid)?;
        if !retirement {
            p.superseded = false;
        }
        let ended = p.ended.subscribe();
        if p.info.state == "stopping" {
            return Some(ended);
        }
        p.info.state = "stopping".into();
        p.stopped = true;
        p.input.take();
        self.0.journal.write(pid, &p.info.plugin, "core", "stop");
        for lock in state.locks.values_mut() {
            lock.queue.retain(|(queued, _)| queued != pid);
        }
        let children: Vec<_> = state
            .processes
            .iter()
            .filter(|(_, p)| p.info.parent.as_deref() == Some(pid))
            .map(|(id, _)| id.clone())
            .collect();
        drop(state);
        for child in children {
            self.begin_stop(&child);
        }
        let h = self.clone();
        let pid = pid.to_string();
        tokio::spawn(async move {
            tokio::time::sleep(std::time::Duration::from_secs(5)).await;
            h.kill_process(&pid, retirement);
        });
        Some(ended)
    }
    /// Abort a process task and perform its ended-state cleanup.
    pub(crate) fn kill(&self, pid: &str) {
        self.kill_process(pid, false);
    }
    fn kill_process(&self, pid: &str, retirement: bool) {
        let mut state = self.0.state.lock().unwrap();
        for lock in state.locks.values_mut() {
            lock.queue.retain(|(queued, _)| queued != pid);
        }
        if let Some(p) = state.processes.get_mut(pid) {
            if !retirement {
                p.superseded = false;
            }
            p.stopped = true;
            p.info.state = "stopping".into();
            p.input.take();
            self.0.journal.write(pid, &p.info.plugin, "core", "kill");
            if let Some(a) = &p.abort {
                a.abort();
            }
        }
    }
    /// Stop every process in parallel and wait for journal writes to finish.
    pub async fn stop(&self) {
        {
            let _state = self.0.state.lock().unwrap();
            self.0
                .shutdown
                .store(true, std::sync::atomic::Ordering::SeqCst);
        }
        loop {
            let ids: Vec<_> = self.processes().into_keys().collect();
            if ids.is_empty() {
                break;
            }
            let mut waits = Vec::new();
            for pid in ids {
                if let Some(n) = self.begin_stop(&pid) {
                    waits.push(n);
                }
            }
            for mut n in waits {
                let _ = n.wait_for(|ended| *ended).await;
            }
        }
        self.0.journal.query(json!({"limit":0})).await;
    }
}