//! `flow`: check pipelines, run them locally, and run jobs for OxygenHub
//! (docs/spec.md, docs/m0.md).

mod cpio;
mod home;
mod microvm;
mod runner;
mod workspace;

use anyhow::{bail, Context, Result};
use clap::{Parser, Subcommand};
use flow_core::api::Record;
use flow_core::{Pipeline, Runtime, PIPELINE_PATH};
use home::Home;
use std::collections::BTreeSet;
use std::path::{Path, PathBuf};
use std::time::Duration;

#[derive(Parser)]
#[command(name = "flow", version, about = "OxygenFlow: hermetic pipelines for Oxygen repositories")]
struct Cli {
    /// The Flow home (default $FLOW_HOME, else ~/.local/share/flow).
    #[arg(long, global = true)]
    home: Option<PathBuf>,
    #[command(subcommand)]
    cmd: Cmd,
}

#[derive(Subcommand)]
enum Cmd {
    /// Validate a pipeline and print its jobs and hash
    Check {
        /// The pipeline file (default: .flow/pipeline.kdl)
        file: Option<PathBuf>,
    },
    /// Print a file's BLAKE3, for pinning in flow.toml
    Hash { file: PathBuf },
    /// Rootfs images for sandboxes
    Rootfs {
        #[command(subcommand)]
        cmd: RootfsCmd,
    },
    /// Run the pipeline on this directory's files, locally
    Run {
        /// Only these jobs (and what they depend on)
        jobs: Vec<String>,
        /// The directory to build (default: the current one)
        #[arg(long, default_value = ".")]
        dir: PathBuf,
        /// Minutes before a job is stopped
        #[arg(long, default_value_t = 60)]
        timeout: u64,
    },
    /// Claim and run jobs for the hub in flow.toml's [runner]
    Runner,
}

#[derive(Subcommand)]
enum RootfsCmd {
    /// Make an ext4 rootfs image from a root file system tarball and store it by hash
    Import {
        tarball: PathBuf,
        /// Refuse the tarball unless its SHA-256 matches (as its publisher states it)
        #[arg(long)]
        sha256: Option<String>,
    },
    /// List the images this home holds
    List,
}

fn main() {
    let cli = Cli::parse();
    if let Err(e) = run(cli) {
        eprintln!("flow: {e:#}");
        std::process::exit(1);
    }
}

fn run(cli: Cli) -> Result<()> {
    match cli.cmd {
        Cmd::Check { file } => check(&file.unwrap_or_else(|| PathBuf::from(PIPELINE_PATH))),
        Cmd::Hash { file } => {
            println!("{}", home::hash_file(&file)?);
            Ok(())
        }
        Cmd::Rootfs { cmd } => {
            let home = Home::locate(cli.home.as_deref())?;
            match cmd {
                RootfsCmd::Import { tarball, sha256 } => rootfs_import(&home, &tarball, sha256.as_deref()),
                RootfsCmd::List => {
                    for e in std::fs::read_dir(home.root.join("cas"))?.flatten() {
                        let size = e.metadata().map(|m| m.len()).unwrap_or(0);
                        if let Some(h) = e.file_name().to_string_lossy().strip_suffix(".ext4") {
                            println!("cas://blob/{h}  {} MiB", size >> 20);
                        }
                    }
                    Ok(())
                }
            }
        }
        Cmd::Run { jobs, dir, timeout } => {
            let home = Home::locate(cli.home.as_deref())?;
            run_local(&home, &dir, &jobs, Duration::from_secs(timeout * 60))
        }
        Cmd::Runner => {
            let home = Home::locate(cli.home.as_deref())?;
            // The runner's oxy state (identity, user config) stays in its home.
            if std::env::var_os("OXY_HOME").is_none() {
                std::env::set_var("OXY_HOME", home.root.join("oxy"));
            }
            runner::serve(&home)
        }
    }
}

fn load(file: &Path) -> Result<Pipeline> {
    let src = std::fs::read_to_string(file).with_context(|| format!("read {}", file.display()))?;
    Pipeline::parse(&src).map_err(|p| anyhow::anyhow!("{} is not valid:\n{p}", file.display()))
}

fn check(file: &Path) -> Result<()> {
    let p = load(file)?;
    let depths = p.depths();
    for j in p.order() {
        let sb = &p.sandboxes[&j.sandbox];
        let after = if j.depends_on.is_empty() { String::new() } else { format!(" after {}", j.depends_on.join(", ")) };
        println!("{}{}  {} [{}]{after}", "  ".repeat(depths[j.name.as_str()]), j.name, j.exec.join(" "), sb.runtime.as_str());
    }
    println!("pipeline {}", hex::encode(p.hash()));
    Ok(())
}

fn rootfs_import(home: &Home, tarball: &Path, sha256: Option<&str>) -> Result<()> {
    if let Some(want) = sha256 {
        let out = std::process::Command::new("sha256sum").arg(tarball).output().context("run sha256sum")?;
        let got = String::from_utf8_lossy(&out.stdout).split_whitespace().next().unwrap_or("").to_string();
        if got != want.to_lowercase() {
            bail!("{} has SHA-256 {got}, expected {want}", tarball.display());
        }
    }
    let tmp = home.root.join("work").join(format!("rootfs-{}", std::process::id()));
    let _ = std::fs::remove_dir_all(&tmp);
    let tree = tmp.join("tree");
    std::fs::create_dir_all(&tree)?;
    let result = (|| -> Result<String> {
        let status = std::process::Command::new("tar").arg("-xf").arg(tarball).arg("-C").arg(&tree).arg("--no-same-owner").status()?;
        if !status.success() {
            bail!("tar could not unpack {}", tarball.display());
        }
        // Mount points flow-init needs (docs/m0.md §2).
        for d in ["src", "tmp", "proc", "dev", "run"] {
            std::fs::create_dir_all(tree.join(d))?;
        }
        let used: u64 = ignore::WalkBuilder::new(&tree)
            .hidden(false)
            .ignore(false)
            .git_ignore(false)
            .build()
            .flatten()
            .filter_map(|e| e.metadata().ok())
            .map(|m| m.len().max(4096))
            .sum();
        let mib = used * 13 / 10 / (1 << 20) + 64;
        let img = tmp.join("rootfs.ext4");
        let status = std::process::Command::new("mkfs.ext4")
            .args(["-q", "-F", "-L", "rootfs", "-E", "root_owner=0:0", "-d"])
            .arg(&tree)
            .arg(&img)
            .arg(format!("{mib}M"))
            .status()
            .context("run mkfs.ext4")?;
        if !status.success() {
            bail!("mkfs.ext4 failed");
        }
        let hash = home::hash_file(&img)?;
        std::fs::rename(&img, home.rootfs(&hash))?;
        Ok(hash)
    })();
    let _ = std::fs::remove_dir_all(&tmp);
    let hash = result?;
    println!("rootfs \"cas://blob/{hash}\"");
    Ok(())
}

fn run_local(home: &Home, dir: &Path, only: &[String], timeout: Duration) -> Result<()> {
    let p = load(&dir.join(PIPELINE_PATH))?;
    // The named jobs and everything they depend on.
    let mut wanted: BTreeSet<String> = only.iter().cloned().collect();
    for name in only {
        if !p.jobs.contains_key(name) {
            bail!("no job {name:?} in the pipeline");
        }
    }
    if !only.is_empty() {
        loop {
            let more: Vec<String> = wanted.iter().flat_map(|j| p.jobs[j].depends_on.clone()).filter(|d| !wanted.contains(d)).collect();
            if more.is_empty() {
                break;
            }
            wanted.extend(more);
        }
    }
    let scratch = home.root.join("work").join(format!("local-{}", std::process::id()));
    let _ = std::fs::remove_dir_all(&scratch);
    std::fs::create_dir_all(scratch.join("tree"))?;
    workspace::copy_tree(dir, &scratch.join("tree"))?;
    let mut failed: BTreeSet<String> = BTreeSet::new();
    let mut worst = 0;
    for job in p.order() {
        if !only.is_empty() && !wanted.contains(&job.name) {
            continue;
        }
        if let Some(d) = job.depends_on.iter().find(|d| failed.contains(*d)) {
            println!("── {} skipped: {d} failed", job.name);
            failed.insert(job.name.clone());
            continue;
        }
        let sb = &p.sandboxes[&job.sandbox];
        if sb.runtime != Runtime::Microvm {
            bail!("job {}: runtime {} is not available in this build yet", job.name, sb.runtime.as_str());
        }
        println!("── {} ({})", job.name, job.exec.join(" "));
        let js = scratch.join(&job.name);
        std::fs::create_dir_all(&js)?;
        // Each job starts from the same files: jobs do not share writes.
        workspace::image(&scratch.join("tree"), &js.join("workspace.ext4"))?;
        let out = microvm::run(home, sb, job, &js.join("workspace.ext4"), &js, timeout, &mut |r: Record| print_record(&r))?;
        let _ = std::fs::remove_dir_all(&js);
        match out.exit_code {
            Some(0) => println!("── {} passed in {} ms", job.name, out.duration_ms),
            other => {
                println!(
                    "── {} failed ({}) in {} ms",
                    job.name,
                    other.map(|c| format!("exit {c}")).unwrap_or_else(|| "no exit code".into()),
                    out.duration_ms
                );
                failed.insert(job.name.clone());
                worst = 1;
            }
        }
    }
    let _ = std::fs::remove_dir_all(&scratch);
    if worst != 0 {
        std::process::exit(worst);
    }
    Ok(())
}

fn print_record(r: &Record) {
    match r.stream.as_str() {
        "out" => println!("   {}", r.text),
        "err" => eprintln!("   {}", r.text),
        s => eprintln!("   [{s}] {}", r.text),
    }
}