mecha10-cli 4.2.15

//! `mecha10 build-worker` handler (LAB-5440/5442/5443/5450): the server-side half of "build via
//! mecha10 dashboard", run *inside* the Kestra-dispatched build container
//! (`mecha10-deploy`'s `docker/build-worker.Dockerfile`, `ENTRYPOINT ["mecha10"]`) rather than
//! on a developer's own machine. Everything else named `build*` in this crate
//! (`handlers::build`, `handlers::build_remote`) is client-side: either a local `cargo build`
//! or a remote-build *trigger* dispatched from a user's machine. This is the "I am the
//! dispatched worker, now do the actual work" mode nothing in the CLI had before.
//!
//! Pipeline, driven entirely by [`BuildWorkerArgs`] and [`WorkerEnv`]:
//! 1. If the working directory is empty, fetch `--source-url` (zip-upload path) and extract
//!    it in place. If it's already non-empty, `run-container-job.yml` has already cloned a
//!    connected GitHub repo there (LAB-5450) - nothing to fetch.
//! 2. Strip any `--excluded-dirs`.
//! 3. Cross-compile for `--target-platform`/`--target-arch` via `handlers::build`'s existing
//!    `build_with_cargo` - no cross-compile logic is duplicated here.
//! 4. Package and upload the resulting binary via `handlers::upload`'s existing tarball/upload
//!    client - same reuse rule.
//! 5. Report the terminal outcome back to mecha10-api's
//!    `POST /api/robots/{robot_id}/build-jobs/{job_id}/status`, authenticated as the
//!    control-plane service account (not a user), so the build-job row this container was
//!    dispatched for always ends up `succeeded`/`failed`, never stuck `running` just because
//!    the container exited.
//!
//! See `commands::build_worker`'s module doc comment for the exact CLI-arg half of the
//! contract a `mecha10-api` dispatch must match; [`WorkerEnv`] documents the env var half.

use crate::commands::{BuildTargetArgs, BuildWorkerArgs};
use crate::context::CliContext;
use crate::handlers::build::{build_with_cargo, BuildTarget};
use crate::handlers::upload::{create_tarball, detect_arch, upload_to_registry, UploadParams, BINARY_REGISTRY_PATH};
use crate::paths;
use crate::services::build_job_client::{self, BuildJobState, ReportBuildJobStatusParams};
use crate::services::build_source;
use crate::ui::style;
use anyhow::{Context, Result};
use tempfile::tempdir;

/// Env vars `run-container-job.yml` injects into every dispatched build-worker container
/// automatically (LAB-5440) - deliberately not CLI args, so this worker reads the same
/// vocabulary every other job dispatched through that template already uses:
/// - `MECHA10_API_KEY`: the control-plane service token (mecha10-api's
///   `AppState::control_plane_service_key`, the Kestra `token` input), used as this worker's
///   own bearer auth when it reports its status and uploads the built binary - *not* a user's
///   personal API key.
/// - `CONTROL_PLANE_URL`: mecha10-api's base URL (the Kestra `api_url` input).
/// - `ROBOT_ID` / `JOB_ID`: which robot and build-job row this dispatch is for.
#[derive(Debug)]
pub struct WorkerEnv {
    pub api_key: String,
    pub control_plane_url: String,
    pub robot_id: String,
    pub job_id: String,
}

impl WorkerEnv {
    /// Read from the real process environment - the only caller in production code.
    pub fn from_process_env() -> Result<Self> {
        Self::from_lookup(|name| std::env::var(name).ok())
    }

    /// Read via an injectable lookup function, so tests can exercise missing/present env vars
    /// without mutating real process-global state (which would race with every other test in
    /// the process, same reason `paths.rs`'s own env var tests need a mutex).
    fn from_lookup(lookup: impl Fn(&str) -> Option<String>) -> Result<Self> {
        let require =
            |name: &str| -> Result<String> { lookup(name).with_context(|| format!("Missing required env var {name}")) };

        Ok(Self {
            api_key: require("MECHA10_API_KEY")?,
            control_plane_url: require("CONTROL_PLANE_URL")?,
            robot_id: require("ROBOT_ID")?,
            job_id: require("JOB_ID")?,
        })
    }
}

/// Entry point for `mecha10 build-worker`. Always reports a terminal status back to
/// mecha10-api before returning - on success with the uploaded binary's id, on *any* failure
/// (source fetch, build, upload) with an error message - so the dispatched build-job row never
/// gets stuck `running` just because this process exited. The one exception is
/// [`WorkerEnv::from_process_env`] itself failing: with no `ROBOT_ID`/`JOB_ID` there is no job
/// row to address at all, so that case can only propagate the error and let the dispatcher's
/// own timeout handle it.
pub async fn handle_build_worker(ctx: &mut CliContext, args: &BuildWorkerArgs) -> Result<()> {
    let env = WorkerEnv::from_process_env()?;

    match build_and_upload(ctx, args, &env).await {
        Ok(binary_id) => {
            report_status(&env, BuildJobState::Succeeded, Some(&binary_id), None).await?;
            println!("Build worker succeeded. binary_id={binary_id}");
            Ok(())
        }
        Err(err) => {
            let error_message = format!("{err:#}");
            if let Err(report_err) = report_status(&env, BuildJobState::Failed, None, Some(&error_message)).await {
                eprintln!("Warning: failed to report build-job failure status: {report_err:#}");
            }
            Err(err)
        }
    }
}

/// The actual fetch -> build -> upload pipeline, split out of [`handle_build_worker`] so the
/// success/failure status-report wrapping above isn't tangled into it.
async fn build_and_upload(ctx: &mut CliContext, args: &BuildWorkerArgs, env: &WorkerEnv) -> Result<String> {
    ensure_source_present(ctx, args).await?;

    if let Some(excluded_dirs) = args.excluded_dirs.as_deref() {
        remove_excluded_dirs(&ctx.working_dir, excluded_dirs)?;
    }

    let triple = resolve_target_triple(&args.target_platform, &args.target_arch)?;

    let build_args = BuildTargetArgs {
        target: Some(triple.clone()),
        bump: None,
        version: None,
        dev: false,
        non_interactive: true,
        location: None,
        source: None,
    };
    build_with_cargo(ctx, BuildTarget::Robot, &build_args)
        .await
        .context("Build failed")?;

    upload_built_binary(ctx, &triple, env).await
}

/// Fetch and extract `--source-url` into the working directory, unless it's already non-empty
/// (the GitHub path - `run-container-job.yml` cloned the repo there before `command` ran).
async fn ensure_source_present(ctx: &CliContext, args: &BuildWorkerArgs) -> Result<()> {
    if !dir_is_empty(&ctx.working_dir)? {
        style::step(format!(
            "Working directory {} is already populated (GitHub source) - skipping source fetch",
            ctx.working_dir.display()
        ));
        return Ok(());
    }

    let source_url = args
        .source_url
        .as_deref()
        .context("Working directory is empty and no --source-url was given")?;
    let source_token = args
        .source_token
        .as_deref()
        .context("Working directory is empty and no --source-token was given")?;

    fetch_and_extract_source(source_url, source_token, &ctx.working_dir).await
}

/// `true` if `dir` has no entries at all (doesn't need to exist - a missing directory counts
/// as empty, since there's nothing there to have already been cloned into).
fn dir_is_empty(dir: &std::path::Path) -> Result<bool> {
    match std::fs::read_dir(dir) {
        Ok(mut entries) => Ok(entries.next().is_none()),
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(true),
        Err(e) => Err(e).with_context(|| format!("Failed to read directory {}", dir.display())),
    }
}

/// `GET source_url` with `Authorization: Bearer source_token`, write the response to a temp
/// file, and extract it into `dest_dir` via `build_source::extract_project_zip`.
async fn fetch_and_extract_source(source_url: &str, source_token: &str, dest_dir: &std::path::Path) -> Result<()> {
    style::step(format!("Fetching project source from {source_url}..."));

    let client = reqwest::Client::new();
    let response = client
        .get(source_url)
        .bearer_auth(source_token)
        .send()
        .await
        .context("Failed to send source-fetch request")?;

    let status = response.status();
    if !status.is_success() {
        let body = response.text().await.unwrap_or_default();
        anyhow::bail!("Failed to fetch project source (status {status}): {body}");
    }

    let bytes = response
        .bytes()
        .await
        .context("Failed to read source-fetch response body")?;

    let temp_dir = tempdir().context("Failed to create a temp directory for the fetched source zip")?;
    let zip_path = temp_dir.path().join("source.zip");
    std::fs::write(&zip_path, &bytes).with_context(|| format!("Failed to write {}", zip_path.display()))?;

    std::fs::create_dir_all(dest_dir).with_context(|| format!("Failed to create {}", dest_dir.display()))?;
    build_source::extract_project_zip(&zip_path, dest_dir).context("Failed to extract project source")?;

    style::step(format!("Project source extracted into {}", dest_dir.display()));
    Ok(())
}

/// Remove each comma-separated relative directory under `root`, ignoring entries that don't
/// exist (mecha10-api's `build_source_excluded_dirs` build-job field; blank entries from stray
/// commas/whitespace are skipped).
fn remove_excluded_dirs(root: &std::path::Path, excluded_dirs: &str) -> Result<()> {
    for raw in excluded_dirs.split(',') {
        let relative = raw.trim();
        if relative.is_empty() {
            continue;
        }

        let path = root.join(relative);
        if !path.exists() {
            continue;
        }

        if path.is_dir() {
            std::fs::remove_dir_all(&path).with_context(|| format!("Failed to remove directory {}", path.display()))?;
        } else {
            std::fs::remove_file(&path).with_context(|| format!("Failed to remove file {}", path.display()))?;
        }
        style::step(format!("Removed excluded path: {relative}"));
    }

    Ok(())
}

/// Resolve a Rust target triple from mecha10-api's `target_platform`/`target_arch` build-job
/// fields - the inverse of `build_remote::platform_arch_for_target`, which derives those same
/// two fields from a triple on the client side.
fn resolve_target_triple(target_platform: &str, target_arch: &str) -> Result<String> {
    let triple = match (target_platform, target_arch) {
        ("linux", "aarch64") => "aarch64-unknown-linux-gnu",
        ("linux", "x86_64") => "x86_64-unknown-linux-gnu",
        ("macos", "aarch64") => "aarch64-apple-darwin",
        ("macos", "x86_64") => "x86_64-apple-darwin",
        _ => anyhow::bail!("Unsupported target_platform/target_arch combination: {target_platform}/{target_arch}"),
    };
    Ok(triple.to_string())
}

/// Package the just-built binary and upload it via `handlers::upload`'s existing client
/// (`create_tarball`/`upload_to_registry`), authenticating as the control-plane service
/// account instead of a logged-in user. Returns the registry's assigned binary id.
async fn upload_built_binary(ctx: &mut CliContext, triple: &str, env: &WorkerEnv) -> Result<String> {
    let project_name = ctx.project()?.name()?;

    let binary_path = ctx
        .working_dir
        .join(paths::target_path_with_triple(Some(triple), "release"))
        .join(&project_name);
    if !binary_path.exists() {
        anyhow::bail!("Build succeeded but binary not found at {}", binary_path.display());
    }

    let config_path = ctx.working_dir.join(paths::PROJECT_CONFIG);
    let dist_config_content = tokio::fs::read_to_string(&config_path)
        .await
        .with_context(|| format!("Failed to read {}", config_path.display()))?;
    let dist_config: serde_json::Value =
        serde_json::from_str(&dist_config_content).context("Failed to parse mecha10.json")?;

    let config_name = dist_config["name"].as_str().unwrap_or(&project_name);
    let version = dist_config["version"].as_str().unwrap_or("0.1.0").to_string();
    let arch = detect_arch(&binary_path);
    let platform = "linux";

    let temp_dir = tempdir().context("Failed to create a temp directory for the upload tarball")?;
    let tarball_name = format!("{config_name}-v{version}-{platform}-{arch}.tar.gz");
    let tarball_path = temp_dir.path().join(&tarball_name);
    let binary_filename = binary_path
        .file_name()
        .and_then(|n| n.to_str())
        .unwrap_or(&project_name);

    create_tarball(&binary_path, binary_filename, &tarball_path)?;

    let registry_url = format!(
        "{}{}",
        env.control_plane_url.trim_end_matches('/'),
        BINARY_REGISTRY_PATH
    );

    style::step(format!("Uploading built binary to {registry_url}..."));
    let response = upload_to_registry(UploadParams {
        registry_url: &registry_url,
        name: config_name,
        version: &version,
        platform,
        arch: &arch,
        target: BuildTarget::Robot.as_str(),
        robot_id: &env.robot_id,
        created_by: None,
        api_key: Some(&env.api_key),
        tarball_path: &tarball_path,
        tarball_name: &tarball_name,
    })
    .await
    .context("Failed to upload built binary")?;

    response
        .get("id")
        .and_then(|v| v.as_str())
        .map(str::to_string)
        .ok_or_else(|| anyhow::anyhow!("Upload response missing binary id: {response}"))
}

/// Report the worker's terminal outcome to mecha10-api, authenticated with the control-plane
/// service token (not a user's own API key) - see [`WorkerEnv`].
async fn report_status(
    env: &WorkerEnv,
    state: BuildJobState,
    binary_id: Option<&str>,
    error_message: Option<&str>,
) -> Result<()> {
    build_job_client::report_build_job_status(ReportBuildJobStatusParams {
        control_plane_url: &env.control_plane_url,
        robot_id: &env.robot_id,
        job_id: &env.job_id,
        service_token: &env.api_key,
        state,
        binary_id,
        error_message,
    })
    .await
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::context::CliContextBuilder;
    use std::collections::HashMap;
    use std::io::{Read, Write};
    use tempfile::tempdir;

    fn env_map(pairs: &[(&str, &str)]) -> HashMap<String, String> {
        pairs.iter().map(|(k, v)| (k.to_string(), v.to_string())).collect()
    }

    #[test]
    fn worker_env_reads_all_four_vars() {
        let map = env_map(&[
            ("MECHA10_API_KEY", "service-token"),
            ("CONTROL_PLANE_URL", "https://api.mecha10.dev"),
            ("ROBOT_ID", "robot-1"),
            ("JOB_ID", "job-1"),
        ]);

        let env = WorkerEnv::from_lookup(|name| map.get(name).cloned()).unwrap();

        assert_eq!(env.api_key, "service-token");
        assert_eq!(env.control_plane_url, "https://api.mecha10.dev");
        assert_eq!(env.robot_id, "robot-1");
        assert_eq!(env.job_id, "job-1");
    }

    #[test]
    fn worker_env_errors_when_a_var_is_missing() {
        let map = env_map(&[
            ("MECHA10_API_KEY", "service-token"),
            ("CONTROL_PLANE_URL", "https://api.mecha10.dev"),
            ("ROBOT_ID", "robot-1"),
            // JOB_ID deliberately missing
        ]);

        let err = WorkerEnv::from_lookup(|name| map.get(name).cloned()).unwrap_err();
        assert!(err.to_string().contains("JOB_ID"));
    }

    #[test]
    fn dir_is_empty_true_for_a_missing_directory() {
        let dir = tempdir().unwrap();
        assert!(dir_is_empty(&dir.path().join("does-not-exist")).unwrap());
    }

    #[test]
    fn dir_is_empty_true_for_an_actually_empty_directory() {
        let dir = tempdir().unwrap();
        assert!(dir_is_empty(dir.path()).unwrap());
    }

    #[test]
    fn dir_is_empty_false_once_something_has_been_cloned_into_it() {
        let dir = tempdir().unwrap();
        std::fs::write(dir.path().join("README.md"), "hi").unwrap();
        assert!(!dir_is_empty(dir.path()).unwrap());
    }

    #[test]
    fn resolve_target_triple_maps_known_platform_arch_pairs() {
        assert_eq!(
            resolve_target_triple("linux", "aarch64").unwrap(),
            "aarch64-unknown-linux-gnu"
        );
        assert_eq!(
            resolve_target_triple("linux", "x86_64").unwrap(),
            "x86_64-unknown-linux-gnu"
        );
        assert_eq!(
            resolve_target_triple("macos", "aarch64").unwrap(),
            "aarch64-apple-darwin"
        );
    }

    #[test]
    fn resolve_target_triple_errors_on_an_unknown_combination() {
        let err = resolve_target_triple("windows", "arm").unwrap_err();
        assert!(err.to_string().contains("Unsupported"));
    }

    #[test]
    fn remove_excluded_dirs_removes_listed_directories_and_ignores_missing_ones() {
        let dir = tempdir().unwrap();
        std::fs::create_dir_all(dir.path().join("models")).unwrap();
        std::fs::write(dir.path().join("models/weights.bin"), b"x").unwrap();
        std::fs::create_dir_all(dir.path().join("src")).unwrap();

        remove_excluded_dirs(dir.path(), "models, does-not-exist").unwrap();

        assert!(!dir.path().join("models").exists());
        assert!(dir.path().join("src").exists());
    }

    #[test]
    fn remove_excluded_dirs_skips_blank_entries() {
        let dir = tempdir().unwrap();
        remove_excluded_dirs(dir.path(), ",,  ,").unwrap();
    }

    /// Minimal fake HTTP server for the source-fetch GET - same single-shot-listener pattern
    /// as `services/build_job_client.rs`'s own test helper.
    fn start_fake_source_server(status_line: &'static str, body: Vec<u8>) -> String {
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();

        std::thread::spawn(move || {
            if let Ok(mut stream) = listener.accept().map(|(s, _)| s) {
                let mut buf = [0u8; 4096];
                let _ = stream.read(&mut buf);
                let header = format!(
                    "{status_line}\r\nContent-Type: application/zip\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
                    body.len()
                );
                let _ = stream.write_all(header.as_bytes());
                let _ = stream.write_all(&body);
                let _ = stream.flush();
            }
        });

        format!("http://{}", addr)
    }

    /// A minimal valid zip with a single `hello.txt` entry, for exercising the
    /// fetch-and-extract path end to end.
    fn build_fake_source_zip() -> Vec<u8> {
        let mut bytes = Vec::new();
        {
            let mut zip = zip::ZipWriter::new(std::io::Cursor::new(&mut bytes));
            let options = zip::write::SimpleFileOptions::default();
            zip.start_file("hello.txt", options).unwrap();
            zip.write_all(b"hello world").unwrap();
            zip.finish().unwrap();
        }
        bytes
    }

    #[tokio::test]
    async fn fetch_and_extract_source_writes_the_fetched_zips_contents() {
        let url = start_fake_source_server("HTTP/1.1 200 OK", build_fake_source_zip());
        let dest_dir = tempdir().unwrap();

        fetch_and_extract_source(&url, "source-token", dest_dir.path())
            .await
            .unwrap();

        assert_eq!(
            std::fs::read_to_string(dest_dir.path().join("hello.txt")).unwrap(),
            "hello world"
        );
    }

    #[tokio::test]
    async fn fetch_and_extract_source_surfaces_the_servers_error_on_failure() {
        let url = start_fake_source_server("HTTP/1.1 403 Forbidden", b"source token expired".to_vec());
        let dest_dir = tempdir().unwrap();

        let err = fetch_and_extract_source(&url, "bad-token", dest_dir.path())
            .await
            .unwrap_err();

        assert!(err.to_string().contains("source token expired"));
    }

    #[tokio::test]
    async fn ensure_source_present_skips_fetch_when_github_already_cloned_the_repo() {
        let dir = tempdir().unwrap();
        std::fs::write(dir.path().join("Cargo.toml"), "[package]\nname=\"x\"\n").unwrap();

        let ctx = CliContextBuilder::new()
            .working_dir(dir.path().to_path_buf())
            .build()
            .unwrap();

        let args = BuildWorkerArgs {
            target_platform: "linux".to_string(),
            target_arch: "aarch64".to_string(),
            source_url: None,
            source_token: None,
            excluded_dirs: None,
        };

        // Would fail trying to fetch (no --source-url) if this didn't skip the fetch.
        ensure_source_present(&ctx, &args).await.unwrap();
    }

    #[tokio::test]
    async fn ensure_source_present_errors_when_empty_and_no_source_url_given() {
        let dir = tempdir().unwrap();
        let ctx = CliContextBuilder::new()
            .working_dir(dir.path().to_path_buf())
            .build()
            .unwrap();

        let args = BuildWorkerArgs {
            target_platform: "linux".to_string(),
            target_arch: "aarch64".to_string(),
            source_url: None,
            source_token: None,
            excluded_dirs: None,
        };

        let err = ensure_source_present(&ctx, &args).await.unwrap_err();
        assert!(err.to_string().contains("--source-url"));
    }

    #[tokio::test]
    async fn ensure_source_present_fetches_when_working_dir_is_empty() {
        let source_dir = tempdir().unwrap();
        let url = start_fake_source_server("HTTP/1.1 200 OK", build_fake_source_zip());

        let ctx = CliContextBuilder::new()
            .working_dir(source_dir.path().to_path_buf())
            .build()
            .unwrap();

        let args = BuildWorkerArgs {
            target_platform: "linux".to_string(),
            target_arch: "aarch64".to_string(),
            source_url: Some(url),
            source_token: Some("source-token".to_string()),
            excluded_dirs: None,
        };

        ensure_source_present(&ctx, &args).await.unwrap();

        assert_eq!(
            std::fs::read_to_string(source_dir.path().join("hello.txt")).unwrap(),
            "hello world"
        );
    }

    #[tokio::test]
    async fn handle_build_worker_errors_without_panicking_when_env_is_missing() {
        // No env vars set via `from_process_env` in this process are guaranteed - this test
        // only needs to confirm a missing-env failure surfaces as a normal `Err`, not a panic,
        // without actually touching real process env vars (which would race other tests).
        let err = WorkerEnv::from_lookup(|_| None).unwrap_err();
        assert!(err.to_string().contains("Missing required env var"));
    }
}