//! The remote-command translation backend (`DESIGN.md` §15, issue #193).
//!
//! A configured command template — `ssh box claude -p` is one instance, a
//! local CLI or a container exec another — is spawned per batch: the prompt
//! and the cue batch go to its stdin as JSON, the translated batch comes back
//! on stdout in the same `{number, text}` shape every LLM-shaped backend
//! speaks. Nothing is resident; `kill_on_drop` means a command that outlives
//! its call does not either. No assumption about ssh is made here, which is
//! the point: it reaches an agent CLI on another host with no API key in
//! arr's config at all.
//!
//! The template may carry `{source}` and `{target}` placeholders, replaced
//! with the batch's language tags before spawning, so a wrapper script can
//! route on them. Tokens split on whitespace — an ssh-style template has no
//! quoted arguments to worry about.
//!
//! [`translate::translate`] still owns chunking and validation: non-zero
//! exit, timeout and unparseable output fail loudly here, and a reply with
//! the wrong cue count or numbering never becomes a subtitle.

use std::{
    io::ErrorKind,
    process::Stdio,
    sync::{atomic::AtomicU64, atomic::Ordering, Arc},
    time::Duration,
};

use serde::{Deserialize, Serialize};
use tokio::{
    io::{AsyncReadExt, AsyncWriteExt},
    process::Command as Process,
    time,
};

use arr_core::Language;

use crate::translate::{
    strip_code_fence, system_prompt, Backend, BackendId, Batch, Error, Result, TranslateFuture,
    TranslatedCue,
};

const BACKEND_NAME: &str = "command";

/// How a language reads in a command argument: a short tag a wrapper script
/// can route on — `pt-PT`, not "European Portuguese (pt-PT)".
fn language_tag(language: &Language) -> &str {
    match language {
        Language::PortuguesePortugal => "pt-PT",
        Language::PortugueseBrazil => "pt-BR",
        Language::PortugueseUnverified => "pt",
        Language::Other(tag) => tag,
    }
}
/// How long one command may run when the configuration does not say.
pub const DEFAULT_TIMEOUT: Duration = Duration::from_secs(120);

/// Settings for the remote-command backend (DESIGN.md §10, issue #198).
///
/// The template is not a credential, but it names a host — it arrives from
/// bootstrap config like the API keys do, never from the database. The
/// timeout is the *initial* value only: the running backend re-reads a
/// shared cell per batch ([`Command::timeout_cell`]), so the operator's edit
/// of `remote_command_timeout_seconds` applies without a restart (#219).
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct CommandConfig {
    /// The command to run per batch, e.g. `ssh box claude -p`. May carry
    /// `{source}` and `{target}` placeholders.
    pub template: String,
    /// How long one command may run before it is killed and the batch fails.
    pub timeout: Duration,
}

/// The remote-command backend behind the [`Backend`] trait.
#[derive(Clone, Debug)]
pub struct Command {
    config: CommandConfig,
    id: BackendId,
    /// The live per-batch timeout in milliseconds, shared with whoever wired
    /// this backend to the settings row. Milliseconds so the tests — and an
    /// impatient operator — can go below one second.
    timeout_ms: Arc<AtomicU64>,
}

impl Command {
    /// A backend running `config.template` once per batch.
    ///
    /// # Errors
    ///
    /// [`Error::Malformed`] when the template splits into no tokens at all.
    pub fn new(config: CommandConfig) -> Result<Self> {
        if config.template.trim().is_empty() {
            return Err(Error::Malformed {
                backend: BackendId::new(BACKEND_NAME),
                detail: "empty command template".to_owned(),
            });
        }
        Ok(Self {
            timeout_ms: Arc::new(AtomicU64::new(
                u64::try_from(config.timeout.as_millis()).unwrap_or(u64::MAX),
            )),
            config,
            id: BackendId::new(BACKEND_NAME),
        })
    }

    /// The cell this backend re-reads for every batch, in milliseconds.
    ///
    /// Whoever constructs the backend — the daemon, at startup — hands a
    /// clone of this to the settings API, which stores the row's
    /// `remote_command_timeout_seconds` into it on every edit. That is the
    /// path from the database to the running process #219 found missing.
    #[must_use]
    pub fn timeout_cell(&self) -> Arc<AtomicU64> {
        Arc::clone(&self.timeout_ms)
    }

    async fn translate_inner(&self, batch: &Batch) -> Result<Vec<TranslatedCue>> {
        let request_cues: Vec<WireCue> = batch
            .cues
            .iter()
            .map(|cue| WireCue {
                number: cue.number,
                text: cue.text.clone(),
            })
            .collect();
        let batch_json = serde_json::to_string(&request_cues).map_err(|err| Error::Malformed {
            backend: self.id.clone(),
            detail: format!("could not encode batch: {err}"),
        })?;
        let payload = format!(
            "{}\n{batch_json}\n",
            system_prompt(&batch.source, &batch.target)
        );

        let mut tokens = argv(&self.config.template);
        for token in &mut tokens {
            *token = token
                .replace("{source}", language_tag(&batch.source))
                .replace("{target}", language_tag(&batch.target));
        }
        let (program, args) = tokens.split_first().expect("the constructor rejects empty");

        // Re-read per call: the settings API writes this cell when the
        // operator edits `remote_command_timeout_seconds` (#219).
        let timeout = Duration::from_millis(self.timeout_ms.load(Ordering::Relaxed));
        let work = run(program, args, payload, self.id.clone());

        // Dropping `work` on timeout drops the child, whose `kill_on_drop`
        // ends the command — nothing is left resident (DESIGN.md §15).
        match time::timeout(timeout, work).await {
            Err(_) => Err(Error::Transport {
                backend: self.id.clone(),
                source: format!("timed out after {timeout:?}").into(),
            }),
            Ok(Err(err)) => Err(err),
            Ok(Ok(out)) => Self::parse_reply(&self.id, &out),
        }
    }

    fn parse_reply(id: &BackendId, out: &[u8]) -> Result<Vec<TranslatedCue>> {
        let malformed = |detail: String| Error::Malformed {
            backend: id.clone(),
            detail,
        };

        let text = std::str::from_utf8(out)
            .map_err(|err| malformed(format!("output was not UTF-8: {err}")))?;
        if text.trim().is_empty() {
            return Err(malformed("produced no output".to_owned()));
        }

        let cues: Vec<WireCue> = serde_json::from_str(strip_code_fence(text))
            .map_err(|err| malformed(format!("reply was not the expected JSON: {err}")))?;

        Ok(cues
            .into_iter()
            .map(|cue| TranslatedCue {
                number: cue.number,
                text: cue.text,
            })
            .collect())
    }
}

/// Split a template into program and arguments on whitespace.
fn argv(template: &str) -> Vec<String> {
    template.split_whitespace().map(str::to_owned).collect()
}

/// The batch shape on both wires: out on stdin, back on stdout.
#[derive(Serialize, Deserialize)]
struct WireCue {
    number: usize,
    text: String,
}

impl Backend for Command {
    fn id(&self) -> BackendId {
        self.id.clone()
    }

    /// True for every target except [`Language::PortugueseUnverified`], same
    /// rule as the `OpenAI` backend: the template usually drives an agent CLI,
    /// which can be told pt-PT from pt-BR but cannot aim at an unverified
    /// variant that names no exact target.
    fn supports(&self, target: &Language) -> bool {
        !matches!(target, Language::PortugueseUnverified)
    }

    fn translate<'a>(&'a self, batch: &'a Batch) -> TranslateFuture<'a> {
        Box::pin(async move { self.translate_inner(batch).await })
    }
}

/// Spawn the command once, feed it `payload`, and collect its stdout.
///
/// The child is killed when this future is dropped — including on timeout.
async fn run(program: &str, args: &[String], payload: String, id: BackendId) -> Result<Vec<u8>> {
    // Overlay filesystems (Docker, some CI) can report ETXTBSY for a moment
    // after an executable is written; retry briefly instead of failing a
    // batch over it.
    let mut child = loop {
        match Process::new(program)
            .args(args)
            .stdin(Stdio::piped())
            .stdout(Stdio::piped())
            .stderr(Stdio::piped())
            .kill_on_drop(true)
            .spawn()
        {
            Ok(child) => break child,
            Err(err) if err.kind() == ErrorKind::ExecutableFileBusy => {
                time::sleep(Duration::from_millis(10)).await;
            }
            Err(err) => {
                return Err(Error::Transport {
                    backend: id.clone(),
                    source: Box::new(err),
                })
            }
        }
    };

    let mut stdout = child.stdout.take().expect("stdout was piped");
    let mut stderr = child.stderr.take().expect("stderr was piped");

    // Drained concurrently: a chatty stderr must not be able to fill its
    // pipe and deadlock the child while we wait on stdout.
    let stderr_task = tokio::spawn(async move {
        let mut buf = Vec::new();
        let _ignored = stderr.read_to_end(&mut buf).await;
        buf
    });

    {
        let mut stdin = child.stdin.take().expect("stdin was piped");
        stdin
            .write_all(payload.as_bytes())
            .await
            .map_err(|err| Error::Transport {
                backend: id.clone(),
                source: Box::new(err),
            })?;
        stdin.flush().await.map_err(|err| Error::Transport {
            backend: id.clone(),
            source: Box::new(err),
        })?;
        // Dropped here rather than merely shut down: the child sees EOF on
        // stdin only when our end of the pipe actually closes.
    }

    let transport = |err: std::io::Error| Error::Transport {
        backend: id.clone(),
        source: Box::new(err),
    };

    let mut out = Vec::new();
    stdout.read_to_end(&mut out).await.map_err(transport)?;
    let status = child.wait().await.map_err(transport)?;
    let stderr = stderr_task
        .await
        .unwrap_or_else(|_| b"stderr lost".to_vec());

    if !status.success() {
        return Err(Error::Transport {
            backend: id,
            source: format!(
                "exited {}: {}",
                status,
                String::from_utf8_lossy(&stderr).trim()
            )
            .into(),
        });
    }

    Ok(out)
}

#[cfg(test)]
mod tests {
    use super::{argv, Command, CommandConfig};
    use crate::translate::Error;

    #[test]
    fn an_empty_template_is_refused() {
        assert!(matches!(
            Command::new(CommandConfig {
                template: "   ".to_owned(),
                timeout: super::DEFAULT_TIMEOUT,
            }),
            Err(Error::Malformed { .. })
        ));
    }

    #[test]
    fn the_template_splits_on_whitespace() {
        assert_eq!(argv("ssh box claude -p"), ["ssh", "box", "claude", "-p"]);
        assert_eq!(argv("  claude  "), ["claude"]);
        assert!(argv("   ").is_empty());
    }
}