//! Execution of native futures and WASI processes, including transport drainage.
use crate::{
    Host, NativeContext, NativePlugin,
    process::ProcessInfo,
    runtime::{Instance, InstanceConfig},
};
use serde_json::{Value, json};
use tokio::sync::mpsc;

impl Host {
    /// Run one process and retain its cleanup guard until all transport reads finish.
    pub(crate) async fn run_process(
        &self,
        pid: String,
        info: ProcessInfo,
        component: Option<std::sync::Arc<crate::runtime::Prepared>>,
        native: Option<NativePlugin>,
        incoming: mpsc::UnboundedReceiver<Value>,
        has_id: bool,
    ) -> String {
        if let Some(native) = native {
            let context = NativeContext {
                host: self.clone(),
                pid: pid.clone(),
            };
            let result = native(context, info.method, info.params).await;
            self.answer(&pid, result);
            return "exit 0".into();
        }
        let result = self
            .run_guest(&pid, &info, &component.unwrap(), incoming, has_id)
            .await;
        result.unwrap_or_else(|error| error.to_string())
    }

    /// Drain stdout before deciding whether the guest returned an answer.
    async fn run_guest(
        &self,
        pid: &str,
        info: &ProcessInfo,
        component: &crate::runtime::Prepared,
        mut incoming: mpsc::UnboundedReceiver<Value>,
        has_id: bool,
    ) -> anyhow::Result<String> {
        let (memory, tracker, store_ended) = {
            let state = self.0.state.lock().unwrap();
            let p = state.processes.get(pid).unwrap();
            (p.memory.clone(), p.files.clone(), p.store_ended.clone())
        };
        let config = InstanceConfig {
            filesystem: self.mounts(Some(pid.into()), memory, tracker).await?,
            store_ended,
            network: true,
            env: vec![
                ("SAUSAGE_PID".into(), pid.into()),
                ("SAUSAGE_PLUGIN".into(), info.plugin.clone()),
                (
                    "SAUSAGE_PARENT".into(),
                    info.parent.clone().unwrap_or_else(|| "core".into()),
                ),
            ],
        };
        let mut instance = self
            .0
            .runtime
            .start_logged(
                component,
                config,
                self.0.journal.clone(),
                pid.into(),
                info.plugin.clone(),
            )
            .await?;
        let mut initial = json!({
            "jsonrpc": "2.0",
            "method": info.method,
            "params": info.params,
        });
        if has_id {
            initial["id"] = json!(pid);
        }
        let mut stdin_open = self.forward_input(pid, &mut instance, Some(initial)).await;
        loop {
            tokio::select! {
                line = instance.stdout.recv() => {
                    if !self.stdout_line(pid,line) { break; }
                }
                value = incoming.recv(), if stdin_open => {
                    stdin_open = self.forward_input(pid,&mut instance,value).await;
                }
            }
        }
        Ok(instance.wait().await.to_string())
    }

    /// Close stdin only after the queued responses reach the guest.
    async fn forward_input(
        &self,
        pid: &str,
        instance: &mut Instance,
        value: Option<Value>,
    ) -> bool {
        if let Some(value) = value
            && self.write_input(pid, instance, value).await.is_ok()
        {
            return true;
        }
        instance.stdin.take();
        false
    }
    async fn write_input(
        &self,
        pid: &str,
        instance: &mut Instance,
        value: Value,
    ) -> std::io::Result<()> {
        instance
            .send_line(&value.to_string(), |line| self.output(pid, line))
            .await
    }
    fn stdout_line(&self, pid: &str, line: Option<String>) -> bool {
        if let Some(line) = line {
            self.output(pid, line);
            true
        } else {
            false
        }
    }
}