//! Names from a program.
//!
//! This is how a deployment reaches anything that is not a word list: a local
//! model asked to name an agent after what it will be working on, a corporate
//! naming service, a script that reads from a database of names already spoken
//! for. The contract is the smallest one that can carry that: a JSON object on
//! standard input, a name on standard output.
//!
//! It is a program the operator configured, so it runs with their privileges
//! and this crate does not pretend otherwise. What it does insist on is that
//! the program cannot hang provisioning: there is a timeout, it bounds the
//! whole call rather than only the wait for the child to exit, and a name that
//! arrives after it does not count.

use std::io::{BufRead, BufReader, Read, Write};
use std::process::{Command, Stdio};
use std::sync::mpsc;
use std::time::{Duration, Instant};

use crate::{check, tidy, NameError, Namer, Seed};

/// How long a namer gets before it is killed, unless told otherwise.
///
/// Generous by the standards of a request path and mean by the standards of a
/// language model, which is the tension: an agent cannot start work until it
/// has an account, so a namer that thinks for a minute has stopped being a
/// naming service and become an outage.
pub const DEFAULT_TIMEOUT: Duration = Duration::from_secs(3);

/// How often the child is checked while waiting for it.
const POLL: Duration = Duration::from_millis(20);

/// The most standard output a namer may produce before it is cut off.
///
/// A name is at most [`crate::MAX_NAME`] octets and only the first line of it
/// is read, so this is already three orders of magnitude more than the
/// contract asks for. It is a bound rather than a budget: the reader below
/// runs on its own thread so that a child cannot stall it, and a thread
/// reading without a limit is one a runaway namer can grow memory with.
const MAX_OUTPUT: u64 = 64 * 1024;

/// A namer that runs a program.
#[derive(Debug, Clone)]
pub struct CommandNamer {
    program: String,
    args: Vec<String>,
    timeout: Duration,
}

impl CommandNamer {
    /// Names by running `program` with `args`.
    pub fn new(program: impl Into<String>, args: Vec<String>) -> Self {
        Self {
            program: program.into(),
            args,
            timeout: DEFAULT_TIMEOUT,
        }
    }

    /// Splits a shell-ish command line into a program and its arguments.
    ///
    /// Whitespace only: no quoting, no expansion, no shell. A configuration
    /// value that reached a shell would make every deployment's naming service
    /// an injection surface for whatever ends up in the seed.
    pub fn parse(line: &str) -> Result<Self, NameError> {
        let mut words = line.split_whitespace().map(str::to_owned);
        let program = words
            .next()
            .ok_or_else(|| NameError::Misconfigured("empty namer command".to_owned()))?;
        Ok(Self::new(program, words.collect()))
    }

    /// How long to wait before killing the child.
    pub fn with_timeout(mut self, timeout: Duration) -> Self {
        self.timeout = timeout;
        self
    }

    /// Runs the program and reads what it says, killing it if it overruns.
    fn ask(&self, request: &str) -> Result<String, NameError> {
        let mut child = Command::new(&self.program)
            .args(&self.args)
            .stdin(Stdio::piped())
            .stdout(Stdio::piped())
            .stderr(Stdio::null())
            .spawn()
            .map_err(|err| NameError::Failed(format!("{}: {err}", self.program)))?;

        if let Some(mut stdin) = child.stdin.take() {
            // A namer that never reads its input closes the pipe, and writing
            // to a closed pipe is not a failure to name — it is a namer that
            // did not need to be told anything.
            let _ = stdin.write_all(request.as_bytes());
        }

        // Standard output is drained on its own thread, and only as far as the
        // first newline. Waiting for the child and then reading the pipe looks
        // equivalent and is not: a pipe stays open while *any* process holds
        // its write end, so a namer that leaves something behind it — `model &
        // echo name`, a wrapper script that backgrounds its work — exits at
        // once, satisfies the wait, and then blocks the read for as long as
        // its leftovers live. That read is on the provisioning request path,
        // and the timeout below has already been declared satisfied by then.
        //
        // Stopping at the newline is not only how that is bounded, it is the
        // whole of what the contract promises to read; see [`Namer::suggest`]
        // below, which has always discarded everything after the first line.
        let mut stdout = child.stdout.take();
        let (sender, receiver) = mpsc::channel();
        std::thread::spawn(move || {
            let mut line = Vec::new();
            if let Some(stdout) = stdout.take() {
                let _ = BufReader::new(stdout.take(MAX_OUTPUT)).read_until(b'\n', &mut line);
            }
            // The receiver is gone when the deadline passed first, which is an
            // overrunning namer rather than anything to report here.
            let _ = sender.send(line);
        });

        let deadline = Instant::now() + self.timeout;
        let overran = || {
            NameError::Failed(format!(
                "{} did not answer within {:?}",
                self.program, self.timeout
            ))
        };
        let status = loop {
            match child.try_wait() {
                Ok(Some(status)) => break status,
                Ok(None) if Instant::now() >= deadline => {
                    let _ = child.kill();
                    let _ = child.wait();
                    return Err(overran());
                }
                Ok(None) => std::thread::sleep(POLL),
                Err(err) => return Err(NameError::Failed(format!("{}: {err}", self.program))),
            }
        };

        // The child is gone; the pipe may not be. The same deadline covers the
        // read, so the call as a whole is bounded by the timeout whatever is
        // still holding the write end open.
        let Ok(line) = receiver.recv_timeout(deadline.saturating_duration_since(Instant::now()))
        else {
            let _ = child.kill();
            return Err(overran());
        };

        if !status.success() {
            return Err(NameError::Failed(format!(
                "{} exited with {status}",
                self.program
            )));
        }
        Ok(String::from_utf8_lossy(&line).into_owned())
    }
}

impl Namer for CommandNamer {
    fn suggest(&self, seed: &Seed) -> Result<String, NameError> {
        let request = serde_json::json!({
            "token": seed.token,
            "zone": seed.zone,
            "parent": seed.parent,
            "attempt": seed.attempt,
        })
        .to_string();

        let spoken = self.ask(&request)?;
        // The first line only. A model told to answer with a name will often
        // answer with a name and then explain itself.
        let first = spoken.lines().next().unwrap_or_default();
        let name = tidy(first);
        if name.is_empty() {
            return Err(NameError::Empty);
        }
        check(&name)?;
        Ok(name)
    }

    fn describe(&self) -> String {
        format!("command `{}`", self.program)
    }
}