From 5cb5a54fb0f84e52e405c6b753cbd3fcbf24b24f Mon Sep 17 00:00:00 2001 From: Brandur Date: Thu, 5 Nov 2026 15:19:02 -0600 Subject: [PATCH] [rust] Implement `river migrate-get` subcommand for Rust CLI Just happened to notice while checking for feature parity with Rust over in Pro land that even after some recent amendments to the Rust migration CLI, it was still missing the `river migrate-get` subcommand. Here, implement that with all the accoutrements of the Go reference implementation. Should operate very similarly: riverqueue migrate-get --all --exclude-version 1 --up > river.up.sql --- rust/CHANGELOG.md | 4 + rust/riverqueue-cli/README.md | 27 ++ rust/riverqueue-cli/src/main.rs | 7 +- rust/riverqueue-cli/src/migrate_get.rs | 205 +++++++++++++++ rust/riverqueue-cli/tests/migrate_get.rs | 315 +++++++++++++++++++++++ 5 files changed, 557 insertions(+), 1 deletion(-) create mode 100644 rust/riverqueue-cli/src/migrate_get.rs create mode 100644 rust/riverqueue-cli/tests/migrate_get.rs diff --git a/rust/CHANGELOG.md b/rust/CHANGELOG.md index 79baad8e9..52b92f01f 100644 --- a/rust/CHANGELOG.md +++ b/rust/CHANGELOG.md @@ -10,6 +10,10 @@ Changes to River for Go are recorded in the [repository changelog](../CHANGELOG. ## [Unreleased] +### Added + +- Add `riverqueue migrate-get` to export Postgres or SQLite migration SQL without a database connection, with Go-compatible version selection, exclusions, ordering, schema substitution, and migration headers. + ## [0.5.1] - 2026-11-05 ### Fixed diff --git a/rust/riverqueue-cli/README.md b/rust/riverqueue-cli/README.md index a1dfde5d9..80b0e3dea 100644 --- a/rust/riverqueue-cli/README.md +++ b/rust/riverqueue-cli/README.md @@ -33,6 +33,33 @@ to apply print `no migrations to apply`. Successful `validate` is silent. Outstanding migrations are reported on stderr with exit status 2; connection and usage errors exit with status 1. +### Exporting SQL + +`migrate-get` exports SQL for another migration framework without connecting to +a database. It defaults to Postgres and ignores `DATABASE_URL`; +use `--database-url sqlite://` to export SQLite SQL. + +```sh +riverqueue migrate-get --version 3 --up > river_3.up.sql +riverqueue migrate-get --version 3,2,1 --down > river.down.sql +riverqueue migrate-get --all --exclude-version 1 --up > river_all.up.sql +riverqueue migrate-get --all --down --database-url sqlite:// > river_sqlite.down.sql +``` + +Pass exactly one of `--all` or `--version`, and exactly one of `--up` or `--down`. +`--version` and `--exclude-version` accept comma-separated lists and repeated +flags. Explicit versions retain their supplied order; `--all` uses ascending +order for up migrations and descending order for down migrations. Exclusions +apply to either selection mode. Version 1 contains River's migration tables, +which are often unnecessary when using another migration framework. + +Each migration includes a header such as `-- River main migration 003 [up]`. +`--schema NAME` substitutes a schema prefix verbatim; include SQL identifier +quoting in the value if needed, for example +`--schema '"MyRiver"'`. Only the `main` migration line is bundled (`--line main`). +Exports require the corresponding `postgres` or `sqlite` build feature; both +are enabled by default in the CLI. + ## Benchmark `riverqueue bench` measures worker throughput and end-to-end latency. It diff --git a/rust/riverqueue-cli/src/main.rs b/rust/riverqueue-cli/src/main.rs index 71f3cd76d..cc2a72b7f 100644 --- a/rust/riverqueue-cli/src/main.rs +++ b/rust/riverqueue-cli/src/main.rs @@ -16,6 +16,7 @@ use std::{env, error::Error, process::ExitCode}; #[cfg(feature = "postgres")] mod bench; mod migrate; +mod migrate_get; const HELP: &str = "River for Rust @@ -24,13 +25,16 @@ Usage: Commands: migrate-down Run down migrations + migrate-get Export migration SQL without connecting to a database migrate-list List migrations, marking the latest applied version with * migrate-up Run up migrations validate Check that all migrations are applied bench Benchmark job throughput against a disposable database Run `riverqueue --help` for a command's options. Commands read the -database URL from --database-url or the DATABASE_URL environment variable. +database URL from --database-url or the DATABASE_URL environment variable, +except migrate-get, which defaults to Postgres and uses --database-url only +to select the database dialect. "; #[tokio::main] @@ -56,6 +60,7 @@ async fn run(mut arguments: Vec) -> Result<(), Box println!("riverqueue {}", env!("CARGO_PKG_VERSION")), #[cfg(feature = "postgres")] "bench" => bench::run(arguments, database_url_env).await?, + "migrate-get" => migrate_get::run(arguments)?, "migrate-down" | "migrate-list" | "migrate-up" | "validate" => { migrate::run(command, arguments, database_url_env).await?; } diff --git a/rust/riverqueue-cli/src/migrate_get.rs b/rust/riverqueue-cli/src/migrate_get.rs new file mode 100644 index 000000000..a78fbc5a1 --- /dev/null +++ b/rust/riverqueue-cli/src/migrate_get.rs @@ -0,0 +1,205 @@ +//! Offline export of River's canonical migration SQL, matching Go's CLI. + +use std::{ + error::Error, + io::{self, Write}, +}; + +#[cfg(feature = "postgres")] +use riverqueue_migrate::PostgresMigrator; +#[cfg(feature = "sqlite")] +use riverqueue_migrate::SqliteMigrator; +use riverqueue_migrate::{MIGRATION_LINE_MAIN, Migration}; + +const HELP: &str = "Retrieve River migration SQL for use with another migration framework + +Usage: + riverqueue migrate-get [options] + +Options: + --all Print all migrations; down uses descending order + --database-url URL Database dialect hint (default: postgres://) + --down Print down migrations + --exclude-version N Exclude versions (comma-separated or repeated) + --line LINE Migration line (default: main) + --schema NAME Schema prefix for migration SQL + --up Print up migrations + --version N Versions to print (comma-separated or repeated) + -h, --help Print help + +Pass exactly one of --all or --version, and exactly one of --up or --down. +Explicit versions retain their supplied order, including for down migrations. +No database connection is made. DATABASE_URL is ignored; use --database-url +to select Postgres (postgres:// or postgresql://) or SQLite (sqlite://). +Only the main migration line is bundled. + +Examples: + riverqueue migrate-get --version 3 --up > river_3.up.sql + riverqueue migrate-get --version 3,2,1 --down > river.down.sql + riverqueue migrate-get --all --exclude-version 1 --up > river_all.up.sql + riverqueue migrate-get --all --down --database-url sqlite:// > river_sqlite.down.sql +"; + +#[derive(Default)] +struct Args { + all: bool, + database_url: String, + down: bool, + exclude_version: Vec, + line: String, + schema: String, + up: bool, + version: Vec, +} + +/// Exports SQL without opening a database pool or consulting connection settings. +pub(crate) fn run( + arguments: impl IntoIterator, +) -> Result<(), Box> { + let Some(args) = parse_args(arguments)? else { + print!("{HELP}"); + return Ok(()); + }; + let migrations = all_versions(&args.database_url)?; + // Resolve every requested version before writing anything, even versions + // that will be excluded, just as Go does. + let mut selected = if args.all { + migrations.iter().collect::>() + } else { + args.version + .iter() + .map(|version| { + migrations + .iter() + .find(|migration| migration.version == *version) + .ok_or_else(|| { + let available: Vec<_> = migrations.iter().map(|m| m.version).collect(); + format!("migration {version} not found (available versions: {available:?})") + }) + }) + .collect::, _>>()? + }; + if args.all && args.down { + selected.reverse(); + } + let direction = if args.down { "down" } else { "up" }; + // Go's migrate-get substitutes the supplied schema verbatim. In contrast + // to executing migrations, callers can supply SQL quoting themselves. + let schema = if args.schema.is_empty() { + String::new() + } else { + format!("{}.", args.schema) + }; + let mut out = io::BufWriter::new(io::stdout().lock()); + for (index, migration) in selected + .into_iter() + .filter(|migration| !args.exclude_version.contains(&migration.version)) + .enumerate() + { + if index > 0 { + writeln!(out)?; + } + let sql = if args.down { + migration.down_sql + } else { + migration.up_sql + }; + writeln!( + out, + "-- River {MIGRATION_LINE_MAIN} migration {:03} [{direction}]", + migration.version + )?; + writeln!( + out, + "{}", + sql.replace("/* TEMPLATE: schema */", &schema).trim() + )?; + } + out.flush()?; + Ok(()) +} + +fn all_versions(database_url: &str) -> Result<&'static [Migration], Box> { + match database_url.split_once("://").map(|(scheme, _)| scheme) { + Some("postgres" | "postgresql") => { + #[cfg(feature = "postgres")] + return Ok(PostgresMigrator::all_versions()); + #[cfg(not(feature = "postgres"))] + return Err("Postgres support requires the `postgres` feature".into()); + } + Some("sqlite") => { + #[cfg(feature = "sqlite")] + return Ok(SqliteMigrator::all_versions()); + #[cfg(not(feature = "sqlite"))] + return Err("SQLite support requires the `sqlite` feature".into()); + } + _ => { + Err("--database-url must use a postgres://, postgresql://, or sqlite:// prefix".into()) + } + } +} + +fn parse_args( + arguments: impl IntoIterator, +) -> Result, Box> { + let mut raw = arguments.into_iter(); + let mut args = Args { + database_url: "postgres://".to_owned(), + ..Args::default() + }; + while let Some(argument) = raw.next() { + let (option, inline_value) = argument + .split_once('=') + .map_or((argument.as_str(), None), |(option, value)| { + (option, Some(value)) + }); + match option { + "--all" | "--down" | "--up" => { + let value = inline_value.map_or(Ok(true), str::parse::)?; + match option { + "--all" => args.all = value, + "--down" => args.down = value, + _ => args.up = value, + } + } + "--database-url" | "--exclude-version" | "--line" | "--schema" | "--version" => { + let value = inline_value + .map(str::to_owned) + .or_else(|| raw.next()) + .ok_or_else(|| format!("{option} requires a value"))?; + match option { + "--database-url" => args.database_url = value, + "--exclude-version" => { + args.exclude_version.extend(parse_versions(&value, option)?); + } + "--line" => args.line = value, + "--schema" => args.schema = value, + _ => args.version.extend(parse_versions(&value, option)?), + } + } + "-h" | "--help" => return Ok(None), + _ => return Err(format!("unknown argument {argument:?}\n\n{HELP}").into()), + } + } + if args.all != args.version.is_empty() { + return Err("pass exactly one of --all or --version".into()); + } + if args.down == args.up { + return Err("pass exactly one of --up or --down".into()); + } + if !args.line.is_empty() && args.line != MIGRATION_LINE_MAIN { + return Err(format!("migration line does not exist: {}", args.line).into()); + } + Ok(Some(args)) +} + +fn parse_versions(value: &str, option: &str) -> Result, Box> { + value + .split(',') + .map(|version| { + version.parse().map_err(|error| { + format!("invalid version {version:?} for {option}: {error}").into() + }) + }) + .collect() +} diff --git a/rust/riverqueue-cli/tests/migrate_get.rs b/rust/riverqueue-cli/tests/migrate_get.rs new file mode 100644 index 000000000..ae87052ee --- /dev/null +++ b/rust/riverqueue-cli/tests/migrate_get.rs @@ -0,0 +1,315 @@ +//! Exercise offline SQL export through the installed CLI entry point. + +use std::process::{Command, Output}; + +use riverqueue_migrate::Migration; + +#[test] +fn help_documents_export() { + let root = success(command(&["--help"]).output().unwrap()); + assert!(root.contains("migrate-get")); + for flag in ["-h", "--help"] { + let help = success(command(&["migrate-get", flag]).output().unwrap()); + for option in [ + "--all", + "--database-url", + "--down", + "--exclude-version", + "--line", + "--schema", + "--up", + "--version", + ] { + assert!(help.contains(option), "{help}"); + } + assert!(help.contains("No database connection is made")); + } +} + +#[cfg(feature = "postgres")] +#[test] +fn postgres_export() { + exercise_export( + "postgres://", + riverqueue_migrate::PostgresMigrator::all_versions(), + ); +} + +#[cfg(feature = "postgres")] +#[test] +fn postgres_is_default_and_ignores_connection_settings() { + let arguments = ["migrate-get", "--version", "1", "--up"]; + let expected = success(command(&arguments).output().unwrap()); + assert!(expected.contains("id bigserial PRIMARY KEY")); + for database_url in ["sqlite://", "not a database URL"] { + let output = command(&arguments) + .env("DATABASE_URL", database_url) + .env("PGHOST", "invalid.invalid") + .env("PGPORT", "not a port") + .output() + .unwrap(); + assert_eq!(success(output), expected); + } + for database_url in ["postgres://127.0.0.1:1/missing", "postgresql://"] { + let output = command(&arguments) + .args(["--database-url", database_url]) + .output() + .unwrap(); + assert_eq!(success(output), expected); + } +} + +#[cfg(not(feature = "postgres"))] +#[test] +fn postgres_requires_feature() { + failure( + command(&["migrate-get", "--all", "--up"]).output().unwrap(), + "Postgres support requires the `postgres` feature", + ); +} + +#[cfg(feature = "sqlite")] +#[test] +fn sqlite_does_not_create_database() { + let path = std::env::temp_dir().join(format!( + "river-cli-export-{}-{}.db", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos() + )); + let database_url = format!("sqlite://{}?mode=rwc", path.display()); + let output = success(run(&database_url, &["--version", "1", "--up"])); + assert!(output.contains("id integer PRIMARY KEY")); + assert!(!path.exists(), "export must not create a database file"); +} + +#[cfg(feature = "sqlite")] +#[test] +fn sqlite_export() { + exercise_export( + "sqlite://", + riverqueue_migrate::SqliteMigrator::all_versions(), + ); +} + +#[cfg(not(feature = "sqlite"))] +#[test] +fn sqlite_requires_feature() { + failure( + run("sqlite://", &["--all", "--up"]), + "SQLite support requires the `sqlite` feature", + ); +} + +fn command(arguments: &[&str]) -> Command { + let mut command = Command::new(env!("CARGO_BIN_EXE_riverqueue")); + command.env_remove("DATABASE_URL").args(arguments); + command +} + +fn exercise_export(database_url: &str, migrations: &[Migration]) { + for direction in ["down", "up"] { + let flag = format!("--{direction}"); + assert_eq!( + success(run(database_url, &["--version", "1", &flag])), + expected_sql(migrations, &[1], direction, "") + ); + assert_eq!( + success(run( + database_url, + &["--version=3,1", "--version", "2,3", &flag] + )), + expected_sql(migrations, &[3, 1, 2, 3], direction, "") + ); + + let mut all: Vec<_> = migrations + .iter() + .map(|migration| migration.version) + .collect(); + if direction == "down" { + all.reverse(); + } + assert_eq!( + success(run(database_url, &["--all", &flag])), + expected_sql(migrations, &all, direction, "") + ); + all.retain(|version| ![1, 2, 3].contains(version)); + assert_eq!( + success(run( + database_url, + &[ + "--all", + "--exclude-version=1,3", + "--exclude-version", + "2,999", + &flag, + ] + )), + expected_sql(migrations, &all, direction, "") + ); + assert_eq!( + success(run( + database_url, + &["--version", "3,1,2", "--exclude-version", "1,3", &flag] + )), + expected_sql(migrations, &[2], direction, "") + ); + assert_eq!( + success(run( + database_url, + &["--version", "1,2", "--exclude-version", "1,2", &flag] + )), + "" + ); + + for schema in ["", "custom_schema", "\"MyRiver\""] { + let output = success(run( + database_url, + &[ + "--version", + "1", + &flag, + "--schema", + schema, + "--line", + "main", + ], + )); + assert_eq!(output, expected_sql(migrations, &[1], direction, schema)); + if !schema.is_empty() { + assert!(output.contains(&format!("{schema}.river_migration"))); + } + assert!(!output.contains("/* TEMPLATE:")); + } + } + exercise_invalid_arguments(database_url); +} + +fn exercise_invalid_arguments(database_url: &str) { + for (arguments, error) in [ + (vec!["--up"], "pass exactly one of --all or --version"), + ( + vec!["--all", "--version", "1", "--up"], + "pass exactly one of --all or --version", + ), + (vec!["--all"], "pass exactly one of --up or --down"), + ( + vec!["--all", "--up", "--down"], + "pass exactly one of --up or --down", + ), + (vec!["--version", "0", "--up"], "migration 0 not found"), + (vec!["--version", "-1", "--down"], "migration -1 not found"), + ( + vec!["--version", "1,999", "--up"], + "migration 999 not found", + ), + ( + vec!["--version", "1,999", "--exclude-version", "999", "--up"], + "migration 999 not found", + ), + (vec!["--version", "1,x", "--up"], "invalid version"), + (vec!["--version", "1,", "--up"], "invalid version"), + (vec!["--version", "", "--up"], "invalid version"), + ( + vec!["--all", "--exclude-version", "x", "--up"], + "invalid version", + ), + ( + vec!["--all", "--up", "--line", "missing"], + "migration line does not exist: missing", + ), + ( + vec!["--all", "--up", "--database-url", "mysql://"], + "--database-url must use a postgres://, postgresql://, or sqlite:// prefix", + ), + ] { + failure(run(database_url, &arguments), error); + } + for option in [ + "--database-url", + "--exclude-version", + "--line", + "--schema", + "--version", + ] { + failure( + run(database_url, &["--all", "--up", option]), + &format!("{option} requires a value"), + ); + } + for argument in [ + "--dry-run", + "--max-steps", + "--show-sql", + "--target-version", + "unexpected", + ] { + failure( + run(database_url, &["--all", "--up", argument]), + "unknown argument", + ); + } +} + +fn expected_sql( + migrations: &[Migration], + versions: &[i64], + direction: &str, + schema: &str, +) -> String { + versions + .iter() + .map(|version| { + let migration = migrations.iter().find(|m| m.version == *version).unwrap(); + let sql = if direction == "down" { + migration.down_sql + } else { + migration.up_sql + }; + let prefix = if schema.is_empty() { + String::new() + } else { + format!("{schema}.") + }; + format!( + "-- River main migration {version:03} [{direction}]\n{}\n", + sql.replace("/* TEMPLATE: schema */", &prefix).trim() + ) + }) + .collect::>() + .join("\n") +} + +fn failure(output: Output, message: &str) { + assert_eq!(output.status.code(), Some(1)); + assert!( + output.stdout.is_empty(), + "errors must not export partial SQL" + ); + let stderr = String::from_utf8(output.stderr).unwrap(); + assert!(stderr.contains(message), "{stderr}"); +} + +fn run(database_url: &str, arguments: &[&str]) -> Output { + command(&["migrate-get", "--database-url", database_url]) + .args(arguments) + .output() + .unwrap() +} + +fn success(output: Output) -> String { + assert!( + output.status.success(), + "stdout: {}\nstderr: {}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + assert!( + output.stderr.is_empty(), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + String::from_utf8(output.stdout).unwrap() +}