rust_ef/migration/
engine_ops.rs

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

rust_ef/migration/
engine_ops.rs
