From d7626d9f5e00c7620bfd122d084378568d34d01e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Matej=20Jane=C5=BEi=C4=8D?= Date: Tue, 8 Sep 2026 11:54:21 +0000 Subject: [PATCH] refactor: give every command we run a type of its own --- Cargo.lock | 7 + Cargo.toml | 1 + src/cmd/argv.rs | 206 +++++++++++++++++++++++++++++ src/cmd/compose.rs | 92 +++++++++++++ src/cmd/django.rs | 44 +++++++ src/cmd/docker.rs | 69 ++++++++++ src/cmd/mod.rs | 80 ++++++++++++ src/cmd/postgres.rs | 206 +++++++++++++++++++++++++++++ src/cmd/shell.rs | 150 +++++++++++++++++++++ src/command_builder.rs | 123 ----------------- src/compose.rs | 21 +-- src/main.rs | 2 +- src/scripts/django.rs | 22 ++-- src/scripts/docker_compose.rs | 29 ----- src/scripts/mod.rs | 1 - src/scripts/postgres.rs | 239 ++++++++++++++-------------------- 16 files changed, 968 insertions(+), 324 deletions(-) create mode 100644 src/cmd/argv.rs create mode 100644 src/cmd/compose.rs create mode 100644 src/cmd/django.rs create mode 100644 src/cmd/docker.rs create mode 100644 src/cmd/mod.rs create mode 100644 src/cmd/postgres.rs create mode 100644 src/cmd/shell.rs delete mode 100644 src/command_builder.rs delete mode 100644 src/scripts/docker_compose.rs diff --git a/Cargo.lock b/Cargo.lock index a18df62..2a63163 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -10,6 +10,7 @@ dependencies = [ "clap", "clap_complete", "serde_json", + "shlex", ] [[package]] @@ -213,6 +214,12 @@ dependencies = [ "zmij", ] +[[package]] +name = "shlex" +version = "2.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba" + [[package]] name = "strsim" version = "0.11.1" diff --git a/Cargo.toml b/Cargo.toml index d6e03fe..0cc0a79 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -17,6 +17,7 @@ clap = { version = "4.6.6", features = ["derive", "env"] } clap_complete = "4.6.9" anyhow = "1.0.104" serde_json = "1.0.145" +shlex = "2.0.1" [build-dependencies] clap = { version = "4.6.6", features = ["derive", "env"] } diff --git a/src/cmd/argv.rs b/src/cmd/argv.rs new file mode 100644 index 0000000..e6d58b7 --- /dev/null +++ b/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); + +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) -> Self { + self.0.push(arg.as_ref().to_string()); + self + } + + pub fn args(mut self, args: I) -> Self + where + I: IntoIterator, + S: AsRef, + { + 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) -> 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 { + 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 { + Ok(self.command(ctx)?.spawn()?.wait()?) + } + + pub fn stdin_from(&self, ctx: &Ctx, input: &Path) -> Result { + 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 { + 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 { + Ok(self + .command(ctx)? + .stdout(Stdio::null()) + .spawn()? + .wait()? + .success()) + } + + fn command(&self, ctx: &Ctx) -> Result { + 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 { + 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""#); + } +} diff --git a/src/cmd/compose.rs b/src/cmd/compose.rs new file mode 100644 index 0000000..77f55c5 --- /dev/null +++ b/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") + } +} diff --git a/src/cmd/django.rs b/src/cmd/django.rs new file mode 100644 index 0000000..dea4580 --- /dev/null +++ b/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) + } +} diff --git a/src/cmd/docker.rs b/src/cmd/docker.rs new file mode 100644 index 0000000..d1fbc10 --- /dev/null +++ b/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) + } +} diff --git a/src/cmd/mod.rs b/src/cmd/mod.rs new file mode 100644 index 0000000..7cabfbe --- /dev/null +++ b/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 { + 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 { + self.argv().status(ctx) + } + + fn stdin_from(&self, ctx: &Ctx, input: &Path) -> Result { + self.argv().stdin_from(ctx, input) + } + + fn stdin_from_captured(&self, ctx: &Ctx, input: &Path) -> Result { + self.argv().stdin_from_captured(ctx, input) + } + + fn quietly_succeeds(&self, ctx: &Ctx) -> Result { + self.argv().quietly_succeeds(ctx) + } +} diff --git a/src/cmd/postgres.rs b/src/cmd/postgres.rs new file mode 100644 index 0000000..a39269f --- /dev/null +++ b/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" + ); + } +} diff --git a/src/cmd/shell.rs b/src/cmd/shell.rs new file mode 100644 index 0000000..850a590 --- /dev/null +++ b/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) -> 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, + 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" + ] + ); + } +} diff --git a/src/command_builder.rs b/src/command_builder.rs deleted file mode 100644 index 3f63082..0000000 --- a/src/command_builder.rs +++ /dev/null @@ -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); - -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, -} - -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) -> Self { - self.args.push(arg.as_ref().to_string()); - self - } - - pub fn args(mut self, args: T) -> Self - where - Args: From, - { - self.args.extend(Args::from(args).0); - self - } - - pub fn build(self, ctx: &Ctx) -> Result { - 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 { - 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"), - } -} diff --git a/src/compose.rs b/src/compose.rs index c604f81..9e89f4d 100644 --- a/src/compose.rs +++ b/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 { - 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() diff --git a/src/main.rs b/src/main.rs index 7546671..7d3aff6 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,7 +1,7 @@ use std::process::ExitCode; mod cli; -mod command_builder; +mod cmd; mod compose; mod ctx; mod scripts; diff --git a/src/scripts/django.rs b/src/scripts/django.rs index 915c9a5..0170839 100644 --- a/src/scripts/django.rs +++ b/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 { + Compose::resolve(ctx)?.django() +} + fn safe_create_file(path: PathBuf) -> Result { OpenOptions::new().write(true).create_new(true).open(path) } diff --git a/src/scripts/docker_compose.rs b/src/scripts/docker_compose.rs deleted file mode 100644 index b889798..0000000 --- a/src/scripts/docker_compose.rs +++ /dev/null @@ -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) -} diff --git a/src/scripts/mod.rs b/src/scripts/mod.rs index 451a4ad..e33f8ca 100644 --- a/src/scripts/mod.rs +++ b/src/scripts/mod.rs @@ -1,5 +1,4 @@ pub mod completions; pub mod django; -pub mod docker_compose; pub mod link; pub mod postgres; diff --git a/src/scripts/postgres.rs b/src/scripts/postgres.rs index e73e4e9..dbf7506 100644 --- a/src/scripts/postgres.rs +++ b/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 { 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> { 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 { - 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 = 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 { 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 }