rws_server/model/
migration.rs

1//! Async migration runner.
2//!
3//! Reads `*.sql` files from a directory in lexicographic order. Creates
4//! `_schema_migrations(version VARCHAR(255) PRIMARY KEY, applied_at VARCHAR(64))`
5//! if absent (`VARCHAR`: MySQL can't key a `TEXT` column; tables made as
6//! `TEXT` by earlier releases keep working).
7//! Runs each unapplied file inside a transaction.
8//!
9//! Rollback: an up file `NNNN_name.sql` may have a companion down file
10//! `NNNN_name.down.sql` in the same directory. [`rollback_last`] and
11//! [`rollback`] undo the most recently applied migration(s) — determined by
12//! version order, the same invariant `run_migrations` relies on to apply
13//! files in order — by running the down file's SQL and deleting the row
14//! from `_schema_migrations`, both inside one transaction.
15
16use super::pool::DbPool;
17use super::{DbError, Value};
18
19// ── MigrationStatus ───────────────────────────────────────────────────────────
20
21/// Status of a single migration file.
22#[derive(Debug, Clone)]
23pub struct MigrationStatus {
24    /// The filename (without directory prefix), used as the version key.
25    pub version: String,
26    /// `true` if this migration has been applied.
27    pub applied: bool,
28    /// `true` if a companion `<version-stem>.down.sql` file exists, meaning
29    /// this migration can be undone with [`rollback_last`] / [`rollback`].
30    pub has_down: bool,
31}
32
33// ── Helpers ───────────────────────────────────────────────────────────────────
34
35const CREATE_TABLE_SQL: &str =
36    "CREATE TABLE IF NOT EXISTS _schema_migrations \
37     (version VARCHAR(255) PRIMARY KEY, applied_at VARCHAR(64) NOT NULL)";
38
39async fn ensure_migrations_table(pool: &DbPool) -> Result<(), DbError> {
40    pool.execute(CREATE_TABLE_SQL, &[]).await?;
41    Ok(())
42}
43
44async fn applied_versions(pool: &DbPool) -> Result<Vec<String>, DbError> {
45    let rows = pool.query_rows("SELECT version FROM _schema_migrations", &[]).await?;
46    rows.into_iter().map(|r| r.get::<String>("version")).collect()
47}
48
49/// The version of the most recently applied migration, or `None` if none
50/// are applied. "Most recent" means highest version string — the same
51/// lexicographic order `run_migrations` applies files in.
52async fn last_applied_version(pool: &DbPool) -> Result<Option<String>, DbError> {
53    let rows = pool
54        .query_rows("SELECT version FROM _schema_migrations ORDER BY version DESC LIMIT 1", &[])
55        .await?;
56    rows.into_iter().next().map(|r| r.get::<String>("version")).transpose()
57}
58
59/// Up migration files only — excludes companion `*.down.sql` files, which
60/// share the `.sql` extension but are never applied as an "up" step.
61fn read_sql_files(dir: &str) -> Result<Vec<(String, String)>, DbError> {
62    let mut files: Vec<(String, String)> = Vec::new();
63    let entries = std::fs::read_dir(dir)
64        .map_err(|e| DbError::new(format!("cannot read migrations directory '{}': {}", dir, e)))?;
65
66    for entry in entries {
67        let entry = entry.map_err(|e| DbError::new(e.to_string()))?;
68        let path = entry.path();
69        let filename = path.file_name().and_then(|n| n.to_str()).unwrap_or("").to_owned();
70        if filename.ends_with(".sql") && !filename.ends_with(".down.sql") {
71            let contents = std::fs::read_to_string(&path)
72                .map_err(|e| DbError::new(format!("cannot read '{}': {}", path.display(), e)))?;
73            files.push((filename, contents));
74        }
75    }
76    files.sort_by(|a, b| a.0.cmp(&b.0));
77    Ok(files)
78}
79
80/// The companion down filename for up migration `version` (e.g.
81/// `0001_x.sql` → `0001_x.down.sql`).
82fn down_filename(version: &str) -> String {
83    match version.strip_suffix(".sql") {
84        Some(stem) => format!("{}.down.sql", stem),
85        None => format!("{}.down.sql", version),
86    }
87}
88
89fn down_file_path(dir: &str, version: &str) -> String {
90    format!("{}/{}", dir.trim_end_matches('/'), down_filename(version))
91}
92
93fn down_file_exists(dir: &str, version: &str) -> bool {
94    std::path::Path::new(&down_file_path(dir, version)).exists()
95}
96
97/// Read the down SQL for `version`, or `None` if no companion down file exists.
98fn read_down_sql(dir: &str, version: &str) -> Result<Option<String>, DbError> {
99    let path = down_file_path(dir, version);
100    match std::fs::read_to_string(&path) {
101        Ok(s) => Ok(Some(s)),
102        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
103        Err(e) => Err(DbError::new(format!("cannot read '{}': {}", path, e))),
104    }
105}
106
107fn now_timestamp() -> String {
108    use std::time::{SystemTime, UNIX_EPOCH};
109    let secs = SystemTime::now().duration_since(UNIX_EPOCH).map(|d| d.as_secs()).unwrap_or(0);
110    let (y, mo, d) = days_to_ymd(secs / 86400);
111    let rem = secs % 86400;
112    format!("{:04}-{:02}-{:02}T{:02}:{:02}:{:02}Z", y, mo, d, rem / 3600, (rem % 3600) / 60, rem % 60)
113}
114
115fn days_to_ymd(mut days: u64) -> (u64, u64, u64) {
116    let mut year = 1970u64;
117    loop {
118        let diy = if is_leap(year) { 366 } else { 365 };
119        if days < diy { break; }
120        days -= diy;
121        year += 1;
122    }
123    let months = [31u64, if is_leap(year) { 29 } else { 28 }, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31];
124    let mut month = 1u64;
125    for &dim in &months {
126        if days < dim { break; }
127        days -= dim;
128        month += 1;
129    }
130    (year, month, days + 1)
131}
132
133fn is_leap(y: u64) -> bool {
134    (y % 4 == 0 && y % 100 != 0) || (y % 400 == 0)
135}
136
137// ── Public API ────────────────────────────────────────────────────────────────
138
139/// Run pending migrations from SQL files in `dir`. Called by `DbPool::migrate`.
140pub async fn run_migrations(pool: &DbPool, dir: &str) -> Result<(), DbError> {
141    ensure_migrations_table(pool).await?;
142    let applied = applied_versions(pool).await?;
143    let files = read_sql_files(dir)?;
144
145    for (filename, sql) in files {
146        if applied.contains(&filename) {
147            continue;
148        }
149
150        // Execute each statement in the file separately (split on `;`).
151        let statements: Vec<&str> = sql.split(';').map(str::trim).filter(|s| !s.is_empty()).collect();
152        let version = filename.clone();
153        let ts = now_timestamp();
154        let insert_sql = format!(
155            "INSERT INTO _schema_migrations (version, applied_at) VALUES ({}, {})",
156            super::repository::placeholder(pool.backend(), 1),
157            super::repository::placeholder(pool.backend(), 2),
158        );
159
160        let mut tx = pool.begin().await?;
161        for stmt in &statements {
162            if let Err(e) = tx.execute(stmt, &[]).await {
163                let _ = tx.rollback().await;
164                return Err(DbError::new(format!("migration '{}' failed: {}", filename, e)));
165            }
166        }
167        if let Err(e) = tx.execute(&insert_sql, &[Value::Text(version), Value::Text(ts)]).await {
168            let _ = tx.rollback().await;
169            return Err(e);
170        }
171        tx.commit().await?;
172    }
173    Ok(())
174}
175
176/// Return status (applied / pending) for each SQL file in `dir`. Called by `DbPool::migration_status`.
177pub async fn migration_status(pool: &DbPool, dir: &str) -> Result<Vec<MigrationStatus>, DbError> {
178    ensure_migrations_table(pool).await?;
179    let applied = applied_versions(pool).await?;
180    let files = read_sql_files(dir)?;
181    Ok(files
182        .into_iter()
183        .map(|(filename, _)| {
184            let is_applied = applied.contains(&filename);
185            let has_down = down_file_exists(dir, &filename);
186            MigrationStatus { version: filename, applied: is_applied, has_down }
187        })
188        .collect())
189}
190
191/// Roll back the most recently applied migration in `dir`. Called by
192/// `DbPool::rollback_last`.
193///
194/// Requires a companion `<version-stem>.down.sql` file next to the up
195/// migration it undoes; returns `Err` if that file is missing. Runs the
196/// down file's statements and deletes the `_schema_migrations` row inside
197/// one transaction — rolled back together on failure, same as `run_migrations`.
198///
199/// Returns `Ok(None)` if no migrations are currently applied.
200pub async fn rollback_last(pool: &DbPool, dir: &str) -> Result<Option<String>, DbError> {
201    ensure_migrations_table(pool).await?;
202    let version = match last_applied_version(pool).await? {
203        Some(v) => v,
204        None => return Ok(None),
205    };
206
207    let down_sql = read_down_sql(dir, &version)?.ok_or_else(|| {
208        DbError::new(format!(
209            "no down migration found for '{}': expected '{}'",
210            version,
211            down_filename(&version),
212        ))
213    })?;
214
215    let statements: Vec<&str> = down_sql.split(';').map(str::trim).filter(|s| !s.is_empty()).collect();
216    let delete_sql = format!(
217        "DELETE FROM _schema_migrations WHERE version = {}",
218        super::repository::placeholder(pool.backend(), 1),
219    );
220
221    let mut tx = pool.begin().await?;
222    for stmt in &statements {
223        if let Err(e) = tx.execute(stmt, &[]).await {
224            let _ = tx.rollback().await;
225            return Err(DbError::new(format!("rollback of '{}' failed: {}", version, e)));
226        }
227    }
228    if let Err(e) = tx.execute(&delete_sql, &[Value::Text(version.clone())]).await {
229        let _ = tx.rollback().await;
230        return Err(e);
231    }
232    tx.commit().await?;
233    Ok(Some(version))
234}
235
236/// Roll back the last `n` applied migrations, most recently applied first.
237/// Called by `DbPool::rollback`.
238///
239/// Stops early (returning fewer than `n` versions) once nothing is left to
240/// roll back. Stops with `Err` at the first missing down file or SQL error;
241/// any migrations already rolled back in earlier iterations stay rolled
242/// back — each step commits its own transaction independently.
243pub async fn rollback(pool: &DbPool, dir: &str, n: usize) -> Result<Vec<String>, DbError> {
244    let mut rolled_back = Vec::with_capacity(n);
245    for _ in 0..n {
246        match rollback_last(pool, dir).await? {
247            Some(version) => rolled_back.push(version),
248            None => break,
249        }
250    }
251    Ok(rolled_back)
252}
253
254/// Run every `*.sql` file in `dir`, in order, each in one transaction.
255/// Called by `DbPool::seed`.
256pub async fn run_seeds(pool: &DbPool, dir: &str) -> Result<Vec<String>, DbError> {
257    let files = read_sql_files(dir)
258        .map_err(|e| DbError::new(e.0.replace("migrations directory", "seeds directory")))?;
259    let mut ran = Vec::with_capacity(files.len());
260    for (filename, sql) in files {
261        let mut tx = pool.begin().await?;
262        for stmt in sql.split(';').map(str::trim).filter(|s| !s.is_empty()) {
263            if let Err(e) = tx.execute(stmt, &[]).await {
264                let _ = tx.rollback().await;
265                return Err(DbError::new(format!("seed '{}' failed: {}", filename, e)));
266            }
267        }
268        tx.commit().await?;
269        ran.push(filename);
270    }
271    Ok(ran)
272}
rws_server/model/migration.rs

rws_server/model/
migration.rs
