//! Host-side client for the subprocess plugin protocol.
//!
//! Why a stateful host rather than fresh subprocesses per call: spawning
//! a Node or Ruby runtime is on the order of 50-200 ms, which would
//! dwarf the ~10-50 ms per-call budget we promise the user. The host
//! spawns on first use, holds the child for the run's lifetime, and
//! drops the pipes (which makes the child see EOF on stdin) when it is
//! shut down. A polite EOF lets the plugin flush before exit; the
//! [`Drop`] impl follows up with a forced kill so a wedged plugin can
//! never wedge the runner.

use std::collections::HashMap;
use std::io;
use std::path::PathBuf;
use std::process::Stdio;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Duration;

use serde_json::Value;
use thiserror::Error;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::process::{Child, ChildStdin, ChildStdout, Command};
use tokio::time::timeout;

use super::protocol::{PluginRequest, PluginResponse};

/// Default per-call timeout.
///
/// Tuned for the ~10-50 ms steady-state path documented in the protocol
/// spec, with one full second of slack for cold paths (first request
/// after spawn, GC pauses in a Node plugin). Callers that need a longer
/// budget set [`PluginHostConfig::call_timeout`] explicitly.
pub const DEFAULT_CALL_TIMEOUT: Duration = Duration::from_secs(2);

/// Construction parameters for [`PluginHost`].
///
/// Kept as a struct so future knobs (stderr capture limit, max-line
/// size) can land without breaking the builder shape.
#[derive(Debug, Clone)]
pub struct PluginHostConfig {
    /// Program to execute. Looked up via `$PATH` if it is not absolute.
    pub command: String,
    /// Argv after the program name.
    pub args: Vec<String>,
    /// Environment variables overlaid on the parent's environment.
    pub env: HashMap<String, String>,
    /// Working directory. Defaults to the parent's cwd when `None`.
    pub working_dir: Option<PathBuf>,
    /// Maximum wall-clock time the host waits for a response before it
    /// gives up on a single call.
    pub call_timeout: Duration,
}

impl PluginHostConfig {
    /// Build a config from the smallest required argument set.
    ///
    /// Defaults: no extra args, inherit the parent environment, inherit
    /// the parent cwd, and [`DEFAULT_CALL_TIMEOUT`] per call.
    pub fn new(command: impl Into<String>) -> Self {
        Self {
            command: command.into(),
            args: Vec::new(),
            env: HashMap::new(),
            working_dir: None,
            call_timeout: DEFAULT_CALL_TIMEOUT,
        }
    }
}

/// Typed errors the host can surface.
///
/// Splitting protocol-level failure modes (crash, timeout, invalid
/// JSON) into distinct variants lets callers map them onto user-facing
/// messages without parsing strings.
#[derive(Debug, Error)]
#[non_exhaustive]
pub enum HostError {
    /// `tokio::process::Command::spawn` failed (binary missing,
    /// permission denied, etc.).
    #[error("failed to spawn plugin `{command}`: {source}")]
    SpawnFailed {
        /// Command we tried to run.
        command: String,
        /// Underlying OS error.
        #[source]
        source: io::Error,
    },
    /// Spawn succeeded but one of stdin or stdout was not piped, so the
    /// host cannot frame messages.
    #[error("plugin did not expose piped stdin/stdout")]
    MissingPipes,
    /// The plugin exited before (or while) the host was waiting for a
    /// response. Carries the exit status if it could be read.
    #[error("plugin exited before responding: {detail}")]
    PluginCrashed {
        /// Best-effort human description of the exit (status code,
        /// signal, or "stdin closed").
        detail: String,
    },
    /// The plugin did not respond within
    /// [`PluginHostConfig::call_timeout`].
    #[error("plugin timed out after {timeout:?} on `{method}`")]
    Timeout {
        /// Method that did not return in time.
        method: String,
        /// Configured timeout the call exceeded.
        timeout: Duration,
    },
    /// The plugin emitted a line that was not valid JSON. The raw line
    /// is captured so callers can show it to the operator.
    #[error("plugin emitted invalid JSON: {source}: {line:?}")]
    InvalidJson {
        /// Raw line from the plugin's stdout.
        line: String,
        /// Underlying serde error.
        #[source]
        source: serde_json::Error,
    },
    /// The plugin emitted a JSON object that did not satisfy
    /// [`PluginResponse`].
    #[error("plugin response did not match the protocol: {0}")]
    InvalidResponse(String),
    /// Plain I/O failure on the pipes (for example a broken pipe when
    /// the plugin exits mid-write).
    #[error("plugin I/O error: {0}")]
    Io(#[from] io::Error),
    /// `serde_json` could not serialize the outbound request. In
    /// practice this only fires when the params contain values that
    /// serde refuses (NaN floats, non-string map keys).
    #[error("could not encode plugin request: {0}")]
    Encode(#[from] serde_json::Error),
}

/// Spawn-once, talk-many client for a subprocess plugin.
///
/// One instance per plugin per run. The host is `Send` but not `Sync`:
/// the `call` path mutates the framed reader and writer, and we want
/// callers to serialize requests through one mutable borrow rather than
/// silently introducing a mutex.
pub struct PluginHost {
    config: PluginHostConfig,
    child: Option<Child>,
    stdin: Option<ChildStdin>,
    stdout: Option<BufReader<ChildStdout>>,
    next_id: AtomicU64,
}

impl PluginHost {
    /// Construct the host without spawning the plugin.
    ///
    /// The child process is launched on the first [`call`](Self::call)
    /// so a host that is never used costs nothing.
    pub fn new(config: PluginHostConfig) -> Self {
        Self {
            config,
            child: None,
            stdin: None,
            stdout: None,
            next_id: AtomicU64::new(1),
        }
    }

    /// Send a request and await the matching response.
    ///
    /// Spawns the plugin lazily on the first call. The `id` field of
    /// the wire envelope is minted internally so callers do not have to
    /// thread a counter. The returned [`PluginResponse`] is whatever
    /// the plugin sent back, including its `error` field; callers
    /// decide whether a plugin-reported error is fatal for their use.
    pub async fn call(
        &mut self,
        method: &str,
        params: Option<Value>,
    ) -> Result<PluginResponse, HostError> {
        self.ensure_spawned()?;
        let id_num = self.next_id.fetch_add(1, Ordering::Relaxed);
        let request = PluginRequest {
            method: method.to_owned(),
            params,
            id: Value::from(id_num),
        };
        let line = serde_json::to_vec(&request)?;
        self.write_line(&line).await?;
        let raw = self.read_line_with_timeout(method).await?;
        parse_response(&raw)
    }

    /// Close the plugin politely.
    ///
    /// Drops the stdin pipe so the plugin sees EOF, then waits up to
    /// the call timeout for the child to exit on its own. A child that
    /// ignores EOF gets killed. Idempotent: calling shutdown twice or
    /// after a crash is a no-op.
    pub async fn shutdown(&mut self) -> Result<(), HostError> {
        self.stdin.take();
        self.stdout.take();
        let Some(mut child) = self.child.take() else {
            return Ok(());
        };
        let wait = timeout(self.config.call_timeout, child.wait()).await;
        match wait {
            Ok(Ok(_status)) => Ok(()),
            Ok(Err(err)) => Err(HostError::Io(err)),
            Err(_) => {
                let _ = child.start_kill();
                let _ = child.wait().await;
                Ok(())
            }
        }
    }

    fn ensure_spawned(&mut self) -> Result<(), HostError> {
        if self.child.is_some() {
            return Ok(());
        }
        let mut cmd = Command::new(&self.config.command);
        cmd.args(&self.config.args)
            .envs(&self.config.env)
            .stdin(Stdio::piped())
            .stdout(Stdio::piped())
            .stderr(Stdio::null())
            .kill_on_drop(true);
        if let Some(dir) = &self.config.working_dir {
            cmd.current_dir(dir);
        }
        let mut child = cmd.spawn().map_err(|source| HostError::SpawnFailed {
            command: self.config.command.clone(),
            source,
        })?;
        let stdin = child.stdin.take().ok_or(HostError::MissingPipes)?;
        let stdout = child.stdout.take().ok_or(HostError::MissingPipes)?;
        self.stdin = Some(stdin);
        self.stdout = Some(BufReader::new(stdout));
        self.child = Some(child);
        Ok(())
    }

    async fn write_line(&mut self, payload: &[u8]) -> Result<(), HostError> {
        let stdin = self.stdin.as_mut().ok_or(HostError::PluginCrashed {
            detail: "stdin closed".into(),
        })?;
        if let Err(err) = stdin.write_all(payload).await {
            return Err(self.classify_pipe_error(err).await);
        }
        if let Err(err) = stdin.write_all(b"\n").await {
            return Err(self.classify_pipe_error(err).await);
        }
        if let Err(err) = stdin.flush().await {
            return Err(self.classify_pipe_error(err).await);
        }
        Ok(())
    }

    async fn read_line_with_timeout(&mut self, method: &str) -> Result<String, HostError> {
        let call_timeout = self.config.call_timeout;
        let mut line = String::new();
        let read_outcome = {
            let stdout = self.stdout.as_mut().ok_or(HostError::PluginCrashed {
                detail: "stdout closed".into(),
            })?;
            timeout(call_timeout, stdout.read_line(&mut line)).await
        };
        match read_outcome {
            Ok(Ok(0)) => Err(self.harvest_exit_detail().await),
            Ok(Ok(_)) => Ok(line),
            Ok(Err(err)) => Err(self.classify_pipe_error(err).await),
            Err(_) => Err(HostError::Timeout {
                method: method.to_owned(),
                timeout: call_timeout,
            }),
        }
    }

    async fn classify_pipe_error(&mut self, err: io::Error) -> HostError {
        if matches!(
            err.kind(),
            io::ErrorKind::BrokenPipe | io::ErrorKind::UnexpectedEof
        ) {
            self.harvest_exit_detail().await
        } else {
            HostError::Io(err)
        }
    }

    async fn harvest_exit_detail(&mut self) -> HostError {
        let detail = if let Some(mut child) = self.child.take() {
            match timeout(Duration::from_millis(250), child.wait()).await {
                Ok(Ok(status)) => format!("exit status {status}"),
                Ok(Err(err)) => format!("wait failed: {err}"),
                Err(_) => {
                    let _ = child.start_kill();
                    "exit pending (killed)".to_owned()
                }
            }
        } else {
            "process already reaped".to_owned()
        };
        self.stdin.take();
        self.stdout.take();
        HostError::PluginCrashed { detail }
    }
}

impl Drop for PluginHost {
    fn drop(&mut self) {
        // `kill_on_drop(true)` on the Command builder hands the cleanup
        // to tokio: when the Child handle drops, tokio schedules a kill
        // on the runtime. We keep this Drop impl explicit so a future
        // reader sees the lifecycle in one place.
        self.stdin.take();
        self.stdout.take();
        self.child.take();
    }
}

fn parse_response(raw: &str) -> Result<PluginResponse, HostError> {
    let trimmed = raw.trim();
    if trimmed.is_empty() {
        return Err(HostError::InvalidResponse(
            "plugin sent an empty line".into(),
        ));
    }
    let response: PluginResponse =
        serde_json::from_str(trimmed).map_err(|source| HostError::InvalidJson {
            line: trimmed.to_owned(),
            source,
        })?;
    if response.result.is_none() && response.error.is_none() {
        return Err(HostError::InvalidResponse(
            "response had neither result nor error".into(),
        ));
    }
    if response.result.is_some() && response.error.is_some() {
        return Err(HostError::InvalidResponse(
            "response had both result and error".into(),
        ));
    }
    Ok(response)
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;

    #[test]
    fn parse_response_rejects_empty_line() {
        let err = parse_response("").expect_err("empty must error");
        assert!(matches!(err, HostError::InvalidResponse(_)));
    }

    #[test]
    fn parse_response_rejects_both_result_and_error() {
        let raw = r#"{"id":1,"result":1,"error":{"code":-1,"message":"x"}}"#;
        let err = parse_response(raw).expect_err("ambiguous must error");
        assert!(matches!(err, HostError::InvalidResponse(_)));
    }

    #[test]
    fn parse_response_rejects_neither_result_nor_error() {
        let raw = r#"{"id":1}"#;
        let err = parse_response(raw).expect_err("empty body must error");
        assert!(matches!(err, HostError::InvalidResponse(_)));
    }

    #[test]
    fn parse_response_surfaces_invalid_json() {
        let err = parse_response("{not json").expect_err("bad json must error");
        match err {
            HostError::InvalidJson { line, .. } => assert!(line.starts_with("{not")),
            other => panic!("expected InvalidJson, got {other:?}"),
        }
    }

    #[test]
    fn parse_response_accepts_well_formed_success() {
        let raw = r#"{"id":42,"result":{"ok":true}}"#;
        let resp = parse_response(raw).expect("parse success");
        assert_eq!(resp.id, json!(42));
        assert_eq!(resp.result, Some(json!({"ok": true})));
        assert!(resp.error.is_none());
    }

    #[test]
    fn parse_response_accepts_well_formed_failure() {
        let raw = r#"{"id":"a","error":{"code":-32601,"message":"no"}}"#;
        let resp = parse_response(raw).expect("parse failure");
        assert_eq!(resp.id, json!("a"));
        assert!(resp.result.is_none());
        let err = resp.error.expect("error populated");
        assert_eq!(err.code, -32601);
    }

    #[test]
    fn config_constructor_sets_default_timeout() {
        let cfg = PluginHostConfig::new("echo");
        assert_eq!(cfg.call_timeout, DEFAULT_CALL_TIMEOUT);
        assert!(cfg.args.is_empty());
    }
}
