use std::env;
use std::fs;
use std::path::{Path, PathBuf};
use anyhow::{Context, Result};
use rayon::prelude::*;
use vespertide_config::VespertideConfig;
use vespertide_core::MigrationPlan;
use vespertide_planner::validate_migration_plan;
use crate::parallel_config::{LOAD_FILES_PAR_MIN_LEN, LOAD_FILES_PAR_THRESHOLD};
pub fn load_migrations(config: &VespertideConfig) -> Result<Vec<MigrationPlan>> {
let migrations_dir = config.migrations_dir();
if !migrations_dir.exists() {
return Ok(Vec::new());
}
let paths = collect_migration_paths(migrations_dir)?;
let results: Vec<Result<MigrationPlan>> = if paths.len() < LOAD_FILES_PAR_THRESHOLD {
paths.iter().map(|path| load_migration_file(path)).collect()
} else {
paths
.par_iter()
.with_min_len(LOAD_FILES_PAR_MIN_LEN)
.map(|path| load_migration_file(path))
.collect()
};
let mut plans = Vec::with_capacity(results.len());
for result in results {
plans.push(result?);
}
plans.sort_by_key(|p| p.version);
Ok(plans)
}
pub fn load_migrations_from_dir(
project_root: Option<PathBuf>,
) -> Result<Vec<MigrationPlan>, Box<dyn std::error::Error>> {
let project_root = if let Some(root) = project_root {
root
} else {
let manifest_dir = env::var("CARGO_MANIFEST_DIR")
.map_err(|_| "CARGO_MANIFEST_DIR environment variable not set")?;
PathBuf::from(manifest_dir)
};
let config = crate::config::load_config_or_default(Some(project_root.clone()))
.map_err(|e| format!("Failed to load config: {e}"))?;
let migrations_dir = project_root.join(config.migrations_dir());
if !migrations_dir.exists() {
return Ok(Vec::new());
}
let paths = collect_migration_paths_internal(&migrations_dir)?;
let results: Vec<Result<MigrationPlan, String>> = if paths.len() < LOAD_FILES_PAR_THRESHOLD {
paths
.iter()
.map(|path| load_migration_file_internal(path))
.collect()
} else {
paths
.par_iter()
.with_min_len(LOAD_FILES_PAR_MIN_LEN)
.map(|path| load_migration_file_internal(path))
.collect()
};
let mut plans = Vec::with_capacity(results.len());
for result in results {
plans.push(result.map_err(|e| -> Box<dyn std::error::Error> { e.into() })?);
}
plans.sort_by_key(|p| p.version);
Ok(plans)
}
fn collect_migration_paths(dir: &Path) -> Result<Vec<PathBuf>> {
let entries = fs::read_dir(dir).context("read migrations directory")?;
let mut paths = Vec::new();
for entry in entries {
let entry = entry.context("read directory entry")?;
let path = entry.path();
if path.is_file() && has_migration_extension(&path) {
paths.push(path);
}
}
Ok(paths)
}
fn collect_migration_paths_internal(dir: &Path) -> Result<Vec<PathBuf>, String> {
let entries =
fs::read_dir(dir).map_err(|e| format!("Failed to read migrations directory: {e}"))?;
let mut paths = Vec::new();
for entry in entries {
let entry = entry.map_err(|e| format!("Failed to read directory entry: {e}"))?;
let path = entry.path();
if path.is_file() && has_migration_extension(&path) {
paths.push(path);
}
}
Ok(paths)
}
fn has_migration_extension(path: &Path) -> bool {
matches!(
path.extension().and_then(|s| s.to_str()),
Some("json" | "yaml" | "yml")
)
}
fn load_migration_file(path: &Path) -> Result<MigrationPlan> {
let ext = path.extension().and_then(|s| s.to_str());
let content = fs::read_to_string(path)
.with_context(|| format!("read migration file: {}", path.display()))?;
let plan: MigrationPlan = if ext == Some("json") {
serde_json::from_str(&content)
.with_context(|| format!("parse migration: {}", path.display()))?
} else {
serde_yaml::from_str(&content)
.with_context(|| format!("parse migration: {}", path.display()))?
};
validate_migration_plan(&plan)
.with_context(|| format!("validate migration: {}", path.display()))?;
Ok(plan)
}
fn load_migration_file_internal(path: &Path) -> Result<MigrationPlan, String> {
let ext = path.extension().and_then(|s| s.to_str());
let content = fs::read_to_string(path)
.map_err(|e| format!("Failed to read migration file {}: {}", path.display(), e))?;
let plan: MigrationPlan = if ext == Some("json") {
serde_json::from_str(&content)
.map_err(|e| format!("Failed to parse JSON migration {}: {}", path.display(), e))?
} else {
serde_yaml::from_str(&content)
.map_err(|e| format!("Failed to parse YAML migration {}: {}", path.display(), e))?
};
validate_migration_plan(&plan)
.map_err(|e| format!("Failed to validate migration {}: {}", path.display(), e))?;
Ok(plan)
}
pub fn load_migrations_at_compile_time() -> Result<Vec<MigrationPlan>, Box<dyn std::error::Error>> {
load_migrations_from_dir(None)
}
#[cfg(test)]
mod tests {
use super::*;
use serial_test::serial;
use std::env;
use std::fs;
use tempfile::TempDir;
struct CwdGuard {
original: PathBuf,
}
impl CwdGuard {
fn new(dir: &PathBuf) -> Self {
let original = env::current_dir().unwrap();
env::set_current_dir(dir).unwrap();
Self { original }
}
}
impl Drop for CwdGuard {
fn drop(&mut self) {
let _ = env::set_current_dir(&self.original);
}
}
fn write_config(dir: &std::path::Path) {
let cfg = VespertideConfig::default();
let text = serde_json::to_string_pretty(&cfg).unwrap();
fs::write(dir.join("vespertide.json"), text).unwrap();
}
#[test]
#[serial]
fn test_load_migrations_returns_empty_when_no_migrations_dir() {
let temp_dir = TempDir::new().unwrap();
let _guard = CwdGuard::new(&temp_dir.path().to_path_buf());
write_config(temp_dir.path());
let result = load_migrations(&VespertideConfig::default()).unwrap();
assert!(result.is_empty());
}
#[test]
#[serial]
fn test_load_migrations_reads_json_and_sorts_versions() {
let temp_dir = TempDir::new().unwrap();
let _guard = CwdGuard::new(&temp_dir.path().to_path_buf());
write_config(temp_dir.path());
fs::create_dir_all("migrations").unwrap();
fs::write(
"migrations/0002_second.json",
r#"{"version": 2, "actions": []}"#,
)
.unwrap();
fs::write(
"migrations/0001_first.json",
r#"{"version": 1, "actions": []}"#,
)
.unwrap();
let plans = load_migrations(&VespertideConfig::default()).unwrap();
assert_eq!(
plans.iter().map(|plan| plan.version).collect::<Vec<_>>(),
vec![1, 2]
);
}
#[test]
#[serial]
fn load_migrations_ignores_non_migration_extension_files() {
let temp_dir = TempDir::new().unwrap();
let _guard = CwdGuard::new(&temp_dir.path().to_path_buf());
write_config(temp_dir.path());
fs::create_dir_all("migrations").unwrap();
fs::write("migrations/notes.txt", "not a migration: {{{ invalid").unwrap();
let plans = load_migrations(&VespertideConfig::default()).unwrap();
assert_eq!(plans.len(), 0, "the .txt file must be skipped");
}
#[test]
fn load_migrations_from_dir_ignores_non_migration_extension_files() {
let temp_dir = TempDir::new().unwrap();
let migrations_dir = temp_dir.path().join("migrations");
fs::create_dir_all(&migrations_dir).unwrap();
fs::write(
migrations_dir.join("notes.txt"),
"not a migration: {{{ invalid",
)
.unwrap();
let result = load_migrations_from_dir(Some(temp_dir.path().to_path_buf()));
assert!(
result.is_ok(),
"the .txt file must be skipped, not parsed: {result:?}"
);
assert_eq!(result.unwrap().len(), 0);
}
#[test]
fn test_load_migrations_from_dir_with_no_migrations_dir() {
let temp_dir = TempDir::new().unwrap();
let result = load_migrations_from_dir(Some(temp_dir.path().to_path_buf()));
assert!(result.is_ok());
assert_eq!(result.unwrap().len(), 0);
}
#[test]
fn test_load_migrations_from_dir_with_empty_migrations_dir() {
let temp_dir = TempDir::new().unwrap();
let migrations_dir = temp_dir.path().join("migrations");
fs::create_dir_all(&migrations_dir).unwrap();
let result = load_migrations_from_dir(Some(temp_dir.path().to_path_buf()));
assert!(result.is_ok());
assert_eq!(result.unwrap().len(), 0);
}
#[test]
fn test_load_migrations_from_dir_with_json_migration() {
let temp_dir = TempDir::new().unwrap();
let migrations_dir = temp_dir.path().join("migrations");
fs::create_dir_all(&migrations_dir).unwrap();
let migration_content = r#"{
"version": 1,
"actions": [
{
"type": "create_table",
"table": "users",
"columns": [
{
"name": "id",
"type": "integer",
"nullable": false
}
],
"constraints": []
}
]
}"#;
fs::write(migrations_dir.join("0001_test.json"), migration_content).unwrap();
let result = load_migrations_from_dir(Some(temp_dir.path().to_path_buf()));
assert!(result.is_ok());
let plans = result.unwrap();
assert_eq!(plans.len(), 1);
assert_eq!(plans[0].version, 1);
}
#[test]
fn test_load_migrations_from_dir_sorts_by_version() {
let temp_dir = TempDir::new().unwrap();
let migrations_dir = temp_dir.path().join("migrations");
fs::create_dir_all(&migrations_dir).unwrap();
let migration1 = r#"{"version": 2, "actions": []}"#;
let migration2 = r#"{"version": 1, "actions": []}"#;
let migration3 = r#"{"version": 3, "actions": []}"#;
fs::write(migrations_dir.join("0002_second.json"), migration1).unwrap();
fs::write(migrations_dir.join("0001_first.json"), migration2).unwrap();
fs::write(migrations_dir.join("0003_third.json"), migration3).unwrap();
let result = load_migrations_from_dir(Some(temp_dir.path().to_path_buf()));
assert!(result.is_ok());
let plans = result.unwrap();
assert_eq!(plans.len(), 3);
assert_eq!(plans[0].version, 1);
assert_eq!(plans[1].version, 2);
assert_eq!(plans[2].version, 3);
}
#[test]
fn test_load_migrations_from_dir_with_yaml_migration() {
let temp_dir = TempDir::new().unwrap();
let migrations_dir = temp_dir.path().join("migrations");
fs::create_dir_all(&migrations_dir).unwrap();
let migration_content = r"---
version: 1
actions:
- type: create_table
table: users
columns:
- name: id
type: integer
nullable: false
constraints: []
";
fs::write(migrations_dir.join("0001_test.yaml"), migration_content).unwrap();
let result = load_migrations_from_dir(Some(temp_dir.path().to_path_buf()));
assert!(result.is_ok());
let plans = result.unwrap();
assert_eq!(plans.len(), 1);
assert_eq!(plans[0].version, 1);
}
#[test]
fn test_load_migrations_from_dir_with_yml_migration() {
let temp_dir = TempDir::new().unwrap();
let migrations_dir = temp_dir.path().join("migrations");
fs::create_dir_all(&migrations_dir).unwrap();
let migration_content = r"---
version: 1
actions:
- type: create_table
table: users
columns:
- name: id
type: integer
nullable: false
constraints: []
";
fs::write(migrations_dir.join("0001_test.yml"), migration_content).unwrap();
let result = load_migrations_from_dir(Some(temp_dir.path().to_path_buf()));
assert!(result.is_ok());
let plans = result.unwrap();
assert_eq!(plans.len(), 1);
assert_eq!(plans[0].version, 1);
}
#[test]
#[serial]
fn test_load_migrations_reads_yaml_for_runtime_loader() {
let temp_dir = TempDir::new().unwrap();
let _guard = CwdGuard::new(&temp_dir.path().to_path_buf());
write_config(temp_dir.path());
fs::create_dir_all("migrations").unwrap();
let migration_content = r"---
version: 1
actions:
- type: create_table
table: users
columns:
- name: id
type: integer
nullable: false
constraints: []
";
fs::write("migrations/0001_test.yaml", migration_content).unwrap();
let plans = load_migrations(&VespertideConfig::default()).unwrap();
assert_eq!(plans.len(), 1);
assert_eq!(plans[0].version, 1);
}
#[test]
fn test_load_migrations_from_dir_with_invalid_json() {
let temp_dir = TempDir::new().unwrap();
let migrations_dir = temp_dir.path().join("migrations");
fs::create_dir_all(&migrations_dir).unwrap();
let invalid_json = r#"{"version": 1, "actions": [invalid]}"#;
fs::write(migrations_dir.join("0001_invalid.json"), invalid_json).unwrap();
let result = load_migrations_from_dir(Some(temp_dir.path().to_path_buf()));
assert!(result.is_err());
let err_msg = result.unwrap_err().to_string();
assert!(err_msg.contains("Failed to parse JSON migration"));
}
#[test]
fn test_load_migrations_from_dir_with_invalid_yaml() {
let temp_dir = TempDir::new().unwrap();
let migrations_dir = temp_dir.path().join("migrations");
fs::create_dir_all(&migrations_dir).unwrap();
let invalid_yaml = r"---
version: 1
actions:
- invalid: [syntax
";
fs::write(migrations_dir.join("0001_invalid.yaml"), invalid_yaml).unwrap();
let result = load_migrations_from_dir(Some(temp_dir.path().to_path_buf()));
assert!(result.is_err());
let err_msg = result.unwrap_err().to_string();
assert!(err_msg.contains("Failed to parse YAML migration"));
}
#[test]
fn test_load_migrations_from_dir_rejects_invalid_plan_with_file_path() {
let temp_dir = TempDir::new().unwrap();
let migrations_dir = temp_dir.path().join("migrations");
fs::create_dir_all(&migrations_dir).unwrap();
let migration_path = migrations_dir.join("0001_invalid_plan.json");
let invalid_plan = r#"{
"version": 1,
"actions": [
{
"type": "add_column",
"table": "nonexistent",
"column": {
"name": "required_value",
"type": "integer",
"nullable": false
}
}
]
}"#;
fs::write(&migration_path, invalid_plan).unwrap();
let result = load_migrations_from_dir(Some(temp_dir.path().to_path_buf()));
assert!(result.is_err());
let err_msg = result.unwrap_err().to_string();
assert!(
err_msg.contains(&migration_path.display().to_string()),
"error did not include file path {migration_path:?}: {err_msg}"
);
}
#[test]
fn test_load_migrations_from_dir_with_unreadable_file() {
let temp_dir = TempDir::new().unwrap();
let migrations_dir = temp_dir.path().join("migrations");
fs::create_dir_all(&migrations_dir).unwrap();
let file_path = migrations_dir.join("0001_test.json");
fs::write(&file_path, r#"{"version": 1, "actions": []}"#).unwrap();
let result = load_migrations_from_dir(Some(temp_dir.path().to_path_buf()));
assert!(result.is_ok());
}
#[test]
#[serial]
fn test_load_migrations_from_dir_without_project_root() {
let original = env::var("CARGO_MANIFEST_DIR").ok();
unsafe {
env::remove_var("CARGO_MANIFEST_DIR");
}
let result = load_migrations_from_dir(None);
assert!(result.is_err());
let err_msg = result.unwrap_err().to_string();
assert!(err_msg.contains("CARGO_MANIFEST_DIR environment variable not set"));
if let Some(val) = original {
unsafe {
env::set_var("CARGO_MANIFEST_DIR", val);
}
}
}
#[test]
fn test_load_migrations_at_compile_time() {
let result = load_migrations_at_compile_time();
let _ = result;
}
}