use crate::utils::trad::{t, tf};
use anyhow::{Context, Result};
use sea_orm::{ConnectionTrait, Database, DatabaseConnection, DbBackend, TransactionTrait};
use std::{fs, path::Path};
pub async fn up(migrations_path: &str) -> Result<()> {
dotenvy::dotenv().ok();
let migration_dir = migrations_path
.trim_end_matches("/src")
.trim_end_matches("\\src");
println!("{}", tf("migrate.applying", &[migration_dir]));
let status = tokio::process::Command::new("sea-orm-cli")
.args(["migrate", "up", "--migration-dir", migration_dir])
.status()
.await
.with_context(
|| "Unable to launch sea-orm-cli. Is it installed? Run: cargo install sea-orm-cli",
)?;
if !status.success() {
anyhow::bail!("sea-orm-cli migrate up failed (code: {:?})", status.code());
}
println!("{}", t("migrate.complete_up"));
Ok(())
}
pub async fn down(migrations_path: &str, files: Vec<String>, batch: Option<String>) -> Result<()> {
if files.is_empty() && batch.is_none() {
list_available(migrations_path)?;
return Ok(());
}
dotenvy::dotenv().ok();
let db_url = std::env::var("DATABASE_URL")
.with_context(|| "DATABASE_URL not set. Add it to your .env file.")?;
let db = Database::connect(&db_url)
.await
.with_context(|| "Failed to connect to database.")?;
if let Some(batch_ts) = batch {
rollback_batch(migrations_path, &batch_ts, &db).await?;
} else {
for file_arg in &files {
rollback_file(migrations_path, file_arg, &db).await?;
}
}
println!("\n{}", t("migrate.complete"));
Ok(())
}
pub fn status(migrations_path: &str) -> Result<()> {
println!("{}", tf("migrate.available_for", &[migrations_path]));
list_available(migrations_path)?;
Ok(())
}
async fn rollback_batch(
migrations_path: &str,
timestamp: &str,
db: &DatabaseConnection,
) -> Result<()> {
let by_time_dir = format!("{}/applied/by_time", migrations_path);
let batch_file = format!("{}/{}.rs", by_time_dir, timestamp);
if !Path::new(&batch_file).exists() {
anyhow::bail!(
"Batch '{}' not found.\nRun 'runique migration status' to list available batches.",
timestamp
);
}
check_order_batch(&by_time_dir, timestamp)?;
println!("{}", tf("migrate.rolling_back_batch", &[timestamp]));
let source = fs::read_to_string(&batch_file)
.with_context(|| format!("Cannot read batch file: {}", batch_file))?;
execute_down_block(&source, db).await?;
println!(" {}", tf("migrate.done", &[timestamp]));
Ok(())
}
async fn rollback_file(
migrations_path: &str,
file_arg: &str,
db: &DatabaseConnection,
) -> Result<()> {
let applied_dir = format!("{}/applied", migrations_path);
let file_path = if file_arg.ends_with(".rs") {
format!("{}/{}", applied_dir, file_arg)
} else {
format!("{}/{}.rs", applied_dir, file_arg)
};
if !Path::new(&file_path).exists() {
anyhow::bail!(
"File not found: {}\nRun 'runique migration status' to list available files.",
file_path
);
}
let parts: Vec<&str> = file_arg.trim_end_matches(".rs").splitn(2, '/').collect();
if parts.len() == 2 {
let table = parts[0];
let timestamp = parts[1];
let table_dir = format!("{}/{}", applied_dir, table);
check_order_file(&table_dir, timestamp)?;
}
println!("{}", tf("migrate.rolling_back", &[file_arg]));
let source = fs::read_to_string(&file_path)
.with_context(|| format!("Cannot read file: {}", file_path))?;
execute_down_block(&source, db).await?;
println!(" {}", tf("migrate.done", &[file_arg]));
Ok(())
}
fn check_order_batch(by_time_dir: &str, timestamp: &str) -> Result<()> {
let mut batches = list_files_in_dir(by_time_dir)?;
batches.sort();
if let Some(latest) = batches.last() {
let latest_ts = latest.trim_end_matches(".rs");
if latest_ts != timestamp {
anyhow::bail!(
"Cannot rollback batch '{}'.\nA more recent batch exists: '{}'.\nRollback most recent first.",
timestamp,
latest_ts
);
}
}
Ok(())
}
fn check_order_file(table_dir: &str, timestamp: &str) -> Result<()> {
let mut files = list_files_in_dir(table_dir)?;
files.sort();
if let Some(latest) = files.last() {
let latest_ts = latest.trim_end_matches(".rs");
if latest_ts != timestamp {
anyhow::bail!(
"Cannot rollback '{}'.\nA more recent migration exists for this table: '{}'.\nRollback most recent first.",
timestamp,
latest_ts
);
}
}
Ok(())
}
fn list_files_in_dir(dir: &str) -> Result<Vec<String>> {
if !Path::new(dir).exists() {
return Ok(vec![]);
}
let files = fs::read_dir(dir)?
.filter_map(|e| e.ok())
.filter(|e| e.path().extension().and_then(|x| x.to_str()) == Some("rs"))
.filter_map(|e| e.file_name().into_string().ok())
.collect();
Ok(files)
}
fn list_available(migrations_path: &str) -> Result<()> {
let applied_dir = format!("{}/applied", migrations_path);
if !Path::new(&applied_dir).exists() {
println!("{}", t("migrate.no_applied_dir"));
return Ok(());
}
let mut found = false;
let mut tables: Vec<String> = fs::read_dir(&applied_dir)?
.filter_map(|e| e.ok())
.filter(|e| e.path().is_dir())
.filter_map(|e| e.file_name().into_string().ok())
.filter(|n| n != "by_time")
.collect();
tables.sort();
for table in &tables {
let table_dir = format!("{}/{}", applied_dir, table);
let mut files = list_files_in_dir(&table_dir)?;
files.sort();
if !files.is_empty() {
found = true;
println!("\n {}:", table);
for file in &files {
println!(" {}/{}", table, file.trim_end_matches(".rs"));
}
}
}
let by_time_dir = format!("{}/by_time", applied_dir);
let mut batches = list_files_in_dir(&by_time_dir)?;
batches.sort();
if !batches.is_empty() {
found = true;
println!("\n {}:", t("migrate.by_time_header"));
for batch in &batches {
println!(" {}", batch.trim_end_matches(".rs"));
}
}
if !found {
println!("{}", t("migrate.none_available"));
} else {
println!("\n{}", t("migrate.usage"));
println!(" {}", t("migrate.usage_files"));
println!(" {}", t("migrate.usage_batch"));
}
Ok(())
}
async fn execute_down_block(source: &str, db: &DatabaseConnection) -> Result<()> {
let backend = db.get_database_backend();
let down_block = extract_fn_block(source, "down").unwrap_or_default();
let statements = extract_statements_from_block(&down_block, source, backend);
if statements.is_empty() {
println!(" {}", t("migrate.no_down"));
return Ok(());
}
let txn = db.begin().await.context("Failed to begin transaction")?;
let result: Result<()> = async {
for sql in &statements {
println!(" {}", tf("migrate.executing", &[sql]));
txn.execute_unprepared(sql)
.await
.with_context(|| format!("Failed to execute: {}", sql))?;
}
Ok(())
}
.await;
match result {
Ok(()) => txn.commit().await.context("Failed to commit transaction")?,
Err(e) => {
let _ = txn.rollback().await;
return Err(e);
}
}
Ok(())
}
fn extract_fn_block(source: &str, fn_name: &str) -> Option<String> {
let mut in_fn = false;
let mut depth: i32 = 0;
let mut out = String::new();
for line in source.lines() {
let trimmed = line.trim();
if !in_fn && trimmed.contains(&format!("async fn {}", fn_name)) {
in_fn = true;
out.push_str(line);
out.push('\n');
depth = depth.saturating_add(trimmed.chars().filter(|&c| c == '{').count() as i32);
depth = depth.saturating_sub(trimmed.chars().filter(|&c| c == '}').count() as i32);
if depth == 0 {
return Some(out);
}
continue;
}
if !in_fn {
continue;
}
out.push_str(line);
out.push('\n');
depth = depth.saturating_add(trimmed.chars().filter(|&c| c == '{').count() as i32);
depth = depth.saturating_sub(trimmed.chars().filter(|&c| c == '}').count() as i32);
if depth == 0 {
return Some(out);
}
}
None
}
fn extract_statements_from_block(
block: &str,
full_source: &str,
backend: DbBackend,
) -> Vec<String> {
let mut statements = Vec::new();
let table_name = extract_table_from_source(block)
.or_else(|| extract_table_from_source(full_source))
.unwrap_or_default();
if table_name.is_empty() {
return statements;
}
for line in block.lines() {
let trimmed = line.trim();
if trimmed.contains(".add_column(")
&& trimmed.contains("Alias::new(\"")
&& let Some(col) = extract_alias_value(trimmed)
{
let col_type = seaorm_sql_type(trimmed);
let null = if trimmed.contains(".null()") {
""
} else {
" NOT NULL"
};
let sql = match backend {
DbBackend::Postgres => format!(
"ALTER TABLE \"{}\" ADD COLUMN \"{}\" {}{};",
table_name, col, col_type, null
),
_ => format!(
"ALTER TABLE `{}` ADD COLUMN `{}` {}{};",
table_name, col, col_type, null
),
};
statements.push(sql);
}
if trimmed.contains(".drop_column(Alias::new(\"")
&& let Some(col) = extract_alias_value(trimmed)
{
let sql = match backend {
DbBackend::Postgres => {
format!("ALTER TABLE \"{}\" DROP COLUMN \"{}\";", table_name, col)
}
_ => format!("ALTER TABLE `{}` DROP COLUMN `{}`;", table_name, col),
};
statements.push(sql);
}
if trimmed.contains(".modify_column(")
&& trimmed.contains("Alias::new(\"")
&& let Some(col) = extract_alias_value(trimmed)
{
let make_null = trimmed.contains(".null()");
let sql = match backend {
DbBackend::Postgres => {
if make_null {
format!(
"ALTER TABLE \"{}\" ALTER COLUMN \"{}\" DROP NOT NULL;",
table_name, col
)
} else {
format!(
"ALTER TABLE \"{}\" ALTER COLUMN \"{}\" SET NOT NULL;",
table_name, col
)
}
}
_ => {
if make_null {
format!(
"-- WARNING: modify_column NULL not supported safely on this backend for `{}`.`{}`",
table_name, col
)
} else {
format!(
"-- WARNING: modify_column NOT NULL not supported safely on this backend for `{}`.`{}`",
table_name, col
)
}
}
};
statements.push(sql);
}
if trimmed.contains(".drop_table(")
&& trimmed.contains("Alias::new(\"")
&& let Some(t) = extract_alias_value(trimmed)
{
let sql = match backend {
DbBackend::Postgres => format!("DROP TABLE IF EXISTS \"{}\";", t),
_ => format!("DROP TABLE IF EXISTS `{}`;", t),
};
statements.push(sql);
}
if trimmed.contains(".drop_index(")
&& trimmed.contains(".name(\"")
&& let Some(idx) = extract_name_value(trimmed)
{
let sql = match backend {
DbBackend::Postgres => format!("DROP INDEX IF EXISTS \"{}\";", idx),
_ => format!("DROP INDEX `{}`;", idx),
};
statements.push(sql);
}
if trimmed.contains(".drop_foreign_key(")
&& trimmed.contains(".name(\"")
&& let Some(fk_name) = extract_name_value(trimmed)
{
let sql = match backend {
DbBackend::Postgres => format!(
"ALTER TABLE \"{}\" DROP CONSTRAINT IF EXISTS \"{}\";",
table_name, fk_name
),
_ => format!(
"ALTER TABLE `{}` DROP FOREIGN KEY `{}`;",
table_name, fk_name
),
};
statements.push(sql);
}
}
statements
}
fn extract_table_from_source(source: &str) -> Option<String> {
for line in source.lines() {
let trimmed = line.trim();
if trimmed.contains(".table(Alias::new(\"") {
return extract_alias_value(trimmed);
}
}
None
}
fn extract_alias_value(s: &str) -> Option<String> {
let marker = "Alias::new(\"";
let pos = s.find(marker)?.saturating_add(marker.len());
let end = s[pos..].find('"')?.saturating_add(pos);
Some(s[pos..end].to_string())
}
fn extract_name_value(s: &str) -> Option<String> {
let marker = ".name(\"";
let pos = s.find(marker)?.saturating_add(marker.len());
let end = s[pos..].find('"')?.saturating_add(pos);
Some(s[pos..end].to_string())
}
fn seaorm_sql_type(line: &str) -> &str {
if line.contains(".blob()")
|| line.contains(".binary(")
|| line.contains(".binary_len(")
|| line.contains(".var_binary(")
{
"BYTEA"
} else if line.contains(".text()") {
"TEXT"
} else if line.contains(".char()") || line.contains(".char_len(") {
"CHAR"
} else if line.contains(".tiny_integer()") || line.contains(".small_integer()") {
"SMALLINT"
} else if line.contains(".big_unsigned()") {
"BIGINT"
} else if line.contains(".unsigned()") {
"INTEGER"
} else if line.contains(".big_integer()") {
"BIGINT"
} else if line.contains(".integer()") {
"INTEGER"
} else if line.contains(".float()") {
"REAL"
} else if line.contains(".double()") {
"DOUBLE PRECISION"
} else if line.contains(".decimal(") || line.contains(".decimal_len(") {
"DECIMAL"
} else if line.contains(".boolean()") {
"BOOLEAN"
} else if line.contains(".timestamp_tz()") || line.contains(".timestamp_with_time_zone()") {
"TIMESTAMP WITH TIME ZONE"
} else if line.contains(".timestamp()")
|| line.contains(".date_time()")
|| line.contains(".datetime()")
{
"TIMESTAMP"
} else if line.contains(".date()") {
"DATE"
} else if line.contains(".time()") {
"TIME"
} else if line.contains(".uuid()") {
"UUID"
} else if line.contains(".json_binary()") {
"JSONB"
} else if line.contains(".json()") {
"JSON"
} else if line.contains(".inet()") {
"INET"
} else if line.contains(".cidr()") {
"CIDR"
} else if line.contains(".mac_address()") {
"MACADDR"
} else if line.contains(".interval()") {
"INTERVAL"
} else {
"VARCHAR(255)"
}
}