//! `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 wait;

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 RunnerCmd {

    /// Register this runner with a hub using a one-time token from its Admin → Runners page

    Register {

        /// The hub, e.g. https://earth.example:7343

        hub: String,

        /// The registration token (read from stdin when not given)

        #[arg(long)]

        token: Option<String>,

        /// The runner's name (default: the host name)

        #[arg(long)]

        name: Option<String>,

        /// The hub's CA certificate, when self-signed

        #[arg(long)]

        ca: Option<PathBuf>,

        /// A label the runner offers (repeatable)

        #[arg(long)]

        label: Vec<String>,

    },

}

#[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] (or the

    /// runner.toml `flow runner register` wrote)

    Runner {

        #[command(subcommand)]

        cmd: Option<RunnerCmd>,

    },

    /// Follow the checks of a pushed state on OxygenHub until they finish:

    /// exit 0 when all passed, 1 when one failed, 2 at the time limit.

    /// Uses $OXY_TOKEN and $OXY_CA as oxy does.

    Wait {

        /// The repository's URL on the hub (https://hub/owner/name)

        url: String,

        /// The state hash the push moved the bookmark to (`oxy push --json` prints it)

        state: String,

        /// Minutes to wait at most

        #[arg(long, default_value_t = 60)]

        timeout: u64,

    },

}

#[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 { cmd: Some(RunnerCmd::Register { hub, token, name, ca, label }) } => {

            let home = Home::locate(cli.home.as_deref())?;

            if std::env::var_os("OXY_HOME").is_none() {

                std::env::set_var("OXY_HOME", home.root.join("oxy"));

            }

            let name = name.unwrap_or_else(|| {

                std::fs::read_to_string("/etc/hostname").map(|h| h.trim().to_string()).unwrap_or_else(|_| "flow".into())

            });

            let token = match token {

                Some(t) => t,

                None => {

                    eprintln!("Paste the registration token from the hub's Admin → Runners, then press Enter:");

                    let mut line = String::new();

                    std::io::stdin().read_line(&mut line)?;

                    line.trim().to_string()

                }

            };

            let got = runner::register(&home, &hub, &token, &name, ca, label)?;

            println!("registered as {} (runner {}); start it with `flow runner`", got.name, got.id);

            Ok(())

        }

        Cmd::Runner { cmd: None } => {

            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)

        }

        Cmd::Wait { url, state, timeout } => {

            let token = std::env::var("OXY_TOKEN").ok();

            let ca = std::env::var_os("OXY_CA").map(PathBuf::from);

            let code =

                wait::wait(&url, &state, token.as_deref(), ca.as_deref(), Duration::from_secs(timeout * 60), Duration::from_secs(15))?;

            std::process::exit(code);

        }

    }

}

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),

    }

}