//! Running a job's `run` script.
//!
//! `BuiltinExecutor` is the default: the image is prepared by an [`ImageSource`] (the
//! [`ImageStore`] in production), and the script runs in a sandbox started by a helper process
//! (`forge-runner __sandbox`). No external tool is needed.
//!
//! `PodmanExecutor` is the optional path: a rootless podman container with the workspace
//! mounted at `/workspace`. Secret values and job env travel in the child process environment,
//! and podman receives only `-e NAME`, so no value ever appears in argv.
//!
//! `HostExecutor` runs the script directly on the host. It exists only for tests and is compiled
//! only under `cfg(test)` or the `test-host-executor` feature.

use std::fs;
use std::future::Future;
use std::path::{Path, PathBuf};
use std::pin::Pin;
use std::process::{ExitStatus, Stdio};
use std::time::Duration;

use forge_ci::proto::{JobAssignment, Outcome};
use nix::sys::signal::{Signal, kill};
use nix::unistd::Pid;
use oci_client::Reference;
use tokio::io::AsyncWriteExt;
use tokio::process::{Child, Command};
use tokio::sync::mpsc;
use tokio::task::JoinHandle;
use tokio_util::sync::CancellationToken;

use crate::config::{Limits, RegistryConfig};
use crate::logs::forward_lines;
use crate::oci::{ImageError, ImageStore, PreparedImage, RegistryAuth};
use crate::sandbox::helper::EXIT_SETUP_FAILED;
use crate::sandbox::idmap::{IdMap, remove_tree};
use crate::sandbox::{SandboxRequest, job_cgroup_path};
use crate::workspace::validate_job_id;

/// The prefix of the sandbox helper's own error lines on stderr.
const SANDBOX_PREFIX: &[u8] = b"forge-sandbox: ";

/// How long to wait for the output readers to drain after the child has exited.
const READER_GRACE: Duration = Duration::from_secs(10);
/// How long to wait for the child to exit after it has been told to stop.
const STOP_GRACE: Duration = Duration::from_secs(30);

/// Everything an executor needs to run one job.
pub struct ExecRequest<'a> {
    pub job: &'a JobAssignment,
    /// The checked-out source tree. The script runs with it as `/workspace` or as its cwd.
    pub workspace: &'a Path,
    /// Where merged stdout and stderr go, one line per message.
    pub logs: mpsc::Sender<Vec<u8>>,
    /// Cancelling this stops the job. The executor then reports `Outcome::Cancelled`.
    pub cancel: CancellationToken,
}

/// The executor's answer. The runner adds timeouts and artifact results on top of it.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ExecResult {
    pub outcome: Outcome,
    pub exit_code: Option<i32>,
    pub message: Option<String>,
}

/// The boxed future an executor returns. Written out so `Executor` stays object safe.
pub type ExecFuture<'a> = Pin<Box<dyn Future<Output = ExecResult> + Send + 'a>>;

/// Runs jobs. The runner holds one as `Arc<dyn Executor>`.
pub trait Executor: Send + Sync {
    fn run<'a>(&'a self, req: ExecRequest<'a>) -> ExecFuture<'a>;
}

/// An executor failure, reported as `Outcome::Error`.
#[derive(Debug, thiserror::Error)]
pub enum ExecError {
    #[error("invalid job: {0}")]
    Invalid(String),
}

/// Env names go to podman as `-e NAME`, so they must be plain identifiers. A name with `=` in it
/// would carry a value into argv.
pub fn validate_env_name(name: &str) -> Result<(), ExecError> {
    let mut chars = name.chars();
    let first_ok = matches!(chars.next(), Some(c) if c.is_ascii_alphabetic() || c == '_');
    let rest_ok = chars.all(|c| c.is_ascii_alphanumeric() || c == '_');
    if first_ok && rest_ok && name.len() <= 256 {
        Ok(())
    } else {
        Err(ExecError::Invalid(format!("bad env name {name:?}")))
    }
}

/// The job's env and secrets as name/value pairs. Secrets win over plain env on a clash.
pub fn job_env(job: &JobAssignment) -> Result<Vec<(String, String)>, ExecError> {
    let mut pairs: Vec<(String, String)> = Vec::new();
    for (k, v) in job.env.iter().chain(job.secrets.iter()) {
        validate_env_name(k)?;
        pairs.retain(|(name, _)| name != k);
        pairs.push((k.clone(), v.clone()));
    }
    Ok(pairs)
}

/// The podman invocation for a job: argv (no values in it) and the env the child needs.
#[derive(Debug, PartialEq, Eq)]
pub struct PodmanInvocation {
    pub args: Vec<String>,
    pub env: Vec<(String, String)>,
    pub container: String,
}

fn check_image(image: &str) -> Result<(), ExecError> {
    // Podman would read a leading `-` as an option, and whitespace is never valid in a reference.
    if image.is_empty() || image.starts_with('-') || image.contains(char::is_whitespace) {
        return Err(ExecError::Invalid(format!("bad image {image:?}")));
    }
    Ok(())
}

fn check_limit(value: &str, what: &str) -> Result<(), ExecError> {
    let ok = !value.is_empty()
        && !value.starts_with('-')
        && value.chars().all(|c| c.is_ascii_alphanumeric() || c == '.');
    if ok {
        Ok(())
    } else {
        Err(ExecError::Invalid(format!("bad {what} limit {value:?}")))
    }
}

/// Builds the podman invocation for `job`. Pure, so the argv can be tested.
pub fn podman_invocation(
    job: &JobAssignment,
    workspace: &Path,
    limits: &Limits,
) -> Result<PodmanInvocation, ExecError> {
    validate_job_id(&job.job_id).map_err(|e| ExecError::Invalid(e.to_string()))?;
    check_image(&job.image)?;
    check_limit(&limits.memory, "memory")?;
    check_limit(&limits.cpus, "cpus")?;
    let ws = workspace
        .to_str()
        .filter(|s| !s.contains(':') && !s.contains(','))
        .ok_or_else(|| ExecError::Invalid("workspace path has a ':' or ','".into()))?;

    let env = job_env(job)?;
    let container = format!("forge-job-{}", job.job_id);
    let network = if job.network { "slirp4netns" } else { "none" };

    let mut args: Vec<String> = [
        "run",
        "--rm",
        "--name",
        &container,
        "--pull=newer",
        &format!("--network={network}"),
        "--memory",
        &limits.memory,
        "--cpus",
        &limits.cpus,
        "--pids-limit",
    ]
    .iter()
    .map(|s| (*s).to_owned())
    .collect();
    args.push(limits.pids.to_string());
    for flag in [
        "--security-opt",
        "no-new-privileges",
        "--cap-drop",
        "ALL",
        "-v",
    ] {
        args.push(flag.to_owned());
    }
    args.push(format!("{ws}:/workspace:Z"));
    args.extend(["-w".to_owned(), "/workspace".to_owned()]);
    for (name, _) in &env {
        args.push("-e".to_owned());
        args.push(name.clone());
    }
    args.push(job.image.clone());
    args.extend([
        "sh".to_owned(),
        "-eu".to_owned(),
        "-c".to_owned(),
        job.run.clone(),
    ]);

    Ok(PodmanInvocation {
        args,
        env,
        container,
    })
}

/// How a finished child maps to a job result.
fn classify(status: ExitStatus) -> ExecResult {
    match status.code() {
        Some(0) => ExecResult {
            outcome: Outcome::Success,
            exit_code: Some(0),
            message: None,
        },
        // podman itself exits 125 when it cannot run the container (no image, no storage, ...).
        Some(125) => ExecResult {
            outcome: Outcome::Error,
            exit_code: Some(125),
            message: Some("podman failed to run the container (exit 125)".to_owned()),
        },
        Some(code) => ExecResult {
            outcome: Outcome::Failure,
            exit_code: Some(code),
            message: None,
        },
        None => ExecResult {
            outcome: Outcome::Failure,
            exit_code: None,
            message: Some("job was terminated by a signal".to_owned()),
        },
    }
}

fn error_result(message: impl Into<String>) -> ExecResult {
    ExecResult {
        outcome: Outcome::Error,
        exit_code: None,
        message: Some(message.into()),
    }
}

/// How to stop a job when it is cancelled.
enum Stopper {
    Podman {
        container: String,
    },
    #[cfg(any(test, feature = "test-host-executor"))]
    ProcessGroup {
        pid: u32,
    },
    /// The sandbox helper. It is sent SIGTERM and stops its own container, see [`stop_child`].
    Helper {
        pid: u32,
    },
}

impl Stopper {
    async fn stop(&self) {
        match self {
            Stopper::Podman { container } => {
                let quiet = |cmd: &mut Command| {
                    cmd.stdin(Stdio::null())
                        .stdout(Stdio::null())
                        .stderr(Stdio::null());
                };
                let mut kill = Command::new("podman");
                kill.args(["kill", container.as_str()]);
                quiet(&mut kill);
                let _ = kill.status().await;
                let mut rm = Command::new("podman");
                rm.args(["rm", "-f", container.as_str()]);
                quiet(&mut rm);
                let _ = rm.status().await;
            }
            #[cfg(any(test, feature = "test-host-executor"))]
            Stopper::ProcessGroup { pid } => {
                let mut kill = Command::new("kill");
                kill.args(["-KILL", "--", &format!("-{pid}")])
                    .stdin(Stdio::null())
                    .stdout(Stdio::null())
                    .stderr(Stdio::null());
                let _ = kill.status().await;
            }
            Stopper::Helper { .. } => {}
        }
    }
}

/// Stops the child on cancellation and returns its exit status.
///
/// The sandbox helper gets SIGTERM and gets to kill and delete its container itself. SIGKILL
/// follows only if it does not exit within [`STOP_GRACE`]. Other children are killed at once.
async fn stop_child(child: &mut Child, stopper: Stopper) -> std::io::Result<ExitStatus> {
    if let Stopper::Helper { pid } = stopper {
        let _ = kill(Pid::from_raw(pid as i32), Signal::SIGTERM);
        return match tokio::time::timeout(STOP_GRACE, child.wait()).await {
            Ok(status) => status,
            Err(_) => {
                let _ = child.start_kill();
                child.wait().await
            }
        };
    }
    stopper.stop().await;
    let _ = child.start_kill();
    tokio::time::timeout(STOP_GRACE, child.wait())
        .await
        .unwrap_or_else(|_| Err(std::io::Error::other("child did not stop")))
}

/// What [`supervise`] saw when the child finished.
struct Supervised {
    cancelled: bool,
    status: std::io::Result<ExitStatus>,
    /// The sandbox helper's last `forge-sandbox: <reason>` line on stderr, when it printed one.
    sandbox_error: Option<String>,
}

/// Copies stderr lines to the log and remembers the last sandbox error line.
async fn tap_sandbox_errors(
    mut lines: mpsc::Receiver<Vec<u8>>,
    logs: mpsc::Sender<Vec<u8>>,
) -> Option<String> {
    let mut found = None;
    while let Some(chunk) = lines.recv().await {
        let is_error = chunk.starts_with(SANDBOX_PREFIX)
            && !chunk[SANDBOX_PREFIX.len()..].starts_with(b"warning:");
        if is_error {
            found = Some(String::from_utf8_lossy(&chunk).trim_end().to_owned());
        }
        if logs.send(chunk).await.is_err() {
            break;
        }
    }
    found
}

/// Forwards the child's output, waits for it, and reports how it ended. On cancellation the
/// child is stopped first. `cleanup` runs as soon as the child has exited, before the output is
/// drained: a container keeps the output pipes open until it is killed.
async fn supervise(
    mut child: Child,
    logs: mpsc::Sender<Vec<u8>>,
    cancel: CancellationToken,
    stopper: Stopper,
    cleanup: Option<JobCleanup>,
) -> Supervised {
    let mut readers: Vec<JoinHandle<()>> = Vec::new();
    if let Some(out) = child.stdout.take() {
        readers.push(tokio::spawn(forward_lines(out, logs.clone())));
    }
    let mut tap = None;
    if let Some(err) = child.stderr.take() {
        let (tx, rx) = mpsc::channel(64);
        readers.push(tokio::spawn(forward_lines(err, tx)));
        tap = Some(tokio::spawn(tap_sandbox_errors(rx, logs.clone())));
    }
    drop(logs);

    // The wait future must be dropped before the stop path borrows the child again.
    let early = tokio::select! {
        status = child.wait() => Some(status),
        _ = cancel.cancelled() => None,
    };
    let (status, cancelled) = match early {
        Some(status) => (status, false),
        None => (stop_child(&mut child, stopper).await, true),
    };
    if let Some(job) = cleanup {
        reap_job(&job).await;
    }

    let drain = async {
        for reader in readers.iter_mut() {
            let _ = reader.await;
        }
    };
    if tokio::time::timeout(READER_GRACE, drain).await.is_err() {
        // A grandchild still holds the pipe. Stop reading rather than hang the job.
        for reader in &readers {
            reader.abort();
        }
    }
    let sandbox_error = match tap {
        Some(tap) => match tokio::time::timeout(READER_GRACE, tap).await {
            Ok(Ok(found)) => found,
            _ => None,
        },
        None => None,
    };

    Supervised {
        cancelled,
        status,
        sandbox_error,
    }
}

/// The result for a child that was supervised with the podman or host rules.
fn finish(supervised: Supervised) -> ExecResult {
    if supervised.cancelled {
        return cancelled_result();
    }
    match supervised.status {
        Ok(status) => classify(status),
        Err(e) => error_result(format!("waiting for the job failed: {e}")),
    }
}

fn cancelled_result() -> ExecResult {
    ExecResult {
        outcome: Outcome::Cancelled,
        exit_code: None,
        message: Some("job was cancelled".to_owned()),
    }
}

/// Runs jobs in rootless podman.
#[derive(Clone, Debug)]
pub struct PodmanExecutor {
    limits: Limits,
}

impl PodmanExecutor {
    pub fn new(limits: Limits) -> Self {
        Self { limits }
    }
}

impl Executor for PodmanExecutor {
    fn run<'a>(&'a self, req: ExecRequest<'a>) -> ExecFuture<'a> {
        Box::pin(async move {
            let inv = match podman_invocation(req.job, req.workspace, &self.limits) {
                Ok(inv) => inv,
                Err(e) => return error_result(e.to_string()),
            };
            let mut cmd = Command::new("podman");
            cmd.args(&inv.args)
                .envs(inv.env.iter().map(|(k, v)| (k.as_str(), v.as_str())))
                .stdin(Stdio::null())
                .stdout(Stdio::piped())
                .stderr(Stdio::piped())
                .kill_on_drop(true);
            let child = match cmd.spawn() {
                Ok(child) => child,
                Err(e) => return error_result(format!("cannot start podman: {e}")),
            };
            finish(
                supervise(
                    child,
                    req.logs,
                    req.cancel,
                    Stopper::Podman {
                        container: inv.container,
                    },
                    None,
                )
                .await,
            )
        })
    }
}

/// Runs the script on the host with no isolation. Tests only.
#[cfg(any(test, feature = "test-host-executor"))]
#[derive(Clone, Copy, Debug, Default)]
pub struct HostExecutor;

#[cfg(any(test, feature = "test-host-executor"))]
impl Executor for HostExecutor {
    fn run<'a>(&'a self, req: ExecRequest<'a>) -> ExecFuture<'a> {
        Box::pin(async move {
            let env = match job_env(req.job) {
                Ok(env) => env,
                Err(e) => return error_result(e.to_string()),
            };
            let mut cmd = Command::new("sh");
            cmd.args(["-eu", "-c", req.job.run.as_str()])
                .current_dir(req.workspace)
                .envs(env.iter().map(|(k, v)| (k.as_str(), v.as_str())))
                .stdin(Stdio::null())
                .stdout(Stdio::piped())
                .stderr(Stdio::piped())
                .process_group(0)
                .kill_on_drop(true);
            let child = match cmd.spawn() {
                Ok(child) => child,
                Err(e) => return error_result(format!("cannot start sh: {e}")),
            };
            let Some(pid) = child.id() else {
                return error_result("child exited before its pid was read");
            };
            finish(
                supervise(
                    child,
                    req.logs,
                    req.cancel,
                    Stopper::ProcessGroup { pid },
                    None,
                )
                .await,
            )
        })
    }
}

/// Where the builtin executor gets its rootfs. [`ImageStore`] pulls and unpacks images. Tests
/// supply a local rootfs directory instead.
pub trait ImageSource: Send + Sync {
    /// Makes `image` available and returns its rootfs and config. Progress lines go to `log`.
    fn prepare<'a>(
        &'a self,
        image: &'a str,
        auth: Option<&'a RegistryAuth>,
        log: &'a mpsc::Sender<Vec<u8>>,
    ) -> impl Future<Output = Result<PreparedImage, ImageError>> + Send + 'a;
}

impl ImageSource for ImageStore {
    fn prepare<'a>(
        &'a self,
        image: &'a str,
        auth: Option<&'a RegistryAuth>,
        log: &'a mpsc::Sender<Vec<u8>>,
    ) -> impl Future<Output = Result<PreparedImage, ImageError>> + Send + 'a {
        self.ensure(image, auth, log)
    }
}

/// Runs jobs in a libcontainer sandbox, started as `helper` (`forge-runner __sandbox`). It needs
/// no external container tool. See `docs/ci.md`.
pub struct BuiltinExecutor<I = ImageStore> {
    images: I,
    limits: Limits,
    helper: PathBuf,
    /// Holds one `<job id>` directory per running job, for the overlay and container state.
    jobs_dir: PathBuf,
    /// `[[registry]]` logins, as (host, login).
    registries: Vec<(String, RegistryAuth)>,
    /// The id map jobs run under. Single unless set by [`BuiltinExecutor::with_idmap`].
    idmap: IdMap,
}

impl<I: ImageSource> BuiltinExecutor<I> {
    pub fn new(images: I, limits: Limits, helper: PathBuf, jobs_dir: PathBuf) -> Self {
        Self {
            images,
            limits,
            helper,
            jobs_dir,
            registries: Vec::new(),
            idmap: IdMap::single(),
        }
    }

    /// Sets the id map that jobs run under. Job dirs are removed under it too.
    pub fn with_idmap(mut self, map: IdMap) -> Self {
        self.idmap = map;
        self
    }

    /// Sets the `[[registry]]` logins. Each one is used for images on its host.
    pub fn with_registries(mut self, registries: &[RegistryConfig]) -> Self {
        self.registries = registries
            .iter()
            .map(|r| {
                (
                    r.host.clone(),
                    RegistryAuth {
                        username: r.username.clone(),
                        password: r.password.clone(),
                    },
                )
            })
            .collect();
        self
    }

    /// The login for `image`, matched by its registry host. Docker Hub is `docker.io`.
    fn auth_for(&self, image: &str) -> Option<&RegistryAuth> {
        let reference: Reference = image.parse().ok()?;
        let host = reference.registry();
        self.registries
            .iter()
            .find(|(registry, _)| registry == host)
            .map(|(_, auth)| auth)
    }

    async fn run_job(&self, req: ExecRequest<'_>) -> ExecResult {
        let ExecRequest {
            job,
            workspace,
            logs,
            cancel,
        } = req;
        if let Err(e) = validate_job_id(&job.job_id) {
            return error_result(e.to_string());
        }
        if let Err(e) = check_image(&job.image) {
            return error_result(e.to_string());
        }
        let env = match job_env(job) {
            Ok(env) => env,
            Err(e) => return error_result(e.to_string()),
        };

        let auth = self.auth_for(&job.image);
        let prepared = tokio::select! {
            prepared = self.images.prepare(&job.image, auth, &logs) => prepared,
            _ = cancel.cancelled() => return cancelled_result(),
        };
        let image = match prepared {
            Ok(image) => image,
            Err(e) => return error_result(format!("cannot prepare image {}: {e}", job.image)),
        };
        let _ = logs
            .send(format!("image {}\n", image.digest).into_bytes())
            .await;

        let job_dir = self.jobs_dir.join(&job.job_id);
        // A leftover from an earlier run of this job id would hold stale overlay state.
        if let Err(e) = remove_job_dir(&self.idmap, &self.helper, &job_dir).await {
            tracing::warn!("cannot remove stale {}: {e}", job_dir.display());
        }
        let request = SandboxRequest {
            job_id: job.job_id.clone(),
            rootfs: image.rootfs,
            job_dir,
            workspace: workspace.to_owned(),
            image_env: image.env,
            env,
            run: job.run.clone(),
            network: job.network,
            limits: self.limits.clone(),
            idmap: self.idmap.clone(),
        };
        // The request holds the secrets. It goes to the helper's stdin and nowhere else.
        let json = match serde_json::to_vec(&request) {
            Ok(json) => json,
            Err(e) => return error_result(format!("cannot encode the sandbox request: {e}")),
        };
        drop(request);

        let mut cmd = Command::new(&self.helper);
        cmd.arg("__sandbox")
            .stdin(Stdio::piped())
            .stdout(Stdio::piped())
            .stderr(Stdio::piped())
            .kill_on_drop(true);
        let mut child = match cmd.spawn() {
            Ok(child) => child,
            Err(e) => return error_result(format!("cannot start the sandbox helper: {e}")),
        };
        let Some(pid) = child.id() else {
            return error_result("sandbox helper exited before its pid was read");
        };
        if let Some(mut stdin) = child.stdin.take() {
            // The helper reads the whole request before it does anything else.
            if let Err(e) = stdin.write_all(&json).await {
                return error_result(format!("cannot send the job to the sandbox helper: {e}"));
            }
        }
        drop(json);

        let Supervised {
            cancelled,
            status,
            sandbox_error,
        } = supervise(
            child,
            logs,
            cancel,
            Stopper::Helper { pid },
            Some(JobCleanup {
                helper_pid: pid,
                job_id: job.job_id.clone(),
                job_dir: self.jobs_dir.join(&job.job_id),
                idmap: self.idmap.clone(),
                helper: self.helper.clone(),
            }),
        )
        .await;
        if cancelled {
            return cancelled_result();
        }
        match status {
            Err(e) => error_result(format!("waiting for the sandbox failed: {e}")),
            Ok(status) => match status.code() {
                Some(0) => ExecResult {
                    outcome: Outcome::Success,
                    exit_code: Some(0),
                    message: None,
                },
                // The helper's own setup failure. A job that exits 125 by itself is a failure.
                Some(EXIT_SETUP_FAILED) if sandbox_error.is_some() => {
                    let line = sandbox_error.unwrap_or_default();
                    let reason = line.strip_prefix("forge-sandbox: ").unwrap_or(&line);
                    ExecResult {
                        outcome: Outcome::Error,
                        exit_code: Some(EXIT_SETUP_FAILED),
                        message: Some(format!("sandbox setup failed: {reason}")),
                    }
                }
                Some(code) => ExecResult {
                    outcome: Outcome::Failure,
                    exit_code: Some(code),
                    message: None,
                },
                None => ExecResult {
                    outcome: Outcome::Failure,
                    exit_code: None,
                    message: Some("job was terminated by a signal".to_owned()),
                },
            },
        }
    }
}

/// Where cgroup2 is mounted. Job cgroup paths are relative to it.
const CGROUP_ROOT: &str = "/sys/fs/cgroup";
/// How often, and how many times, to retry removing a job cgroup while its processes exit.
const REAP_POLL: Duration = Duration::from_millis(100);
const REAP_POLLS: u32 = 50;

/// Kills every process in a job cgroup. `cgroup.kill` does it in one write (Linux 5.14 and up).
/// Older kernels get a SIGKILL to each pid in `cgroup.procs`.
fn kill_cgroup(dir: &Path) {
    if fs::write(dir.join("cgroup.kill"), "1").is_ok() {
        return;
    }
    let Ok(procs) = fs::read_to_string(dir.join("cgroup.procs")) else {
        return;
    };
    for pid in procs
        .lines()
        .filter_map(|line| line.trim().parse::<i32>().ok())
    {
        let _ = kill(Pid::from_raw(pid), Signal::SIGKILL);
    }
}

/// What to clean up once a job's sandbox helper has exited. See [`reap_job`].
struct JobCleanup {
    helper_pid: u32,
    job_id: String,
    job_dir: PathBuf,
    idmap: IdMap,
    helper: PathBuf,
}

/// Cleans up after a job's helper has exited, whatever the reason. A helper killed with SIGKILL
/// cannot delete its container, so the container could otherwise keep running. This kills what is
/// left in the job cgroup, removes the cgroup, and removes the job dir.
async fn reap_job(job: &JobCleanup) {
    if let Ok(relative) = job_cgroup_path(&job.job_id, job.helper_pid) {
        let dir = Path::new(CGROUP_ROOT).join(relative.strip_prefix("/").unwrap_or(&relative));
        if dir.exists() {
            kill_cgroup(&dir);
            // rmdir fails while processes remain, so retry until they have exited.
            for _ in 0..REAP_POLLS {
                if fs::remove_dir(&dir).is_ok() || !dir.exists() {
                    break;
                }
                tokio::time::sleep(REAP_POLL).await;
            }
        }
    }
    if let Err(e) = remove_job_dir(&job.idmap, &job.helper, &job.job_dir).await {
        tracing::warn!("cannot remove {}: {e}", job.job_dir.display());
    }
}

/// Removes a job dir under `map`. Its files may be owned by subordinate ids, so this goes through
/// the helper in subuid mode (see [`remove_tree`]). The removal blocks, so it runs off the async
/// runtime.
async fn remove_job_dir(map: &IdMap, helper: &Path, job_dir: &Path) -> std::io::Result<()> {
    let (map, helper, job_dir) = (map.clone(), helper.to_owned(), job_dir.to_owned());
    tokio::task::spawn_blocking(move || remove_tree(&map, &job_dir, &helper))
        .await
        .map_err(std::io::Error::other)?
}

impl<I: ImageSource> Executor for BuiltinExecutor<I> {
    fn run<'a>(&'a self, req: ExecRequest<'a>) -> ExecFuture<'a> {
        Box::pin(self.run_job(req))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::collections::BTreeMap;

    const SECRET: &str = "hunter2-SUPER-SECRET-value";

    pub(crate) fn fixture_job(run: &str) -> JobAssignment {
        JobAssignment {
            job_id: "job_01".to_owned(),
            job_token: "tok_01_secret_token".to_owned(),
            repo: "owner/name".to_owned(),
            clone_url: "https://forge.example/owner/name.git".to_owned(),
            commit: "a".repeat(40),
            git_ref: "refs/heads/main".to_owned(),
            job_name: "build".to_owned(),
            image: "docker.io/library/alpine:3".to_owned(),
            run: run.to_owned(),
            env: BTreeMap::from([("PLAIN".to_owned(), "visible".to_owned())]),
            secrets: BTreeMap::from([("MY_TOKEN".to_owned(), SECRET.to_owned())]),
            artifacts: Vec::new(),
            timeout_minutes: 5,
            network: false,
        }
    }

    #[test]
    fn podman_argv_never_contains_a_secret_value() {
        let job = fixture_job("echo $MY_TOKEN");
        let inv = podman_invocation(&job, Path::new("/work/job_01"), &Limits::default())
            .expect("invocation");
        assert!(inv.args.iter().all(|a| !a.contains(SECRET)));
        assert!(inv.args.iter().all(|a| !a.contains("visible")));
        // The values travel in the child env, by name.
        assert!(
            inv.env
                .contains(&("MY_TOKEN".to_owned(), SECRET.to_owned()))
        );
        assert!(
            inv.env
                .contains(&("PLAIN".to_owned(), "visible".to_owned()))
        );
        let names: Vec<&str> = inv
            .args
            .windows(2)
            .filter(|w| w[0] == "-e")
            .map(|w| w[1].as_str())
            .collect();
        assert_eq!(names, vec!["PLAIN", "MY_TOKEN"]);
    }

    #[test]
    fn podman_argv_has_the_isolation_flags() {
        let job = fixture_job("true");
        let inv = podman_invocation(&job, Path::new("/work/job_01"), &Limits::default())
            .expect("invocation");
        let joined = inv.args.join(" ");
        assert!(joined.starts_with("run --rm --name forge-job-job_01 --pull=newer"));
        assert!(joined.contains("--network=none"));
        assert!(joined.contains("--memory 4g --cpus 2 --pids-limit 4096"));
        assert!(joined.contains("--security-opt no-new-privileges --cap-drop ALL"));
        assert!(joined.contains("-v /work/job_01:/workspace:Z -w /workspace"));
        assert!(joined.ends_with("docker.io/library/alpine:3 sh -eu -c true"));
    }

    #[test]
    fn network_true_selects_slirp() {
        let mut job = fixture_job("true");
        job.network = true;
        let inv = podman_invocation(&job, Path::new("/w"), &Limits::default()).expect("inv");
        assert!(inv.args.contains(&"--network=slirp4netns".to_owned()));
    }

    #[test]
    fn hostile_values_are_refused() {
        let ws = Path::new("/w");
        let mut job = fixture_job("true");
        job.image = "--privileged".to_owned();
        assert!(podman_invocation(&job, ws, &Limits::default()).is_err());

        let mut job = fixture_job("true");
        job.env.insert("A=B".to_owned(), "x".to_owned());
        assert!(podman_invocation(&job, ws, &Limits::default()).is_err());

        let mut job = fixture_job("true");
        job.job_id = "../escape".to_owned();
        assert!(podman_invocation(&job, ws, &Limits::default()).is_err());

        let job = fixture_job("true");
        assert!(podman_invocation(&job, Path::new("/w:evil"), &Limits::default()).is_err());
    }

    #[test]
    fn exit_codes_map_to_outcomes() {
        use std::os::unix::process::ExitStatusExt;
        let ok = classify(ExitStatus::from_raw(0));
        assert_eq!(ok.outcome, Outcome::Success);
        let fail = classify(ExitStatus::from_raw(3 << 8));
        assert_eq!(fail.outcome, Outcome::Failure);
        assert_eq!(fail.exit_code, Some(3));
        let podman = classify(ExitStatus::from_raw(125 << 8));
        assert_eq!(podman.outcome, Outcome::Error);
    }

    /// Needs rootless podman and network access to pull the image.
    #[tokio::test]
    #[ignore = "requires rootless podman and registry access"]
    async fn podman_runs_a_real_container() {
        let dir = tempfile::tempdir().expect("tempdir");
        let ws = dir.path().join("ws");
        std::fs::create_dir_all(&ws).expect("mkdir");
        let job = fixture_job(
            "echo hi-from-container; echo \"s=$MY_TOKEN\" >&2; touch /workspace/out.txt",
        );
        let (tx, mut rx) = mpsc::channel(64);
        let exec = PodmanExecutor::new(Limits::default());
        let result = exec
            .run(ExecRequest {
                job: &job,
                workspace: &ws,
                logs: tx,
                cancel: CancellationToken::new(),
            })
            .await;
        assert_eq!(result.outcome, Outcome::Success, "{result:?}");
        let mut all = Vec::new();
        while let Some(chunk) = rx.recv().await {
            all.extend_from_slice(&chunk);
        }
        let text = String::from_utf8_lossy(&all).into_owned();
        assert!(text.contains("hi-from-container"));
        assert!(ws.join("out.txt").exists());
    }
}