toolu_orm_cli/migrate/
run.rs

1//! Migration runner that applies pending migrations in order.
2
3use std::path::Path;
4
5use toolu_orm_connection::DbConnection;
6use toolu_orm_core::dialect::Dialect;
7use toolu_orm_core::journal::{compute_hash, Journal, JournalEntry};
8
9use super::error::MigrateError;
10use super::pending::get_pending_migrations;
11use super::store::{ensure_migrations_table, get_applied_migrations, record_migration};
12
13/// Applies pending migrations from the given directory to the database.
14///
15/// # Errors
16///
17/// Returns `MigrateError` on database, I/O, or hash mismatch failures.
18pub async fn run_migrate(
19  conn: &impl DbConnection,
20  migrations_dir: &str,
21  dialect: Dialect,
22) -> Result<u32, MigrateError> {
23  ensure_migrations_table(conn, dialect).await?;
24  let applied = get_applied_migrations(conn).await?;
25
26  let journal_path = Path::new(migrations_dir).join("_journal.json");
27  let journal_path_str = journal_path.to_str().unwrap_or("");
28
29  let journal = Journal::read_from_path(journal_path_str)
30    .map_err(|e| MigrateError::ReadFile(format!("{e}")))?;
31
32  if journal.entries.is_empty() {
33    let pending = get_pending_migrations(migrations_dir, &applied)?;
34    let mut count: u32 = 0;
35    for migration_file in &pending {
36      apply_migration_legacy(conn, migrations_dir, migration_file, dialect).await?;
37      count += 1;
38    }
39    return Ok(count);
40  }
41
42  let mut count: u32 = 0;
43  for entry in &journal.entries {
44    if applied.contains(&entry.name) {
45      continue;
46    }
47    apply_journal_entry(conn, migrations_dir, entry, dialect).await?;
48    count += 1;
49  }
50
51  Ok(count)
52}
53
54/// Verifies the entry's hash, then applies its statements and records it in
55/// one transaction; any failure rolls the whole file back.
56async fn apply_journal_entry(
57  conn: &impl DbConnection,
58  migrations_dir: &str,
59  entry: &JournalEntry,
60  dialect: Dialect,
61) -> Result<(), MigrateError> {
62  let sql_path = Path::new(migrations_dir).join(&entry.name);
63  let content = std::fs::read_to_string(&sql_path)
64    .map_err(|e| MigrateError::ReadFile(format!("{}: {e}", sql_path.display())))?;
65
66  let actual_hash = compute_hash(&content);
67  if actual_hash != entry.hash {
68    return Err(MigrateError::HashMismatch {
69      file: entry.name.clone(),
70      expected: entry.hash.clone(),
71      actual: actual_hash,
72    });
73  }
74
75  conn
76    .execute_batch("BEGIN")
77    .await
78    .map_err(|e| MigrateError::Database(format!("begin transaction: {e}")))?;
79
80  let exec_result = async {
81    execute_migration_statements(conn, &content, &entry.name).await?;
82    record_migration(conn, &entry.name, &entry.hash, dialect).await
83  }
84  .await;
85
86  match exec_result {
87    Err(e) => Err(rollback_after(conn, e).await),
88    Ok(()) => conn
89      .execute_batch("COMMIT")
90      .await
91      .map_err(|e| MigrateError::Database(format!("commit transaction: {e}"))),
92  }
93}
94
95/// Rolls back after `err`. A ROLLBACK that fails itself (connection lost) is
96/// appended to the message so neither error is lost.
97async fn rollback_after(conn: &impl DbConnection, err: MigrateError) -> MigrateError {
98  match conn.execute_batch("ROLLBACK").await {
99    Ok(()) => err,
100    Err(rollback_err) => {
101      MigrateError::Database(format!("{err}; rollback also failed: {rollback_err}"))
102    },
103  }
104}
105
106async fn execute_migration_statements(
107  conn: &impl DbConnection,
108  content: &str,
109  file_label: &str,
110) -> Result<(), MigrateError> {
111  for statement in content.split("--> statement-breakpoint") {
112    if !has_statement(statement) {
113      continue;
114    }
115    conn
116      .execute_sql(statement.trim(), vec![])
117      .await
118      .map_err(|e| MigrateError::Database(format!("{file_label}: {e}")))?;
119  }
120  Ok(())
121}
122
123/// A chunk with only blank lines and `--` line comments (the generator emits
124/// such chunks for operations a dialect cannot express, and never emits block
125/// comments) must not reach the driver: libsql reports "not an error" when
126/// asked to execute an empty statement.
127fn has_statement(chunk: &str) -> bool {
128  chunk
129    .lines()
130    .map(str::trim)
131    .any(|line| !line.is_empty() && !line.starts_with("--"))
132}
133
134async fn apply_migration_legacy(
135  conn: &impl DbConnection,
136  migrations_dir: &str,
137  migration_file: &str,
138  dialect: Dialect,
139) -> Result<(), MigrateError> {
140  let path = Path::new(migrations_dir).join(migration_file);
141  let sql = std::fs::read_to_string(&path)
142    .map_err(|e| MigrateError::ReadFile(format!("{}: {e}", path.display())))?;
143
144  conn
145    .execute_batch("BEGIN")
146    .await
147    .map_err(|e| MigrateError::Database(format!("begin transaction: {e}")))?;
148
149  let result = async {
150    conn
151      .execute_batch(&sql)
152      .await
153      .map_err(|e| MigrateError::Database(format!("migration {migration_file}: {e}")))?;
154    record_migration(conn, migration_file, "", dialect).await
155  }
156  .await;
157
158  if let Err(e) = result {
159    return Err(rollback_after(conn, e).await);
160  }
161
162  conn
163    .execute_batch("COMMIT")
164    .await
165    .map_err(|e| MigrateError::Database(format!("commit transaction: {e}")))?;
166
167  result
168}
toolu_orm_cli/migrate/run.rs

toolu_orm_cli/migrate/
run.rs
