populate: deliver sources other than flakes with the Rust program

The program delivers derivation, file, git, pass, passage, pipe, and
symlink sources, makes backups, and prunes, as the shell implementation
does, with the same messages.  Commands on the target are fixed shell
snippets in src/sh.rs, which get values as positional parameters, run
directly for local targets and over ssh, sharing one master connection,
for remote ones.  rsync's transfers and its check for .rsync-filter files
it can't read are in src/rsync.rs.  Given flake or flake-input sources, or
sources drawing inputs from others, the program fails before changing
anything, saying that these aren't implemented yet.

SIGHUP, SIGINT, and SIGTERM make populate stop once the command it waits
for ends, stop the ssh master, remove its temporary directory, and exit
with 128 plus the signal's number.

The spec's target gets local, which rust.nix sets during evaluation, like
the other functions of krops decide, and which the program decides when
it runs if it's null.  The spec comes on file descriptor 3, so the
commands populate runs keep standard input, and the tools populate runs
come first in PATH.

1 files changed, 216 insertions, 0 deletions

diff --git a/pkgs/populate/src/populate.rs b/pkgs/populate/src/populate.rs
new file mode 100644
index 0000000..d3567d3
--- /dev/null
+++ b/pkgs/populate/src/populate.rs
@@ -0,0 +1,216 @@
+//! Delivering the sources of a spec.
+
+use crate::rsync::{self, Config};
+use crate::run::{self, Context, Failure, Result};
+use crate::sh;
+use crate::spec::{Source, Spec};
+use serde_json::{Map, Value};
+use std::fs;
+use std::path::PathBuf;
+use std::process::Command;
+
+/// Checks the sentinel, makes a backup if asked, delivers each source, and
+/// prunes if asked.
+pub fn populate(context: &Context, spec: &Spec) -> Result<()> {
+    let dir = &spec.target.path;
+    unsupported(context, spec)?;
+    let sentinel = format!("{dir}/.populate");
+    let force = if spec.force { "1" } else { "" };
+    context.run(
+        sh::CHECK,
+        &[&sentinel, &context.name(&sentinel), dir, force],
+    )?;
+    if spec.backup {
+        context.run(&sh::backup(), &[dir, &context.name(&sentinel)])?;
+    }
+    for (name, source) in &spec.source {
+        let path = format!("{dir}/{name}");
+        eprintln!("populating '{}'...", context.name(&path));
+        let workdir = context.tmpdir.join("workdirs").join(name);
+        fs::create_dir_all(&workdir).map_err(|e| {
+            Failure::Message(format!("error: can't create {}: {e}", workdir.display()))
+        })?;
+        deliver(context, source, &path, workdir)?;
+    }
+    if spec.prune {
+        let flakes = spec
+            .source
+            .iter()
+            .filter(|(_, source)| matches!(source.kind.as_str(), "flake" | "flake-input"))
+            .map(|(name, _)| name.as_str());
+        let mut args = vec![dir.as_str()];
+        let log = context.name(dir);
+        args.push(&log);
+        args.extend(spec.source.keys().map(String::as_str));
+        args.push("--");
+        args.extend(flakes);
+        context.run(&sh::prune(), &args)?;
+    }
+    Ok(())
+}
+
+/// Fails, before changing anything, if the spec needs what this program
+/// doesn't do yet.
+fn unsupported(context: &Context, spec: &Spec) -> Result<()> {
+    for (name, source) in &spec.source {
+        let copying = matches!(source.kind.as_str(), "flake" | "flake-input")
+            && source
+                .settings
+                .get("fetchOnTarget")
+                .and_then(Value::as_bool)
+                != Some(true);
+        if copying && context.target.sudo && !context.local {
+            return Err(Failure::Message(format!(
+                "krops: source {}/{name}: copying flakes to targets with sudo is not supported",
+                spec.target.path
+            )));
+        }
+    }
+    for source in spec.source.values() {
+        let what = match source.kind.as_str() {
+            "derivation" | "file" | "git" | "pass" | "passage" | "pipe" | "symlink" => None,
+            kind => Some(format!("{kind} sources")),
+        };
+        let what = what.or_else(|| {
+            (!source.follows().is_empty()).then(|| "drawing inputs from sources".to_string())
+        });
+        if let Some(what) = what {
+            return Err(Failure::Message(format!(
+                "populate: {what} aren't implemented yet, so it delivers nothing to {}",
+                context.name(&spec.target.path)
+            )));
+        }
+    }
+    Ok(())
+}
+
+fn string<'a>(settings: &'a Map<String, Value>, key: &str) -> Result<&'a str> {
+    settings
+        .get(key)
+        .and_then(Value::as_str)
+        .ok_or_else(|| Failure::Message(format!("populate: the source has no {key}")))
+}
+
+/// Delivers a source to path on the target, using workdir for files it
+/// prepares on the deploying machine.
+fn deliver(context: &Context, source: &Source, path: &str, workdir: PathBuf) -> Result<()> {
+    let settings = &source.settings;
+    match source.kind.as_str() {
+        "derivation" => context.run(
+            &sh::derivation(),
+            &[path, &context.name(path), string(settings, "text")?],
+        ),
+        "file" => {
+            let source_path = string(settings, "path")?;
+            // Store paths have the same time, so only checksums tell
+            // different contents apart.
+            let use_checksum = settings
+                .get("useChecksum")
+                .and_then(Value::as_bool)
+                .unwrap_or(run::is_store_path(source_path));
+            let config = Config {
+                use_checksum,
+                ..Config::from_settings(settings)
+            };
+            rsync::transfer(context, &config, source_path, path)
+        }
+        "pipe" => {
+            let out = workdir.join("out");
+            let file = fs::File::create(&out).map_err(|e| {
+                Failure::Message(format!("error: can't create {}: {e}", out.display()))
+            })?;
+            run::status(
+                Command::new(string(settings, "command")?)
+                    .current_dir(&workdir)
+                    .stdout(file),
+            )?;
+            run::status(
+                Command::new("touch")
+                    .arg("-t")
+                    .arg("300001010000")
+                    .arg(&out),
+            )?;
+            // With the fixed modification time, only checksums tell outputs
+            // of the same size apart.
+            let config = Config {
+                use_checksum: true,
+                ..Config::default()
+            };
+            rsync::transfer(context, &config, &out.to_string_lossy(), path)
+        }
+        "git" => {
+            let flag = |key| match settings.get(key).and_then(Value::as_bool) {
+                Some(true) => "1",
+                _ => "",
+            };
+            let excludes = settings
+                .get("clean")
+                .and_then(|clean| clean.get("exclude"))
+                .and_then(Value::as_array)
+                .into_iter()
+                .flatten()
+                .filter_map(Value::as_str);
+            let name = context.name(path);
+            let mut args = vec![
+                path,
+                &name,
+                string(settings, "url")?,
+                string(settings, "ref")?,
+                flag("shallow"),
+                flag("fetchAlways"),
+                flag("reclone"),
+            ];
+            args.extend(excludes);
+            context.run(&sh::git(), &args)
+        }
+        "pass" => {
+            let dir = string(settings, "dir")?;
+            let name = string(settings, "name")?;
+            let info = run::output(&mut run::local_shell(
+                sh::local::PASS_INFO,
+                &[dir, name],
+                &workdir,
+            ))?;
+            let info = info.trim_end_matches('\n');
+            if !info.is_empty() {
+                let delivered = context.output(sh::PASS_INFO, &[path])?;
+                if delivered.trim_end_matches('\n') == info {
+                    return Ok(());
+                }
+            }
+            run::status(&mut run::local_shell(
+                sh::local::PASS,
+                &[dir, name, info],
+                &workdir,
+            ))?;
+            rsync::transfer(
+                context,
+                &Config::default(),
+                &workdir.to_string_lossy(),
+                path,
+            )
+        }
+        "passage" => {
+            let host = context.target.host.split('.').next().unwrap_or_default();
+            let mut command = run::local_shell(sh::local::PASSAGE, &[host], &workdir);
+            for (variable, key) in [
+                ("PASSAGE_AGE", "age"),
+                ("PASSAGE_DIR", "dir"),
+                ("PASSAGE_IDENTITIES_FILE", "identities_file"),
+            ] {
+                command.env(variable, string(settings, key)?);
+            }
+            run::status(&mut command)?;
+            rsync::transfer(
+                context,
+                &Config::default(),
+                &workdir.to_string_lossy(),
+                path,
+            )
+        }
+        "symlink" => context.run(sh::SYMLINK, &[path, string(settings, "target")?]),
+        kind => Err(Failure::Message(format!(
+            "populate: {kind} sources aren't implemented yet"
+        ))),
+    }
+}