refactor: give every command we run a type of its own

src/cmd/argv.rs

@@ -0,0 +1,206 @@
+use std::{
+    fmt::Display,
+    fs::File,
+    os::unix::process::CommandExt,
+    path::Path,
+    process::{Command, ExitStatus, Output, Stdio},
+};
+
+use anyhow::{Context, Result, anyhow};
+
+use crate::ctx::Ctx;
+
+// a program and its arguments: the only thing in ahab that knows what argv looks like
+#[derive(Default, Clone)]
+pub struct Argv(Vec<String>);
+
+impl Display for Argv {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str(&self.quoted())
+    }
+}
+
+impl Argv {
+    pub fn new(program: &str) -> Self {
+        Self(vec![program.to_string()])
+    }
+
+    pub fn arg(mut self, arg: impl AsRef<str>) -> Self {
+        self.0.push(arg.as_ref().to_string());
+        self
+    }
+
+    pub fn args<I, S>(mut self, args: I) -> Self
+    where
+        I: IntoIterator<Item = S>,
+        S: AsRef<str>,
+    {
+        self.0
+            .extend(args.into_iter().map(|arg| arg.as_ref().to_string()));
+        self
+    }
+
+    // a long option and the value it takes
+    pub fn flag(self, name: &str, value: impl AsRef<str>) -> Self {
+        self.arg(name).arg(value)
+    }
+
+    pub fn program(&self) -> &str {
+        self.0.first().map(String::as_str).unwrap_or_default()
+    }
+
+    pub fn words(&self) -> &[String] {
+        &self.0
+    }
+
+    // one word list for `sh -c`, quoted so a value with a space survives the shell
+    pub fn quoted(&self) -> String {
+        // a nul byte is the only thing shlex refuses, and no argv can hold one
+        shlex::try_join(self.0.iter().map(String::as_str)).unwrap_or_default()
+    }
+
+    pub fn run(&self, ctx: &Ctx) -> Result<()> {
+        if self.skipped(ctx) {
+            return Ok(());
+        }
+
+        let status = self.command(ctx)?.spawn()?.wait()?;
+
+        self.check(status)
+    }
+
+    // reading changes nothing, so a dry run answers the question for real
+    pub fn capture(&self, ctx: &Ctx) -> Result<String> {
+        let out = self.command(ctx)?.output()?;
+
+        if !out.status.success() {
+            // output() holds stderr back, so the command's own complaint has to be
+            // passed on here or it is lost
+            let stderr = String::from_utf8_lossy(&out.stderr);
+
+            return Err(match stderr.trim() {
+                "" => self.failure(out.status),
+                reason => anyhow!("`{self}` failed: {reason}"),
+            });
+        }
+
+        Ok(String::from_utf8(out.stdout)?)
+    }
+
+    // replaces this process, so the command's exit code and signals become ours
+    pub fn replace(&self, ctx: &Ctx) -> Result<()> {
+        if self.skipped(ctx) {
+            return Ok(());
+        }
+
+        let error = self.command(ctx)?.exec();
+
+        Err(error).with_context(|| format!("running `{self}`"))
+    }
+
+    pub fn stream_to(&self, ctx: &Ctx, out: Stdio) -> Result<()> {
+        if self.skipped(ctx) {
+            return Ok(());
+        }
+
+        let status = self.command(ctx)?.stdout(out).spawn()?.wait()?;
+
+        self.check(status)
+    }
+
+    // the status rather than an error, for callers with something better to say
+    pub fn status(&self, ctx: &Ctx) -> Result<ExitStatus> {
+        Ok(self.command(ctx)?.spawn()?.wait()?)
+    }
+
+    pub fn stdin_from(&self, ctx: &Ctx, input: &Path) -> Result<ExitStatus> {
+        let stdin = self.opened(input)?;
+
+        Ok(self.command(ctx)?.stdin(stdin).spawn()?.wait()?)
+    }
+
+    // both streams held back, for callers that read the command's complaints
+    pub fn stdin_from_captured(&self, ctx: &Ctx, input: &Path) -> Result<Output> {
+        let stdin = self.opened(input)?;
+
+        Ok(self.command(ctx)?.stdin(stdin).output()?)
+    }
+
+    // whether it succeeded, without failure being an error
+    pub fn quietly_succeeds(&self, ctx: &Ctx) -> Result<bool> {
+        Ok(self
+            .command(ctx)?
+            .stdout(Stdio::null())
+            .spawn()?
+            .wait()?
+            .success())
+    }
+
+    fn command(&self, ctx: &Ctx) -> Result<Command> {
+        if ctx.verbose {
+            eprintln!("running `{self}`");
+        }
+
+        let (program, rest) = self.0.split_first().context("empty command")?;
+        let mut command = Command::new(program);
+        command.args(rest);
+
+        Ok(command)
+    }
+
+    fn skipped(&self, ctx: &Ctx) -> bool {
+        if ctx.dry_run {
+            eprintln!("would run `{self}`");
+            return true;
+        }
+
+        false
+    }
+
+    fn opened(&self, path: &Path) -> Result<Stdio> {
+        let file = File::open(path).with_context(|| format!("opening {}", path.display()))?;
+
+        Ok(Stdio::from(file))
+    }
+
+    fn check(&self, status: ExitStatus) -> Result<()> {
+        if status.success() {
+            return Ok(());
+        }
+
+        Err(self.failure(status))
+    }
+
+    fn failure(&self, status: ExitStatus) -> anyhow::Error {
+        match status.code() {
+            Some(code) => anyhow!("`{self}` exited with {code}"),
+            None => anyhow!("`{self}` was killed by a signal"),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::Argv;
+
+    #[test]
+    fn plain_words_are_left_as_they_are() {
+        let argv = Argv::new("pg_dump")
+            .flag("--username", "myproject")
+            .arg("--data-only")
+            .arg("myproject_db");
+
+        assert_eq!(
+            argv.quoted(),
+            "pg_dump --username myproject --data-only myproject_db"
+        );
+    }
+
+    #[test]
+    fn anything_a_shell_would_read_as_more_than_one_word_is_quoted() {
+        assert_eq!(Argv::new("psql").arg("my db").quoted(), "psql 'my db'");
+        assert_eq!(Argv::new("sh").arg("a | b").quoted(), "sh 'a | b'");
+        assert_eq!(Argv::new("echo").arg("").quoted(), "echo ''");
+        assert_eq!(Argv::new("echo").arg("it's").quoted(), r#"echo "it's""#);
+    }
+}

src/cmd/compose.rs

@@ -0,0 +1,92 @@
+use super::{Argv, Cmd};
+
+fn compose() -> Argv {
+    Argv::new("docker").arg("compose")
+}
+
+// docker compose run --rm, which runs the image's entrypoint and so fixes up the
+// container user before handing over
+pub struct Run {
+    service: String,
+    inner: Argv,
+}
+
+impl Run {
+    pub fn wrapping(service: &str, inner: Argv) -> Self {
+        Self {
+            service: service.to_string(),
+            inner,
+        }
+    }
+}
+
+impl Cmd for Run {
+    fn argv(&self) -> Argv {
+        compose()
+            .arg("run")
+            .arg("--rm")
+            .arg(&self.service)
+            .args(self.inner.words())
+    }
+}
+
+// the merged project file, with every extends and env_file resolved
+pub struct Config;
+
+impl Cmd for Config {
+    fn argv(&self) -> Argv {
+        compose().arg("config").flag("--format", "json")
+    }
+}
+
+// the container id a service is running as, empty when it is not running
+pub struct Ps {
+    service: String,
+}
+
+impl Ps {
+    pub fn id_of(service: &str) -> Self {
+        Self {
+            service: service.to_string(),
+        }
+    }
+}
+
+impl Cmd for Ps {
+    fn argv(&self) -> Argv {
+        compose().arg("ps").arg("--quiet").arg(&self.service)
+    }
+}
+
+pub struct Up;
+
+impl Cmd for Up {
+    fn argv(&self) -> Argv {
+        compose().arg("up").arg("--detach")
+    }
+}
+
+pub struct Start {
+    service: String,
+}
+
+impl Start {
+    pub fn service(service: &str) -> Self {
+        Self {
+            service: service.to_string(),
+        }
+    }
+}
+
+impl Cmd for Start {
+    fn argv(&self) -> Argv {
+        compose().arg("start").arg(&self.service)
+    }
+}
+pub struct Stop;
+
+impl Cmd for Stop {
+    fn argv(&self) -> Argv {
+        compose().arg("stop")
+    }
+}

src/cmd/django.rs

@@ -0,0 +1,44 @@
+use super::{Argv, Cmd};
+
+// an interactive shell in the django container
+pub struct Bash;
+
+impl Cmd for Bash {
+    fn argv(&self) -> Argv {
+        Argv::new("bash")
+    }
+}
+
+// django's manage.py, with the subcommand and its arguments
+pub struct Manage<'a> {
+    args: &'a [String],
+}
+
+impl<'a> Manage<'a> {
+    pub fn new(args: &'a [String]) -> Self {
+        Self { args }
+    }
+}
+
+impl Cmd for Manage<'_> {
+    fn argv(&self) -> Argv {
+        Argv::new("python").arg("manage.py").args(self.args)
+    }
+}
+
+// whatever the caller typed, passed through as it stands
+pub struct Words<'a> {
+    words: &'a [String],
+}
+
+impl<'a> Words<'a> {
+    pub fn new(words: &'a [String]) -> Self {
+        Self { words }
+    }
+}
+
+impl Cmd for Words<'_> {
+    fn argv(&self) -> Argv {
+        Argv::default().args(self.words)
+    }
+}

src/cmd/docker.rs

@@ -0,0 +1,69 @@
+use super::{Argv, Cmd};
+
+// docker exec, around a command that runs inside the container
+pub struct Exec {
+    container: String,
+    interactive: bool,
+    inner: Argv,
+}
+
+impl Exec {
+    pub fn wrapping(container: &str, inner: Argv) -> Self {
+        Self {
+            container: container.to_string(),
+            interactive: false,
+            inner,
+        }
+    }
+
+    // keep stdin open, for a command that is fed a dump
+    pub fn interactive(mut self) -> Self {
+        self.interactive = true;
+        self
+    }
+}
+
+impl Cmd for Exec {
+    fn argv(&self) -> Argv {
+        let mut argv = Argv::new("docker").arg("exec");
+
+        if self.interactive {
+            argv = argv.arg("--interactive");
+        }
+
+        argv.arg(&self.container).args(self.inner.words())
+    }
+}
+
+// docker cp, in either direction
+pub struct Cp {
+    from: String,
+    to: String,
+}
+
+impl Cp {
+    pub fn into_container(local: &str, container: &str, remote: &str) -> Self {
+        Self {
+            from: local.to_string(),
+            to: format!("{container}:{remote}"),
+        }
+    }
+
+    pub fn out_of_container(container: &str, remote: &str, local: &str) -> Self {
+        Self {
+            from: format!("{container}:{remote}"),
+            to: local.to_string(),
+        }
+    }
+}
+
+impl Cmd for Cp {
+    fn argv(&self) -> Argv {
+        // -L has no long form; it copies what a symlink points at
+        Argv::new("docker")
+            .arg("cp")
+            .arg("-L")
+            .arg(&self.from)
+            .arg(&self.to)
+    }
+}

src/cmd/mod.rs

@@ -0,0 +1,80 @@
+mod argv;
+mod compose;
+mod django;
+mod docker;
+mod postgres;
+mod shell;
+
+use std::{
+    path::Path,
+    process::{ExitStatus, Output, Stdio},
+};
+
+use anyhow::Result;
+
+pub use argv::Argv;
+pub use compose::{Config, Ps, Run, Start, Stop, Up};
+pub use django::{Bash, Manage, Words};
+pub use docker::{Cp, Exec};
+pub use postgres::{CreateDb, DropDb, PgDump, PgDumpAll, PgIsReady, PgRestore, Psql};
+pub use shell::{Gunzip, Gzip, Head, Pipeline, Rm};
+
+use crate::ctx::Ctx;
+
+// every command ahab runs, from `docker compose config` to `pg_dump`, is a type
+// that renders itself here; adapters wrap one command in another and terminals
+// run it, so a call site reads as one chain
+pub trait Cmd {
+    fn argv(&self) -> Argv;
+
+    // this command as one word list, for a shell to read
+    fn shell(&self) -> String {
+        self.argv().quoted()
+    }
+
+    // adapters
+    fn pipe(&self, next: &dyn Cmd) -> Pipeline {
+        Pipeline::starting(self.shell()).pipe(next)
+    }
+
+    fn in_container(&self, container: &str) -> Exec {
+        Exec::wrapping(container, self.argv())
+    }
+
+    fn in_service(&self, service: &str) -> Run {
+        Run::wrapping(service, self.argv())
+    }
+    // terminals
+    fn run(&self, ctx: &Ctx) -> Result<()> {
+        self.argv().run(ctx)
+    }
+
+    fn capture(&self, ctx: &Ctx) -> Result<String> {
+        self.argv().capture(ctx)
+    }
+
+    fn replace(&self, ctx: &Ctx) -> Result<()> {
+        self.argv().replace(ctx)
+    }
+
+    fn stream_to(&self, ctx: &Ctx, out: Stdio) -> Result<()> {
+        self.argv().stream_to(ctx, out)
+    }
+
+    // the status rather than an error, for callers with something better to say
+    fn status(&self, ctx: &Ctx) -> Result<ExitStatus> {
+        self.argv().status(ctx)
+    }
+
+    fn stdin_from(&self, ctx: &Ctx, input: &Path) -> Result<ExitStatus> {
+        self.argv().stdin_from(ctx, input)
+    }
+
+    fn stdin_from_captured(&self, ctx: &Ctx, input: &Path) -> Result<Output> {
+        self.argv().stdin_from_captured(ctx, input)
+    }
+
+    fn quietly_succeeds(&self, ctx: &Ctx) -> Result<bool> {
+        self.argv().quietly_succeeds(ctx)
+    }
+}

src/cmd/postgres.rs

@@ -0,0 +1,206 @@
+use super::{Argv, Cmd};
+
+// whether the server is accepting connections yet
+pub struct PgIsReady<'a> {
+    pub username: &'a str,
+    pub dbname: &'a str,
+}
+
+impl Cmd for PgIsReady<'_> {
+    fn argv(&self) -> Argv {
+        Argv::new("pg_isready")
+            .flag("--username", self.username)
+            .flag("--dbname", self.dbname)
+    }
+}
+
+pub struct DropDb<'a> {
+    pub username: &'a str,
+    pub dbname: &'a str,
+}
+
+impl Cmd for DropDb<'_> {
+    fn argv(&self) -> Argv {
+        Argv::new("dropdb")
+            .flag("--username", self.username)
+            .arg(self.dbname)
+    }
+}
+
+pub struct CreateDb<'a> {
+    pub username: &'a str,
+    pub dbname: &'a str,
+}
+
+impl Cmd for CreateDb<'_> {
+    fn argv(&self) -> Argv {
+        // template0 rather than template1, so nothing the local cluster picked up
+        // ends up in a restored database
+        Argv::new("createdb")
+            .flag("--username", self.username)
+            .flag("--encoding", "utf8")
+            .flag("--template", "template0")
+            .arg(self.dbname)
+    }
+}
+
+// reads a custom, tar or directory format dump
+pub struct PgRestore<'a> {
+    username: &'a str,
+    dbname: &'a str,
+    from: Option<&'a str>,
+}
+
+impl<'a> PgRestore<'a> {
+    pub fn new(username: &'a str, dbname: &'a str) -> Self {
+        Self {
+            username,
+            dbname,
+            from: None,
+        }
+    }
+
+    // a path in the container, for a dump that could not be streamed in
+    pub fn from(mut self, path: &'a str) -> Self {
+        self.from = Some(path);
+        self
+    }
+}
+
+impl Cmd for PgRestore<'_> {
+    fn argv(&self) -> Argv {
+        let argv = Argv::new("pg_restore")
+            .flag("--username", self.username)
+            .flag("--dbname", self.dbname);
+
+        match self.from {
+            Some(path) => argv.arg(path),
+            None => argv,
+        }
+    }
+}
+
+// reads a plain sql dump, of one database or of a whole cluster
+pub struct Psql<'a> {
+    username: &'a str,
+    dbname: &'a str,
+    atomic: bool,
+}
+
+impl<'a> Psql<'a> {
+    pub fn new(username: &'a str, dbname: &'a str) -> Self {
+        Self {
+            username,
+            dbname,
+            atomic: false,
+        }
+    }
+
+    // stop at the first error and undo the rest, which a cluster dump cannot do:
+    // it connects to each database itself, and trips over roles already there
+    pub fn atomic(mut self) -> Self {
+        self.atomic = true;
+        self
+    }
+}
+
+impl Cmd for Psql<'_> {
+    fn argv(&self) -> Argv {
+        let argv = Argv::new("psql")
+            .arg("--quiet")
+            .flag("--output", "/dev/null")
+            .flag("--username", self.username)
+            .flag("--dbname", self.dbname);
+
+        if self.atomic {
+            return argv
+                .flag("--variable", "ON_ERROR_STOP=1")
+                .arg("--single-transaction");
+        }
+
+        argv
+    }
+}
+
+pub struct PgDump<'a> {
+    username: &'a str,
+    dbname: &'a str,
+    format: &'a str,
+    to: Option<&'a str>,
+}
+
+impl<'a> PgDump<'a> {
+    pub fn new(username: &'a str, dbname: &'a str, format: &'a str) -> Self {
+        Self {
+            username,
+            dbname,
+            format,
+            to: None,
+        }
+    }
+
+    // a path in the container, for the directory format, which pg_dump writes
+    // itself rather than to stdout
+    pub fn to(mut self, path: &'a str) -> Self {
+        self.to = Some(path);
+        self
+    }
+}
+
+impl Cmd for PgDump<'_> {
+    fn argv(&self) -> Argv {
+        let argv = Argv::new("pg_dump")
+            .flag("--username", self.username)
+            .flag("--format", self.format);
+
+        match self.to {
+            Some(path) => argv.flag("--file", path),
+            None => argv,
+        }
+        .arg(self.dbname)
+    }
+}
+
+// the whole cluster, roles and all
+pub struct PgDumpAll<'a> {
+    pub username: &'a str,
+}
+
+impl Cmd for PgDumpAll<'_> {
+    fn argv(&self) -> Argv {
+        Argv::new("pg_dumpall").flag("--username", self.username)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{PgDump, Psql};
+    use crate::cmd::Cmd;
+
+    #[test]
+    fn a_directory_dump_names_the_file_it_writes() {
+        assert_eq!(
+            PgDump::new("u", "db", "d").to("/tmp/dump").argv().quoted(),
+            "pg_dump --username u --format d --file /tmp/dump db"
+        );
+        assert_eq!(
+            PgDump::new("u", "db", "c").argv().quoted(),
+            "pg_dump --username u --format c db"
+        );
+    }
+
+    #[test]
+    fn only_a_single_database_restore_stops_at_the_first_error() {
+        assert!(
+            Psql::new("u", "db")
+                .atomic()
+                .argv()
+                .quoted()
+                .ends_with("--variable 'ON_ERROR_STOP=1' --single-transaction")
+        );
+        assert_eq!(
+            Psql::new("u", "postgres").argv().quoted(),
+            "psql --quiet --output /dev/null --username u --dbname postgres"
+        );
+    }
+}

src/cmd/shell.rs

@@ -0,0 +1,150 @@
+use super::{Argv, Cmd};
+
+// sh -c, the only way to reach a shell feature inside a container
+pub struct Sh(String);
+
+impl Sh {
+    pub fn new(script: impl Into<String>) -> Self {
+        Self(script.into())
+    }
+}
+
+impl Cmd for Sh {
+    fn argv(&self) -> Argv {
+        // -c has no long form
+        Argv::new("sh").arg("-c").arg(&self.0)
+    }
+}
+
+// commands joined by pipes, which only a shell can run
+pub struct Pipeline {
+    stages: Vec<String>,
+    pipefail: bool,
+}
+
+impl Pipeline {
+    pub fn starting(first: String) -> Self {
+        Self {
+            stages: vec![first],
+            pipefail: true,
+        }
+    }
+
+    pub fn pipe(mut self, next: &dyn Cmd) -> Self {
+        self.stages.push(next.shell());
+        self
+    }
+
+    // for a pipeline whose last stage closes the pipe on purpose, where the
+    // SIGPIPE that kills an earlier stage is the expected end and not a failure
+    pub fn allow_early_close(mut self) -> Self {
+        self.pipefail = false;
+        self
+    }
+
+    fn script(&self) -> String {
+        let piped = self.stages.join(" | ");
+
+        // a pipeline reports only its last stage's status, so a first stage that
+        // dies mid-stream reads as success without this
+        if self.pipefail {
+            format!("set -o pipefail; {piped}")
+        } else {
+            piped
+        }
+    }
+}
+
+impl Cmd for Pipeline {
+    fn argv(&self) -> Argv {
+        Sh::new(self.script()).argv()
+    }
+}
+
+pub struct Gzip;
+
+impl Cmd for Gzip {
+    fn argv(&self) -> Argv {
+        Argv::new("gzip")
+    }
+}
+
+// busybox, which alpine based images ship, has no long options for these three
+pub struct Gunzip;
+
+impl Cmd for Gunzip {
+    fn argv(&self) -> Argv {
+        Argv::new("gunzip").arg("-c")
+    }
+}
+
+pub struct Head {
+    bytes: usize,
+}
+
+impl Head {
+    pub fn bytes(bytes: usize) -> Self {
+        Self { bytes }
+    }
+}
+
+impl Cmd for Head {
+    fn argv(&self) -> Argv {
+        Argv::new("head").arg("-c").arg(self.bytes.to_string())
+    }
+}
+
+pub struct Rm {
+    path: String,
+}
+
+impl Rm {
+    pub fn recursive(path: &str) -> Self {
+        Self {
+            path: path.to_string(),
+        }
+    }
+}
+
+impl Cmd for Rm {
+    fn argv(&self) -> Argv {
+        Argv::new("rm").arg("-rf").arg(&self.path)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{Gunzip, Gzip, Head};
+    use crate::cmd::Cmd;
+
+    #[test]
+    fn a_pipeline_reports_a_stage_that_dies_mid_stream() {
+        let script = Gunzip.pipe(&Gzip).shell();
+
+        assert_eq!(script, "sh -c 'set -o pipefail; gunzip -c | gzip'");
+    }
+
+    #[test]
+    fn a_pipeline_that_closes_the_pipe_on_purpose_keeps_the_default() {
+        let script = Gunzip.pipe(&Head::bytes(512)).allow_early_close().shell();
+
+        assert_eq!(script, "sh -c 'gunzip -c | head -c 512'");
+    }
+
+    #[test]
+    fn a_pipeline_reaches_a_container_as_one_argument() {
+        let argv = Gunzip.pipe(&Gzip).in_container("abc123").argv();
+
+        assert_eq!(
+            argv.words(),
+            [
+                "docker",
+                "exec",
+                "abc123",
+                "sh",
+                "-c",
+                "set -o pipefail; gunzip -c | gzip"
+            ]
+        );
+    }
+}

src/command_builder.rs

@@ -1,123 +0,0 @@
-use anyhow::{Context, Result, bail};
-use std::{
-    fmt::Display,
-    os::unix::process::CommandExt,
-    process::{Command, ExitStatus, Stdio},
-};
-
-use crate::ctx::Ctx;
-
-pub struct Args(Vec<String>);
-
-impl From<&str> for Args {
-    fn from(value: &str) -> Self {
-        Self(Vec::from_iter(value.split_whitespace().map(String::from)))
-    }
-}
-
-impl From<&[String]> for Args {
-    fn from(value: &[String]) -> Self {
-        Self(value.to_vec())
-    }
-}
-
-#[derive(Default)]
-pub struct CommandBuilder {
-    args: Vec<String>,
-}
-
-impl Display for CommandBuilder {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        f.write_str(&self.args.join(" "))
-    }
-}
-
-impl CommandBuilder {
-    pub fn docker() -> Self {
-        Self::default().args("docker")
-    }
-
-    pub fn docker_compose() -> Self {
-        Self::default().args("docker compose")
-    }
-
-    pub fn arg(mut self, arg: impl AsRef<str>) -> Self {
-        self.args.push(arg.as_ref().to_string());
-        self
-    }
-
-    pub fn args<T>(mut self, args: T) -> Self
-    where
-        Args: From<T>,
-    {
-        self.args.extend(Args::from(args).0);
-        self
-    }
-
-    pub fn build(self, ctx: &Ctx) -> Result<Command> {
-        if ctx.verbose {
-            eprintln!("running `{self}`");
-        }
-        let (first, rest) = self.args.split_first().context("empty args")?;
-        let mut command = Command::new(first);
-        command.args(rest);
-
-        Ok(command)
-    }
-
-    pub fn exec_get_stdout(self, ctx: &Ctx) -> Result<String> {
-        let shown = self.to_string();
-        let out = self.build(ctx)?.output()?;
-
-        check(&shown, out.status)?;
-        Ok(String::from_utf8(out.stdout)?)
-    }
-
-    pub fn exec(self, ctx: &Ctx) -> Result<()> {
-        if ctx.dry_run {
-            eprintln!("would run `{self}`");
-            return Ok(());
-        }
-
-        let shown = self.to_string();
-        let status = self.build(ctx)?.spawn()?.wait()?;
-
-        check(&shown, status)
-    }
-
-    // replaces this process, so the command's exit code and signals become ours
-    pub fn exec_replace(self, ctx: &Ctx) -> Result<()> {
-        if ctx.dry_run {
-            eprintln!("would run `{self}`");
-            return Ok(());
-        }
-
-        let shown = self.to_string();
-        let error = self.build(ctx)?.exec();
-
-        Err(error).with_context(|| format!("running `{shown}`"))
-    }
-
-    pub fn exec_redirect_stdout(self, ctx: &Ctx, stdio: Stdio) -> Result<()> {
-        if ctx.dry_run {
-            eprintln!("would run `{self}`");
-            return Ok(());
-        }
-
-        let shown = self.to_string();
-        let status = self.build(ctx)?.stdout(stdio).spawn()?.wait()?;
-
-        check(&shown, status)
-    }
-}
-
-fn check(command: &str, status: ExitStatus) -> Result<()> {
-    if status.success() {
-        return Ok(());
-    }
-
-    match status.code() {
-        Some(code) => bail!("`{command}` exited with {code}"),
-        None => bail!("`{command}` was killed by a signal"),
-    }
-}

src/compose.rs

@@ -1,7 +1,7 @@
-use anyhow::{Context, Result, anyhow, bail};
+use anyhow::{Context, Result, anyhow};
 use serde_json::Value;
 
-use crate::command_builder::CommandBuilder;
+use crate::cmd::{Cmd, Config};
 use crate::ctx::Ctx;
 
 const POSTGRES_IMAGES: [&str; 4] = ["postg", "timescale", "pgvector", "citus"];
@@ -13,21 +13,8 @@ pub struct Compose {
 
 impl Compose {
     pub fn resolve(ctx: &Ctx) -> Result<Self> {
-        let out = CommandBuilder::docker_compose()
-            .args("config --format json")
-            .build(ctx)?
-            .output()
-            .context("running docker compose config")?;
-
-        if !out.status.success() {
-            bail!(
-                "docker compose config failed: {}",
-                String::from_utf8_lossy(&out.stderr).trim()
-            );
-        }
-
-        let config: Value =
-            serde_json::from_slice(&out.stdout).context("parsing docker compose config")?;
+        let json = Config.capture(ctx)?;
+        let config: Value = serde_json::from_str(&json).context("parsing docker compose config")?;
         let services = config
             .get("services")
             .cloned()

src/main.rs

@@ -1,7 +1,7 @@
 use std::process::ExitCode;
 
 mod cli;
-mod command_builder;
+mod cmd;
 mod compose;
 mod ctx;
 mod scripts;

src/scripts/django.rs

@@ -4,9 +4,9 @@ use std::path::{Path, PathBuf};
 
 use anyhow::{Result, anyhow};
 
+use crate::cmd::{Bash, Cmd, Manage, Words};
 use crate::compose::Compose;
 use crate::ctx::Ctx;
-use crate::scripts::docker_compose;
 
 const DEBUG_TEMPLATE: &str = r#"from django.core.management.base import BaseCommand
 
@@ -55,19 +55,15 @@ pub fn make_command(app: &PathBuf, name: &str) -> Result<()> {
 }
 
 pub fn bash(ctx: &Ctx) -> Result<()> {
-    run(ctx, &["bash".to_string()])
+    Bash.in_service(&service(ctx)?).replace(ctx)
 }
 
 pub fn run(ctx: &Ctx, rest: &[String]) -> Result<()> {
-    let service = Compose::resolve(ctx)?.django()?;
-    docker_compose::run(ctx, &service, rest)
+    Words::new(rest).in_service(&service(ctx)?).replace(ctx)
 }
 
 pub fn manage(ctx: &Ctx, rest: &[String]) -> Result<()> {
-    let mut args = vec!["python".to_string(), "manage.py".to_string()];
-    args.extend_from_slice(rest);
-
-    run(ctx, &args)
+    Manage::new(rest).in_service(&service(ctx)?).replace(ctx)
 }
 
 // shortcuts
@@ -76,10 +72,10 @@ pub fn makemigrations(ctx: &Ctx) -> Result<()> {
 }
 
 pub fn migrate(ctx: &Ctx, rest: &[String]) -> Result<()> {
-    let mut full_rest = vec!["migrate".to_string()];
-    full_rest.extend_from_slice(rest);
+    let mut args = vec!["migrate".to_string()];
+    args.extend_from_slice(rest);
 
-    manage(ctx, &full_rest)
+    manage(ctx, &args)
 }
 
 pub fn shell(ctx: &Ctx) -> Result<()> {
@@ -90,6 +86,10 @@ pub fn test(ctx: &Ctx) -> Result<()> {
     manage(ctx, &["test".to_string()])
 }
 
+fn service(ctx: &Ctx) -> Result<String> {
+    Compose::resolve(ctx)?.django()
+}
+
 fn safe_create_file(path: PathBuf) -> Result<File, std::io::Error> {
     OpenOptions::new().write(true).create_new(true).open(path)
 }

src/scripts/docker_compose.rs

@@ -1,29 +0,0 @@
-use anyhow::Result;
-
-use crate::command_builder::CommandBuilder;
-use crate::ctx::Ctx;
-
-pub fn run(ctx: &Ctx, service: &str, rest: &[String]) -> Result<()> {
-    CommandBuilder::docker_compose()
-        .args("run --rm")
-        .arg(service)
-        .args(rest)
-        .exec_replace(ctx)
-}
-
-pub fn start(ctx: &Ctx, service: Option<&str>) -> Result<()> {
-    let mut command = CommandBuilder::docker_compose().args("start");
-    if let Some(service) = service {
-        command = command.arg(service);
-    }
-
-    command.exec(ctx)
-}
-
-pub fn stop(ctx: &Ctx) -> Result<()> {
-    CommandBuilder::docker_compose().args("stop").exec(ctx)
-}
-
-pub fn up(ctx: &Ctx) -> Result<()> {
-    CommandBuilder::docker_compose().args("up -d").exec(ctx)
-}

src/scripts/mod.rs

@@ -1,5 +1,4 @@
 pub mod completions;
 pub mod django;
-pub mod docker_compose;
 pub mod link;
 pub mod postgres;

src/scripts/postgres.rs

@@ -8,10 +8,13 @@ use std::{
     time::{Duration, Instant},
 };
 
-use super::docker_compose;
 use crate::cli::Format;
+use crate::cmd::{
+    Cmd, Cp, CreateDb, DropDb, Gunzip, Gzip, Head, PgDump, PgDumpAll, PgIsReady, PgRestore, Ps,
+    Psql, Rm, Start, Stop, Up,
+};
+use crate::compose::Compose;
 use crate::ctx::Ctx;
-use crate::{command_builder::CommandBuilder, compose::Compose};
 
 const CUSTOM_MAGIC: &[u8] = b"PGDMP";
 const TAR_MAGIC: &[u8] = b"toc.dat";
@@ -42,12 +45,7 @@ impl Database {
         let service = compose.postgres()?;
         let (user, name) = compose.postgres_credentials(&service);
 
-        let container = CommandBuilder::docker_compose()
-            .args("ps -q")
-            .arg(&service)
-            .exec_get_stdout(ctx)?
-            .trim()
-            .to_string();
+        let container = Ps::id_of(&service).capture(ctx)?.trim().to_string();
 
         if container.is_empty() {
             return Err(anyhow!("service {service} has no running container"));
@@ -61,16 +59,12 @@ impl Database {
         })
     }
 
-    fn restore_with(&self, kind: Kind) -> String {
+    // what reads this shape of dump back in
+    fn restore_with(&self, kind: Kind) -> Box<dyn Cmd + '_> {
         match kind {
-            Kind::Archive => format!("pg_restore -U {} --dbname={}", self.user, self.name),
-            Kind::Sql => format!(
-                "psql -q -o /dev/null -U {} -d {} -v ON_ERROR_STOP=1 --single-transaction",
-                self.user, self.name
-            ),
-            // a cluster dump connects to each database itself, and stopping at the
-            // first error would stop at a role that is already there
-            Kind::Cluster => format!("psql -q -o /dev/null -U {} -d postgres", self.user),
+            Kind::Archive => Box::new(PgRestore::new(&self.user, &self.name)),
+            Kind::Sql => Box::new(Psql::new(&self.user, &self.name).atomic()),
+            Kind::Cluster => Box::new(Psql::new(&self.user, "postgres")),
         }
     }
 }
@@ -132,9 +126,11 @@ fn is_existing_role_error(line: &str) -> bool {
     line.starts_with("ERROR:") && line.contains("role \"") && line.ends_with("already exists")
 }
 
-fn restore_cluster(ctx: &Ctx, db: &Database, script: &str, file: &Path) -> Result<()> {
-    let out = piped(ctx, db, script, file)?
-        .output()
+fn restore_cluster(ctx: &Ctx, db: &Database, restore: &dyn Cmd, file: &Path) -> Result<()> {
+    let out = restore
+        .in_container(&db.container)
+        .interactive()
+        .stdin_from_captured(ctx, file)
         .context("running psql")?;
 
     io::stdout().write_all(&out.stdout).ok();
@@ -182,26 +178,6 @@ fn read_header(path: &Path) -> Result<Vec<u8>> {
     Ok(header)
 }
 
-// a pipeline reports the last stage's status by default, so a first stage that
-// dies mid-stream looks like success; not for pipelines that close the pipe
-// early on purpose, where SIGPIPE would then read as failure
-fn pipefail(script: &str) -> String {
-    format!("set -o pipefail; {script}")
-}
-
-fn piped(ctx: &Ctx, db: &Database, script: &str, input: &Path) -> Result<std::process::Command> {
-    let file = File::open(input).with_context(|| format!("opening {}", input.display()))?;
-    let mut command = CommandBuilder::docker()
-        .args("exec -i")
-        .arg(&db.container)
-        .args("sh -c")
-        .arg(script)
-        .build(ctx)?;
-
-    command.stdin(Stdio::from(file));
-    Ok(command)
-}
-
 fn wait_until_ready(ctx: &Ctx, db: &Database) -> Result<()> {
     if ctx.dry_run {
         return Ok(());
@@ -210,18 +186,12 @@ fn wait_until_ready(ctx: &Ctx, db: &Database) -> Result<()> {
     let deadline = Instant::now() + READY_TIMEOUT;
 
     loop {
-        let ready = CommandBuilder::docker()
-            .args("exec")
-            .arg(&db.container)
-            .args("pg_isready -U")
-            .arg(&db.user)
-            .args("-d")
-            .arg(&db.name)
-            .build(ctx)?
-            .stdout(Stdio::null())
-            .spawn()?
-            .wait()?
-            .success();
+        let ready = PgIsReady {
+            username: &db.user,
+            dbname: &db.name,
+        }
+        .in_container(&db.container)
+        .quietly_succeeds(ctx)?;
 
         if ready {
             return Ok(());
@@ -239,13 +209,9 @@ fn wait_until_ready(ctx: &Ctx, db: &Database) -> Result<()> {
     }
 }
 
-fn when_ready(ctx: &Ctx, db: &Database, command: CommandBuilder) -> Result<()> {
+fn when_ready(ctx: &Ctx, db: &Database, command: &dyn Cmd) -> Result<()> {
     wait_until_ready(ctx, db)?;
-    command.exec(ctx)
-}
-
-fn in_container(db: &Database) -> CommandBuilder {
-    CommandBuilder::docker().args("exec").arg(&db.container)
+    command.run(ctx)
 }
 
 pub fn import(ctx: &Ctx, file: &Path) -> Result<()> {
@@ -253,10 +219,10 @@ pub fn import(ctx: &Ctx, file: &Path) -> Result<()> {
     let db = Database::resolve(ctx)?;
 
     eprintln!("stopping all containers");
-    docker_compose::stop(ctx)?;
+    Stop.run(ctx)?;
 
     eprintln!("starting db container");
-    docker_compose::start(ctx, Some(&db.service))?;
+    Start::service(&db.service).run(ctx)?;
 
     let remote = remote_dump();
 
@@ -265,24 +231,22 @@ pub fn import(ctx: &Ctx, file: &Path) -> Result<()> {
         when_ready(
             ctx,
             &db,
-            CommandBuilder::docker()
-                .args("cp -L")
-                .arg(file.to_string_lossy())
-                .arg(format!("{}:{remote}", db.container)),
+            &Cp::into_container(&file.to_string_lossy(), &db.container, &remote),
         )?;
     }
 
-    let (kind, restore) = match &dump {
-        Dump::Directory => (Kind::Archive, None),
-        Dump::Header(header) => {
-            let kind = Kind::of(header);
-
-            (kind, Some(db.restore_with(kind)))
-        }
+    let kind = match &dump {
+        Dump::Directory => Kind::Archive,
+        Dump::Header(header) => Kind::of(header),
         Dump::Gzip => {
             wait_until_ready(ctx, &db)?;
-            let out = piped(ctx, &db, &format!("gunzip -c | head -c {HEADER_LEN}"), file)?
-                .output()
+            let out = Gunzip
+                .pipe(&Head::bytes(HEADER_LEN))
+                // head closes the pipe once it has its bytes, which kills gunzip
+                .allow_early_close()
+                .in_container(&db.container)
+                .interactive()
+                .stdin_from_captured(ctx, file)
                 .context("reading the compressed dump's header")?;
 
             // head exits 0 whatever gunzip did, so an empty header is the only
@@ -302,12 +266,7 @@ pub fn import(ctx: &Ctx, file: &Path) -> Result<()> {
                 );
             }
 
-            let kind = Kind::of(&out.stdout);
-
-            (
-                kind,
-                Some(pipefail(&format!("gunzip -c | {}", db.restore_with(kind)))),
-            )
+            Kind::of(&out.stdout)
         }
     };
 
@@ -317,10 +276,11 @@ pub fn import(ctx: &Ctx, file: &Path) -> Result<()> {
     when_ready(
         ctx,
         &db,
-        in_container(&db)
-            .args("dropdb -U")
-            .arg(&db.user)
-            .arg(&db.name),
+        &DropDb {
+            username: &db.user,
+            dbname: &db.name,
+        }
+        .in_container(&db.container),
     )?;
 
     // a cluster dump creates the database itself, and would trip over one that
@@ -329,11 +289,11 @@ pub fn import(ctx: &Ctx, file: &Path) -> Result<()> {
         when_ready(
             ctx,
             &db,
-            in_container(&db)
-                .args("createdb -U")
-                .arg(&db.user)
-                .args("-E utf8 -T template0")
-                .arg(&db.name),
+            &CreateDb {
+                username: &db.user,
+                dbname: &db.name,
+            }
+            .in_container(&db.container),
         )?;
     }
 
@@ -341,22 +301,34 @@ pub fn import(ctx: &Ctx, file: &Path) -> Result<()> {
     if ctx.dry_run {
         eprintln!("would restore with {tool}");
     } else {
-        match (kind, restore.as_deref()) {
-            // psql's output is read rather than streamed here, to keep the expected
-            // role errors out of the way
-            (Kind::Cluster, Some(script)) => restore_cluster(ctx, &db, script, file)?,
-            (_, restore) => {
-                let status = match restore {
-                    Some(script) => piped(ctx, &db, script, file)?.spawn()?.wait()?,
-                    None => in_container(&db)
-                        .args("pg_restore -U")
-                        .arg(&db.user)
-                        .arg(format!("--dbname={}", db.name))
-                        .arg(&remote)
-                        .build(ctx)?
-                        .spawn()?
-                        .wait()?,
-                };
+        // a directory dump was copied in whole, so pg_restore reads it from the
+        // container; every other shape is fed in on stdin, through gunzip when it
+        // arrives compressed
+        if matches!(dump, Dump::Directory) {
+            let status = PgRestore::new(&db.user, &db.name)
+                .from(&remote)
+                .in_container(&db.container)
+                .status(ctx)?;
+
+            if !status.success() {
+                bail!("{tool} failed, the database is left empty");
+            }
+        } else {
+            let restore = db.restore_with(kind);
+            let restore: Box<dyn Cmd> = match dump {
+                Dump::Gzip => Box::new(Gunzip.pipe(&*restore)),
+                _ => restore,
+            };
+
+            if kind == Kind::Cluster {
+                // psql's output is read rather than streamed here, to keep the
+                // expected role errors out of the way
+                restore_cluster(ctx, &db, &*restore, file)?;
+            } else {
+                let status = restore
+                    .in_container(&db.container)
+                    .interactive()
+                    .stdin_from(ctx, file)?;
 
                 if !status.success() {
                     bail!("{tool} failed, the database is left empty");
@@ -366,12 +338,12 @@ pub fn import(ctx: &Ctx, file: &Path) -> Result<()> {
     }
 
     if matches!(dump, Dump::Directory) {
-        let _ = in_container(&db).args("rm -rf").arg(&remote).exec(ctx);
+        let _ = Rm::recursive(&remote).in_container(&db.container).run(ctx);
     }
 
     eprintln!("restarting containers");
-    docker_compose::stop(ctx)?;
-    docker_compose::up(ctx)?;
+    Stop.run(ctx)?;
+    Up.run(ctx)?;
 
     Ok(())
 }
@@ -401,25 +373,15 @@ pub fn dump(ctx: &Ctx, file: &PathBuf, format: Format, gzip: bool) -> Result<()>
         )
     };
 
-    let dumped = if gzip {
-        // the whole pipeline has to arrive as one shell argument
-        CommandBuilder::docker()
-            .args("exec")
-            .arg(&db.container)
-            .args("sh -c")
-            .arg(pipefail(&format!("{} | gzip", dump_command(&db, format))))
-            .exec_redirect_stdout(ctx, stdout)
-    } else {
-        let dumping = match format.flag() {
-            Some(flag) => in_container(&db)
-                .args("pg_dump -U")
-                .arg(&db.user)
-                .arg(format!("--format={flag}"))
-                .arg(&db.name),
-            None => in_container(&db).args("pg_dumpall -U").arg(&db.user),
-        };
+    let dumping = dump_command(&db, format);
 
-        dumping.exec_redirect_stdout(ctx, stdout)
+    let dumped = if gzip {
+        dumping
+            .pipe(&Gzip)
+            .in_container(&db.container)
+            .stream_to(ctx, stdout)
+    } else {
+        dumping.in_container(&db.container).stream_to(ctx, stdout)
     };
 
     if let Err(e) = dumped {
@@ -437,11 +399,11 @@ pub fn dump(ctx: &Ctx, file: &PathBuf, format: Format, gzip: bool) -> Result<()>
     Ok(())
 }
 
-// what to run inside the container, without the docker exec in front of it
-fn dump_command(db: &Database, format: Format) -> String {
+// pg_dump for one database, pg_dumpall for a cluster, which has no format letter
+fn dump_command(db: &Database, format: Format) -> Box<dyn Cmd + '_> {
     match format.flag() {
-        Some(flag) => format!("pg_dump -U {} --format={flag} {}", db.user, db.name),
-        None => format!("pg_dumpall -U {}", db.user),
+        Some(flag) => Box::new(PgDump::new(&db.user, &db.name, flag)),
+        None => Box::new(PgDumpAll { username: &db.user }),
     }
 }
 
@@ -458,21 +420,14 @@ fn dump_directory(ctx: &Ctx, db: &Database, target: &Path) -> Result<()> {
     eprintln!("dumping to local directory {}", target.display());
     let remote = remote_dump();
 
-    in_container(db)
-        .args("pg_dump -U")
-        .arg(&db.user)
-        .args("--format=d -f")
-        .arg(&remote)
-        .arg(&db.name)
-        .exec(ctx)?;
+    PgDump::new(&db.user, &db.name, "d")
+        .to(&remote)
+        .in_container(&db.container)
+        .run(ctx)?;
 
-    let copied = CommandBuilder::docker()
-        .args("cp")
-        .arg(format!("{}:{remote}", db.container))
-        .arg(target.to_string_lossy())
-        .exec(ctx);
+    let copied = Cp::out_of_container(&db.container, &remote, &target.to_string_lossy()).run(ctx);
 
-    let _ = in_container(db).args("rm -rf").arg(&remote).exec(ctx);
+    let _ = Rm::recursive(&remote).in_container(&db.container).run(ctx);
 
     copied
 }