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

rust_ef/migration/
engine_ops.rs
