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() +}