//! The Podman runtime: one fresh container per call, with the limits of `[runner]`. Whatever the
//! tool prints is the result's content; every failure is a fixed sentence, and what Podman itself
//! said goes only to `brokerd`'s log. M3b spec, section 6.

use std::ffi::OsString;
use std::os::unix::fs::{DirBuilderExt, PermissionsExt};
use std::path::{Path, PathBuf};
use std::process::{Child, Command, ExitStatus, Stdio};
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, Instant};

use crate::config::Runner;
use crate::pipes::{GRACE, Io};
use crate::podman;
use crate::runner::{RunError, RunOutput, RunSpec, Runtime};

pub const RUNBOOK: &str = "see docs/runbook.md#runner-unavailable";
pub const COULD_NOT_RUN: &str = "the tool could not run";
pub const CANNOT_START: &str = "the tool runner could not start the container";
pub const KILLED: &str = "the tool was stopped: it ran out of memory or was killed";
pub const TIMED_OUT: &str = "the tool ran past its time limit";
pub const OUTPUT_OPEN: &str = "the tool left its output open";
pub const UNEXPECTED: &str = "the tool failed with an unexpected status";
/// How often a running container is checked.
pub const POLL: Duration = Duration::from_millis(50);
/// How much of Podman's standard error is kept for the log.
pub const STDERR_KEPT: usize = 4096;
/// How long `run` waits for the egress proxy to make its socket.
pub const EGRESS_WAIT: Duration = Duration::from_secs(5);
/// How often the egress proxy's socket is checked while waiting.
pub const EGRESS_POLL: Duration = Duration::from_millis(20);

/// A log sink. Call it as `(self.log)("a line")`.
pub type Log = Arc<dyn Fn(&str) + Send + Sync>;

pub struct Podman {
    runner: Runner,
    egress_dir: PathBuf,
    egress_wait: Duration,
    log: Log,
    next: AtomicU64,
}

impl Podman {
    /// `egress_dir` is `Config::egress_dir()`; task 12 uses it.
    pub fn new(runner: Runner, egress_dir: PathBuf, log: Log) -> Podman {
        Podman {
            runner,
            egress_dir,
            egress_wait: EGRESS_WAIT,
            log,
            next: AtomicU64::new(0),
        }
    }

    /// The same runtime with another wait for the proxy's socket, for tests.
    pub fn with_egress_wait(self, egress_wait: Duration) -> Podman {
        Podman {
            egress_wait,
            ..self
        }
    }

    /// Where `http_fetch` calls get their directories. Public, so the field counts as read.
    pub fn egress_dir(&self) -> &Path {
        &self.egress_dir
    }

    /// Run one short `podman` command (`kill`, `rm -f`) whose result only matters for the log.
    pub(crate) fn podman(&self, args: &[&str]) {
        let what = args.join(" ");
        let podman = &self.runner.podman;
        match Command::new(podman)
            .args(args)
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::null())
            .status()
        {
            Ok(status) if status.success() => {}
            Ok(status) => (self.log)(&format!("brokerd: podman {what} exited {status}")),
            Err(e) => (self.log)(&format!("brokerd: podman {what} failed: {e}")),
        }
    }

    /// The log for a `podman` that cannot be launched at all (task 11 step 3).
    fn cannot_launch(&self, podman: &Path, e: std::io::Error) {
        (self.log)(&format!(
            "brokerd: cannot start {}: {}\n{}",
            podman.display(),
            e,
            RUNBOOK
        ));
    }

    /// The log for a directory that cannot be made (task 12 step 1).
    fn cannot_make(&self, dir: &Path, e: std::io::Error) -> RunError {
        (self.log)(&format!(
            "brokerd: cannot make {}: {e}\n{}",
            dir.display(),
            RUNBOOK
        ));
        RunError::Unavailable(CANNOT_START.to_string())
    }

    /// Wait for `child` until `limit` has passed since `started`. `Some(status)` if it ended;
    /// `None` if it ran too long, after `podman kill <name>`, `podman rm -f <name>`,
    /// `child.kill()` and `child.wait()` (step 5).
    fn wait(
        &self,
        child: &mut Child,
        name: &str,
        started: Instant,
        limit: Duration,
    ) -> Option<ExitStatus> {
        loop {
            if let Ok(Some(status)) = child.try_wait() {
                return Some(status);
            }
            if started.elapsed() >= limit {
                self.podman(&["kill", name]);
                self.podman(&["rm", "-f", name]);
                let _ = child.kill();
                let _ = child.wait();
                return None;
            }
            std::thread::sleep(POLL);
        }
    }
}

impl Podman {
    /// Steps 3 to 7 for one container: spawn, feed and read it, wait, answer. Task 12 calls it too.
    /// Already written: it only joins the helpers below.
    fn run_container(
        &self,
        name: &str,
        args: Vec<OsString>,
        input: String,
        limit: Duration,
    ) -> Result<RunOutput, RunError> {
        let Some(mut child) = self.spawn(args) else {
            return Err(RunError::Unavailable(CANNOT_START.to_string()));
        };
        let started = Instant::now();
        let cap = usize::try_from(self.runner.output_cap).unwrap_or(usize::MAX);
        let io = match Io::start(&mut child, input, cap, STDERR_KEPT) {
            Ok(io) => io,
            Err(e) => {
                // The container may be running: stop it before answering.
                self.podman(&["kill", name]);
                self.podman(&["rm", "-f", name]);
                let _ = child.kill();
                let _ = child.wait();
                (self.log)(&format!(
                    "brokerd: cannot start a thread for {name}: {e}\n{RUNBOOK}"
                ));
                return Err(RunError::Unavailable(CANNOT_START.to_string()));
            }
        };
        let status = self.wait(&mut child, name, started, limit);
        let done = io.finish(GRACE);
        if done.open && status.is_some() {
            (self.log)(&format!(
                "brokerd: {name} ended but something still holds its output; it was abandoned"
            ));
            return Err(RunError::Failed(OUTPUT_OPEN.to_string()));
        }
        self.answer(name, status, done.out, done.truncated, &done.err)
    }

    /// Step 3: spawn `podman` with `args` and all three standard streams piped. On failure, log
    /// `brokerd: cannot start {podman path}: {e}` + "\n" + RUNBOOK and return `None`.
    fn spawn(&self, args: Vec<OsString>) -> Option<Child> {
        let podman = &self.runner.podman;
        match Command::new(podman)
            .args(args)
            .stdin(Stdio::piped())
            .stdout(Stdio::piped())
            .stderr(Stdio::piped())
            .spawn()
        {
            Ok(child) => Some(child),
            Err(e) => {
                self.cannot_launch(podman, e);
                None
            }
        }
    }

    /// Step 7: the answer, by the table in the task. `status` is `None` when the time limit was
    /// passed. `out` is the kept standard output, `err` the kept standard error (for the log only:
    /// it never goes into a `RunError`).
    fn answer(
        &self,
        name: &str,
        status: Option<ExitStatus>,
        out: Vec<u8>,
        truncated: bool,
        err: &str,
    ) -> Result<RunOutput, RunError> {
        match status {
            None => {
                (self.log)(&format!(
                    "brokerd: stopped container {}: it ran past its time limit",
                    name
                ));
                Err(RunError::Failed(TIMED_OUT.to_string()))
            }
            Some(status) => match status.code() {
                Some(0) | Some(1) => Ok(RunOutput {
                    content: String::from_utf8_lossy(&out).into_owned(),
                    truncated,
                }),
                Some(2) => {
                    (self.log)(&format!("brokerd: {name}: the tool could not run: {err:?}"));
                    Err(RunError::Failed(COULD_NOT_RUN.to_string()))
                }
                Some(125..=127) => {
                    (self.log)(&format!(
                        "brokerd: podman could not start {name}: {err:?}\n{RUNBOOK}"
                    ));
                    Err(RunError::Unavailable(CANNOT_START.to_string()))
                }
                Some(137) => Err(RunError::Failed(KILLED.to_string())),
                _ => {
                    (self.log)(&format!(
                        "brokerd: container {name} exited {status}: {err:?}"
                    ));
                    Err(RunError::Failed(UNEXPECTED.to_string()))
                }
            },
        }
    }
}

impl Runtime for Podman {
    /// Step 1 (the container number), then step 2's choice: no egress runs the tool as before,
    /// an egress starts the proxy first and gives the tool the directory. `run_container` is step 3
    /// to 7 for either. Already written.
    fn run(&self, spec: &RunSpec) -> Result<RunOutput, RunError> {
        let n = self.next.fetch_add(1, Ordering::SeqCst);
        let name = podman::container_name(spec.session(), spec.call(), n);
        let input = spec.arguments().canonical_json();
        let limit = self.runner.time_limit(spec.tool());
        match spec.egress() {
            None => {
                let args = podman::tool_args(spec, &self.runner, &name, None);
                self.run_container(&name, args, input, limit)
            }
            Some(hosts) => {
                let dir = self.egress_dir.join(&name);
                let _guard = self.start_egress(&name, &self.egress_dir, &dir, hosts)?;
                let args = podman::tool_args(spec, &self.runner, &name, Some(&dir));
                self.run_container(&name, args, input, limit)
            }
        }
    }
}

impl Podman {
    /// Steps 1 to 4 of the egress proxy's start (task 12). The guard it returns cleans up the
    /// proxy and directory on every path; `run` keeps it alive until the tool's container ends.
    fn start_egress(
        &self,
        name: &str,
        egress_dir: &Path,
        dir: &Path,
        hosts: &[String],
    ) -> Result<EgressGuard<'_>, RunError> {
        // The guard is created before anything can fail, so every return below
        // cleans up the proxy and directory by dropping it.
        let _guard = EgressGuard::new(self, name, dir.to_path_buf());
        // 1. Make the egress directory (and its parents) 0700, then replace the
        //    call's directory if a crash left one behind.
        std::fs::DirBuilder::new()
            .recursive(true)
            .mode(0o700)
            .create(egress_dir)
            .map_err(|e| self.cannot_make(dir, e))?;
        let _ = std::fs::set_permissions(egress_dir, std::fs::Permissions::from_mode(0o700));
        if dir.exists() {
            match std::fs::remove_dir_all(dir) {
                Ok(()) => {}
                Err(ref e) if e.kind() == std::io::ErrorKind::NotFound => {}
                Err(e) => return Err(self.cannot_make(dir, e)),
            }
        }
        std::fs::DirBuilder::new()
            .mode(0o700)
            .create(dir)
            .map_err(|e| self.cannot_make(dir, e))?;

        // 2. Start the proxy, detached. A run that exits non-zero is logged and
        //    unavailable; one that cannot be launched at all uses task 11's log.
        let args = podman::egress_args(&self.runner, name, dir, hosts);
        match Command::new(&self.runner.podman)
            .args(args)
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::piped())
            .output()
        {
            Ok(output) if output.status.success() => {}
            Ok(output) => {
                let stderr = String::from_utf8_lossy(&output.stderr).into_owned();
                (self.log)(&format!(
                    "brokerd: podman could not start {name}-egress: {stderr:?}\n{RUNBOOK}"
                ));
                return Err(RunError::Unavailable(CANNOT_START.to_string()));
            }
            Err(e) => {
                self.cannot_launch(&self.runner.podman, e);
                return Err(RunError::Unavailable(CANNOT_START.to_string()));
            }
        }

        // 3. Wait for the proxy to make its socket, or give up.
        let socket = dir.join("egress.sock");
        let deadline = Instant::now() + self.egress_wait;
        loop {
            if socket.exists() {
                break;
            }
            if Instant::now() >= deadline {
                let ms = self.egress_wait.as_millis();
                (self.log)(&format!(
                    "brokerd: {name}-egress did not make its socket within {ms} ms\n{RUNBOOK}"
                ));
                return Err(RunError::Unavailable(CANNOT_START.to_string()));
            }
            std::thread::sleep(EGRESS_POLL);
        }

        // 4. The tool's container may now run; `run` holds the guard until it
        //    returns, then drops it to remove the proxy and directory.
        Ok(_guard)
    }
}

/// Removes the egress proxy's container and its directory when `run` returns, on every path.
///
/// `run` creates it before anything can fail and holds it until the tool's container has ended, so
/// the proxy and directory never survive the call that started it.
struct EgressGuard<'a> {
    podman: &'a Podman,
    name: String,
    dir: PathBuf,
}

impl<'a> Drop for EgressGuard<'a> {
    fn drop(&mut self) {
        self.podman.podman(&["rm", "-f", self.name.as_str()]);
        match std::fs::remove_dir_all(&self.dir) {
            Ok(()) => {}
            Err(ref e) if e.kind() == std::io::ErrorKind::NotFound => {}
            Err(e) => (self.podman.log)(&format!(
                "brokerd: cannot remove {}: {e}",
                self.dir.display()
            )),
        }
    }
}

impl<'a> EgressGuard<'a> {
    fn new(podman: &'a Podman, name: &str, dir: PathBuf) -> EgressGuard<'a> {
        EgressGuard {
            podman,
            name: format!("{name}-egress"),
            dir,
        }
    }
}