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

rws_server/model/
migration.rs
