rust_ef/migration/
engine_ops.rs

1//! `MigrationEngine` database operations — apply/revert migrations, ensure schema, seed data.
2
3use super::dialect::MigrationDialect;
4use super::engine::MigrationEngine;
5use super::history::{
6    create_migration_history_table_sql, seed_insert_sql, split_sql_statements,
7    MigrationHistoryEntry,
8};
9use super::types::Migration;
10use crate::entity_snapshot::EntitySnapshot;
11use crate::error::{EFError, EFResult};
12use crate::metadata::EntityTypeMeta;
13use crate::provider::IDatabaseProvider;
14
15impl MigrationEngine {
16    /// Ensures the migration history table exists.
17    pub async fn ensure_history_table(&self, provider: &dyn IDatabaseProvider) -> EFResult<()> {
18        let sql = create_migration_history_table_sql(self.dialect);
19        provider.execute_migration_command(&sql).await
20    }
21
22    /// Applies a migration's Up SQL against a database provider.
23    ///
24    /// PostgreSQL/SQLite execute the whole script (including the history row)
25    /// in one transaction, so a failed statement rolls the migration back
26    /// atomically. MySQL DDL implicitly commits, so its statements are executed
27    /// individually (partial application is possible; the migration stays
28    /// unrecorded and must be repaired manually).
29    pub async fn apply(
30        &self,
31        provider: &dyn IDatabaseProvider,
32        migration: &Migration,
33    ) -> EFResult<()> {
34        if self.is_applied(provider, &migration.id).await? {
35            return Ok(());
36        }
37        self.ensure_history_table(provider).await?;
38        let sql = migration.up_sql.replace("{migration_id}", &migration.id);
39        self.execute_script(provider, &sql).await
40    }
41
42    /// Executes a migration script with the best available atomicity.
43    async fn execute_script(&self, provider: &dyn IDatabaseProvider, sql: &str) -> EFResult<()> {
44        let statements = split_sql_statements(sql, self.dialect);
45        if statements.is_empty() {
46            return Ok(());
47        }
48
49        // MySQL DDL forces an implicit commit per statement; a transaction
50        // cannot make the script atomic. Keep the provider's DDL path (the
51        // prepared-statement protocol rejects DDL).
52        if self.dialect == MigrationDialect::MySql {
53            for statement in statements {
54                provider.execute_migration_command(&statement).await?;
55            }
56            return Ok(());
57        }
58
59        let mut conn = provider.get_connection().await?;
60        conn.begin_transaction().await?;
61        for statement in &statements {
62            if let Err(err) = conn.execute(statement, &[]).await {
63                let _ = conn.rollback_transaction().await;
64                return Err(err);
65            }
66        }
67        if let Err(err) = conn.commit_transaction().await {
68            let _ = conn.rollback_transaction().await;
69            return Err(err);
70        }
71        Ok(())
72    }
73
74    /// Reverts a migration using its Down SQL.
75    pub async fn revert(
76        &self,
77        provider: &dyn IDatabaseProvider,
78        migration: &Migration,
79    ) -> EFResult<()> {
80        if !self.is_applied(provider, &migration.id).await? {
81            return Err(EFError::migration(format!(
82                "migration '{}' is not applied",
83                migration.id
84            )));
85        }
86        let sql = migration.down_sql.replace("{migration_id}", &migration.id);
87        self.execute_script(provider, &sql).await
88    }
89
90    /// Returns migrations already recorded in the database history table.
91    pub async fn get_applied_migrations(
92        &self,
93        provider: &dyn IDatabaseProvider,
94    ) -> EFResult<Vec<MigrationHistoryEntry>> {
95        self.ensure_history_table(provider).await?;
96        let q = |s: &str| self.dialect.quote(s);
97        let table = q(super::history::MIGRATION_HISTORY_TABLE);
98        let sql = format!(
99            "SELECT {}, {} FROM {} ORDER BY {}",
100            q("migration_id"),
101            q("product_version"),
102            table,
103            q("migration_id")
104        );
105        let mut conn = provider.get_connection().await?;
106        let rows = conn.query(&sql, &[]).await?;
107        Ok(rows
108            .into_iter()
109            .map(|row| MigrationHistoryEntry {
110                migration_id: row
111                    .first()
112                    .and_then(|v| String::try_from(v.clone()).ok())
113                    .unwrap_or_default(),
114                product_version: row
115                    .get(1)
116                    .and_then(|v| String::try_from(v.clone()).ok())
117                    .unwrap_or_default(),
118            })
119            .collect())
120    }
121
122    /// Returns whether a migration id is already recorded in history.
123    pub async fn is_applied(
124        &self,
125        provider: &dyn IDatabaseProvider,
126        migration_id: &str,
127    ) -> EFResult<bool> {
128        let applied = self.get_applied_migrations(provider).await?;
129        Ok(applied.iter().any(|e| e.migration_id == migration_id))
130    }
131
132    /// Applies all pending migrations in order, skipping already-applied ids.
133    pub async fn apply_pending(
134        &self,
135        provider: &dyn IDatabaseProvider,
136        migrations: &[Migration],
137    ) -> EFResult<usize> {
138        let applied: std::collections::HashSet<String> = self
139            .get_applied_migrations(provider)
140            .await?
141            .into_iter()
142            .map(|e| e.migration_id)
143            .collect();
144        let mut count = 0usize;
145        for migration in migrations {
146            if applied.contains(&migration.id) {
147                continue;
148            }
149            self.apply(provider, migration).await?;
150            count += 1;
151        }
152        Ok(count)
153    }
154
155    /// Reverts the most recently applied migration from the given list.
156    pub async fn revert_last(
157        &self,
158        provider: &dyn IDatabaseProvider,
159        migrations: &[Migration],
160    ) -> EFResult<Option<String>> {
161        let applied = self.get_applied_migrations(provider).await?;
162        let last = applied.last().map(|e| e.migration_id.clone());
163        let Some(last_id) = last else {
164            return Ok(None);
165        };
166        let migration = migrations.iter().find(|m| m.id == last_id).ok_or_else(|| {
167            EFError::migration(format!(
168                "applied migration '{}' not found in local migration set",
169                last_id
170            ))
171        })?;
172        self.revert(provider, migration).await?;
173        Ok(Some(last_id))
174    }
175
176    /// Reverts all migrations applied strictly after `target` (exclusive),
177    /// leaving the database at `target`'s state. Returns the reverted ids in
178    /// the order they were reverted (most-recent first).
179    ///
180    /// If `target` is `None`, reverts ALL applied migrations.
181    /// Returns an error if `target` is not currently applied.
182    pub async fn revert_to_target(
183        &self,
184        provider: &dyn IDatabaseProvider,
185        migrations: &[Migration],
186        target: Option<&str>,
187    ) -> EFResult<Vec<String>> {
188        let applied = self.get_applied_migrations(provider).await?;
189        let applied_ids: Vec<&str> = applied.iter().map(|e| e.migration_id.as_str()).collect();
190
191        let to_revert: Vec<String> = match target {
192            Some(t) => {
193                let idx = applied_ids.iter().position(|id| *id == t).ok_or_else(|| {
194                    EFError::migration(format!("target migration '{t}' is not applied"))
195                })?;
196                applied_ids[idx + 1..]
197                    .iter()
198                    .map(|s| s.to_string())
199                    .collect()
200            }
201            None => applied_ids.iter().map(|s| s.to_string()).collect(),
202        };
203
204        let mut reverted = Vec::with_capacity(to_revert.len());
205        for id in to_revert.iter().rev() {
206            let migration = migrations.iter().find(|m| m.id == *id).ok_or_else(|| {
207                EFError::migration(format!(
208                    "applied migration '{id}' not found in local migration set"
209                ))
210            })?;
211            self.revert(provider, migration).await?;
212            reverted.push(id.clone());
213        }
214        Ok(reverted)
215    }
216
217    /// Generates a combined SQL script transitioning the database from the
218    /// `from` migration state to the `to` migration state.
219    ///
220    /// Semantics (mirroring `dotnet ef migrations script`):
221    /// - `from` is exclusive: the DB is assumed to already be at `from`.
222    /// - `to` is inclusive: the script brings the DB up to and including `to`.
223    /// - `from = None`: start from an empty database (before any migration).
224    /// - `to = None`: end at the latest migration.
225    /// - When `from` precedes `to`: emits Up SQL for migrations in `(from, to]`.
226    /// - When `from` follows `to`: emits Down SQL for migrations in `(to, from]`.
227    /// - When `from == to`: emits a no-op comment.
228    pub fn generate_script(
229        migrations: &[Migration],
230        from: Option<&str>,
231        to: Option<&str>,
232    ) -> EFResult<String> {
233        // Resolve from/to to indices. from=None means "before first migration" (-1).
234        // to=None means "after last migration" (migrations.len() - 1).
235        let from_idx: i64 = match from {
236            Some(f) => migrations.iter().position(|m| m.id == f).ok_or_else(|| {
237                EFError::migration(format!("from migration '{f}' not found in local set"))
238            })? as i64,
239            None => -1,
240        };
241        let to_idx: i64 = match to {
242            Some(t) => migrations.iter().position(|m| m.id == t).ok_or_else(|| {
243                EFError::migration(format!("to migration '{t}' not found in local set"))
244            })? as i64,
245            None => migrations.len() as i64 - 1,
246        };
247
248        let mut sql = String::new();
249        if from_idx < to_idx {
250            // Forward: up scripts for migrations in (from_idx, to_idx].
251            sql.push_str("-- Migration script (forward)\n\n");
252            let start = (from_idx + 1) as usize;
253            let end = (to_idx + 1) as usize; // exclusive
254            for m in &migrations[start..end] {
255                sql.push_str(&format!("-- Up: {}\n", m.id));
256                sql.push_str(&m.up_sql.replace("{migration_id}", &m.id));
257                sql.push('\n');
258            }
259        } else if from_idx > to_idx {
260            // Reverse: down scripts for migrations in (to_idx, from_idx].
261            sql.push_str("-- Migration script (reverse)\n\n");
262            let start = to_idx as usize + 1; // inclusive
263            let end = from_idx as usize + 1; // exclusive
264            for m in migrations[start..end].iter().rev() {
265                sql.push_str(&format!("-- Down: {}\n", m.id));
266                sql.push_str(&m.down_sql.replace("{migration_id}", &m.id));
267                sql.push('\n');
268            }
269        } else {
270            sql.push_str("-- Nothing to do (from == to)\n");
271        }
272        Ok(sql)
273    }
274
275    /// Generates and applies the initial schema for the given entity types.
276    /// Corresponds to EF Core `Database.EnsureCreated()`.
277    /// Does not use the migrations history table (idempotent DDL only).
278    pub async fn ensure_created(
279        &self,
280        provider: &dyn IDatabaseProvider,
281        entity_types: &[EntityTypeMeta],
282    ) -> EFResult<()> {
283        let snapshot = self.create_snapshot("__ensure_created__", entity_types);
284        let changes = self.initial_create(&snapshot);
285        let sql = self.generate_ddl_sql(&changes);
286        self.execute_script(provider, &sql).await
287    }
288
289    /// Drops all tables for the given entity types.
290    /// Corresponds to EF Core `Database.EnsureDeleted()`.
291    pub async fn ensure_deleted(
292        &self,
293        provider: &dyn IDatabaseProvider,
294        entity_types: &[EntityTypeMeta],
295    ) -> EFResult<()> {
296        let q = |s: &str| self.dialect.quote(s);
297        for meta in entity_types.iter().filter(|m| !m.is_view) {
298            let sql = format!("DROP TABLE IF EXISTS {};", q(meta.table_name.as_ref()));
299            provider.execute_migration_command(&sql).await?;
300        }
301        Ok(())
302    }
303
304    /// Inserts seed data configured via `ModelBuilder::has_data`.
305    pub async fn apply_seed_data(
306        &self,
307        provider: &dyn IDatabaseProvider,
308        meta: &EntityTypeMeta,
309        rows: &[EntitySnapshot],
310    ) -> EFResult<()> {
311        if rows.is_empty() {
312            return Ok(());
313        }
314
315        let gen = provider.sql_generator();
316        let scalar_props: Vec<_> = meta.writable_scalar_properties().collect();
317        if scalar_props.is_empty() {
318            return Ok(());
319        }
320
321        let col_names: Vec<&str> = scalar_props
322            .iter()
323            .map(|p| p.column_name.as_ref())
324            .collect();
325
326        let mut conn = provider.get_connection().await?;
327        for row in rows {
328            let params: Vec<crate::provider::DbValue> = scalar_props
329                .iter()
330                .map(|p| {
331                    row.get(p.field_name.as_ref())
332                        .cloned()
333                        .unwrap_or(crate::provider::DbValue::Null)
334                })
335                .collect();
336
337            let sql = seed_insert_sql(self.dialect, meta.table_name.as_ref(), &col_names, gen);
338            conn.execute(&sql, &params).await?;
339        }
340        Ok(())
341    }
342}
rust_ef/migration/engine_ops.rs

rust_ef/migration/
engine_ops.rs
