use std::fs;
use std::io::Write;
use std::path::{Path, PathBuf};
use color_eyre::eyre::{Result, WrapErr, bail};
use fse_schema::{Schema, diff_schemas, parse, snapshot};
use crate::config::{self, OrmConfig};
#[derive(Debug, Default)]
pub struct MigrateOpts {
pub dry_run: bool,
pub assume_yes: bool,
pub no_prepare: bool,
pub database_url: Option<String>,
}
#[derive(Debug)]
pub struct MigrateOutcome {
pub generated: Option<PathBuf>,
pub needs_manual_edit: bool,
}
pub async fn run(root: &Path, opts: &MigrateOpts) -> Result<MigrateOutcome> {
let cfg = config::load(root)?;
let mut external = Vec::new();
let mut module_schemas = Vec::new();
for module in crate::modules::discover(root, &cfg)? {
let schema = crate::modules::load_schema(&module)?;
external.extend(schema.tables.iter().cloned());
module_schemas.push((module.name, schema));
}
let mut new_schema = parse_tables(root, &cfg, &external)?;
for (name, schema) in module_schemas {
new_schema.merge(schema, &name)?;
}
validate_required_columns(&new_schema, &cfg)?;
let snapshot_path = root.join(&cfg.snapshot_path);
let old_schema = if snapshot_path.exists() {
let raw = fs::read_to_string(&snapshot_path)
.wrap_err_with(|| format!("cannot read {}", snapshot_path.display()))?;
snapshot::schema_from_json(&raw)?
} else {
Schema::default()
};
let migrations_dir = root.join(&cfg.migrations_dir);
let mut outcome = MigrateOutcome {
generated: None,
needs_manual_edit: false,
};
if let Some(migration) = diff_schemas(&old_schema, &new_schema)? {
println!("Schema change: {}\n", migration.summary);
println!("{}", migration.sql);
if migration.destructive {
println!("!! this migration is destructive (data is dropped)\n");
}
if opts.dry_run {
println!("dry run: nothing written.");
return Ok(outcome);
}
if !opts.assume_yes && !confirm(migration.destructive)? {
println!("aborted: nothing written.");
return Ok(outcome);
}
fs::create_dir_all(&migrations_dir)
.wrap_err_with(|| format!("cannot create {}", migrations_dir.display()))?;
let file = migrations_dir.join(format!(
"{}_{}.sql",
next_version(&migrations_dir)?,
slug_or_default(&migration.filename_slug()),
));
fs::write(&file, &migration.sql)
.wrap_err_with(|| format!("cannot write {}", file.display()))?;
write_snapshot(&snapshot_path, &new_schema)?;
println!("wrote {}", file.display());
outcome.needs_manual_edit = migration.needs_manual_edit;
outcome.generated = Some(file);
if migration.needs_manual_edit {
println!(
"\nThe migration contains a TODO (a new NOT NULL column needs a backfill).\n\
Edit the file, then run `fse migrate` again to apply it."
);
return Ok(outcome);
}
} else {
if !snapshot_path.exists() {
write_snapshot(&snapshot_path, &new_schema)?;
}
println!("schema up to date.");
}
if opts.dry_run {
return Ok(outcome);
}
apply_pending(root, &cfg, opts, &migrations_dir).await?;
if !opts.no_prepare {
crate::prepare::run(root, &cfg, opts.database_url.as_deref())?;
}
Ok(outcome)
}
fn parse_tables(
root: &Path,
cfg: &OrmConfig,
external: &[fse_schema::TableDef],
) -> Result<Schema> {
let dir = root.join(&cfg.tables_dir);
if !dir.exists() {
bail!(
"tables folder {} does not exist (set orm.tables_dir in fse.toml)",
dir.display()
);
}
let mut sources = Vec::new();
for entry in fs::read_dir(&dir).wrap_err_with(|| dir.display().to_string())? {
let path = entry.wrap_err_with(|| dir.display().to_string())?.path();
if path.extension().is_some_and(|e| e == "rs") {
sources.push((
path.file_name().unwrap().to_string_lossy().into_owned(),
fs::read_to_string(&path).wrap_err_with(|| path.display().to_string())?,
));
}
}
sources.sort();
if sources.is_empty() {
bail!("no .rs files in {}", dir.display());
}
Ok(parse::parse_sources_with_external(&sources, external)?)
}
fn validate_required_columns(schema: &Schema, cfg: &OrmConfig) -> Result<()> {
for (table_name, columns) in &cfg.required_columns {
let Some(table) = schema.table(table_name) else {
bail!(
"fse.toml requires a `{table_name}` table, but no #[derive(Table)] struct defines it"
);
};
for column in columns {
if table.column(column).is_none() {
bail!(
"fse.toml requires column `{column}` on `{table_name}` — the framework depends on it; add it back to the struct"
);
}
}
}
Ok(())
}
fn write_snapshot(path: &Path, schema: &Schema) -> Result<()> {
if let Some(parent) = path.parent() {
fs::create_dir_all(parent)
.wrap_err_with(|| format!("cannot create {}", parent.display()))?;
}
fs::write(path, snapshot::schema_to_json(schema))
.wrap_err_with(|| format!("cannot write {}", path.display()))
}
fn next_version(migrations_dir: &Path) -> Result<u64> {
let mut version: u64 = chrono::Utc::now()
.format("%Y%m%d%H%M%S")
.to_string()
.parse()
.expect("timestamp is numeric");
let mut existing = Vec::new();
if migrations_dir.exists() {
for entry in
fs::read_dir(migrations_dir).wrap_err_with(|| migrations_dir.display().to_string())?
{
let name = entry
.wrap_err_with(|| migrations_dir.display().to_string())?
.file_name();
let name = name.to_string_lossy();
let digits: String = name.chars().take_while(char::is_ascii_digit).collect();
if let Ok(v) = digits.parse::<u64>() {
existing.push(v);
}
}
}
while existing.contains(&version) {
version += 1;
}
Ok(version)
}
fn slug_or_default(slug: &str) -> &str {
if slug.is_empty() { "schema" } else { slug }
}
fn confirm(destructive: bool) -> Result<bool> {
if destructive {
print!("apply this DESTRUCTIVE migration? type `yes` to continue: ");
} else {
print!("write and apply? [Y/n] ");
}
std::io::stdout().flush().ok();
let mut answer = String::new();
std::io::stdin()
.read_line(&mut answer)
.wrap_err("cannot read from stdin")?;
let answer = answer.trim().to_lowercase();
Ok(if destructive {
answer == "yes"
} else {
answer.is_empty() || answer == "y" || answer == "yes"
})
}
async fn apply_pending(
root: &Path,
cfg: &OrmConfig,
opts: &MigrateOpts,
migrations_dir: &Path,
) -> Result<()> {
if !migrations_dir.exists() {
return Ok(());
}
let url = config::resolve_database_url(root, cfg, opts.database_url.as_deref())?;
let options = url
.parse::<sqlx::sqlite::SqliteConnectOptions>()
.wrap_err("invalid database url")?
.create_if_missing(true)
.foreign_keys(false);
let pool = sqlx::SqlitePool::connect_with(options)
.await
.wrap_err("cannot open database")?;
let migrator = sqlx::migrate::Migrator::new(migrations_dir.to_path_buf())
.await
.wrap_err("invalid migrations folder")?;
migrator.run(&pool).await.wrap_err("migration failed")?;
let violations: Vec<(String, Option<i64>, String)> =
sqlx::query_as("SELECT \"table\", rowid, parent FROM pragma_foreign_key_check")
.fetch_all(&pool)
.await
.wrap_err("foreign_key_check failed")?;
pool.close().await;
if !violations.is_empty() {
let examples: Vec<String> = violations
.iter()
.take(5)
.map(|(table, rowid, parent)| match rowid {
Some(rowid) => format!("{table} rowid {rowid} -> missing {parent} row"),
None => format!("{table} -> missing {parent} row"),
})
.collect();
bail!(
"migrations applied, but the database now has {} foreign key violation(s):\n {}\n\
fix the offending rows (or the migration that orphaned them) and rerun.",
violations.len(),
examples.join("\n "),
);
}
println!("database is up to date.");
Ok(())
}