toolu_orm_cli/migrate/
run.rs

1//! Migration runner that applies pending migrations from a directory in order.
2
3use std::path::Path;
4
5use toolu_orm_connection::DbConnection;
6use toolu_orm_core::dialect::Dialect;
7use toolu_orm_core::journal::{Journal, JournalEntry};
8
9use super::apply::apply_migration;
10use super::error::MigrateError;
11use super::pending::get_pending_migrations;
12use super::store::{ensure_migrations_table, get_applied_migrations, record_migration};
13use super::transaction::{begin, commit, rollback_after};
14
15/// Applies pending migrations from the given directory to the database.
16///
17/// Migrations without a directory on the target machine — a single-binary
18/// distribution — use [`run_migrate_embedded`](super::run_migrate_embedded)
19/// instead; both share the same apply path.
20///
21/// # Errors
22///
23/// Returns `MigrateError` on database, I/O, or hash mismatch failures.
24pub async fn run_migrate(
25  conn: &impl DbConnection,
26  migrations_dir: &str,
27  dialect: Dialect,
28) -> Result<u32, MigrateError> {
29  ensure_migrations_table(conn, dialect).await?;
30  let applied = get_applied_migrations(conn).await?;
31
32  let journal_path = Path::new(migrations_dir).join("_journal.json");
33  let journal_path_str = journal_path.to_str().unwrap_or("");
34
35  let journal = Journal::read_from_path(journal_path_str)
36    .map_err(|e| MigrateError::ReadFile(format!("{e}")))?;
37
38  if journal.entries.is_empty() {
39    let pending = get_pending_migrations(migrations_dir, &applied)?;
40    let mut count: u32 = 0;
41    for migration_file in &pending {
42      apply_migration_legacy(conn, migrations_dir, migration_file, dialect).await?;
43      count += 1;
44    }
45    return Ok(count);
46  }
47
48  let mut count: u32 = 0;
49  for entry in &journal.entries {
50    if applied.contains(&entry.name) {
51      continue;
52    }
53    apply_journal_entry(conn, migrations_dir, entry, dialect).await?;
54    count += 1;
55  }
56
57  Ok(count)
58}
59
60/// Reads the entry's file and hands it to the shared apply path.
61async fn apply_journal_entry(
62  conn: &impl DbConnection,
63  migrations_dir: &str,
64  entry: &JournalEntry,
65  dialect: Dialect,
66) -> Result<(), MigrateError> {
67  let sql_path = Path::new(migrations_dir).join(&entry.name);
68  let content = std::fs::read_to_string(&sql_path)
69    .map_err(|e| MigrateError::ReadFile(format!("{}: {e}", sql_path.display())))?;
70
71  apply_migration(conn, &entry.name, &content, &entry.hash, dialect).await
72}
73
74async fn apply_migration_legacy(
75  conn: &impl DbConnection,
76  migrations_dir: &str,
77  migration_file: &str,
78  dialect: Dialect,
79) -> Result<(), MigrateError> {
80  let path = Path::new(migrations_dir).join(migration_file);
81  let sql = std::fs::read_to_string(&path)
82    .map_err(|e| MigrateError::ReadFile(format!("{}: {e}", path.display())))?;
83
84  begin(conn).await?;
85
86  let result = async {
87    conn
88      .execute_batch(&sql)
89      .await
90      .map_err(|e| MigrateError::Database(format!("migration {migration_file}: {e}")))?;
91    record_migration(conn, migration_file, "", dialect).await
92  }
93  .await;
94
95  if let Err(e) = result {
96    return Err(rollback_after(conn, e).await);
97  }
98
99  commit(conn).await
100}
toolu_orm_cli/migrate/run.rs

toolu_orm_cli/migrate/
run.rs
