orbit_orm/
migrate.rs

1//! Migrations are SQL folders that change the schema. Each folder is named `<timestamp>_<name>`
2//! and holds `up.sql` and `down.sql`. The applied ones are recorded in Diesel's
3//! `__diesel_schema_migrations` table with a batch number, and a rollback reverts the last batch.
4//! The table has the same name and columns on every database.
5
6#[cfg(feature = "sqlite")]
7use std::collections::BTreeSet;
8use std::collections::{HashMap, HashSet};
9use std::future::Future;
10use std::path::Path;
11#[cfg(feature = "sqlite")]
12use std::sync::{Mutex, PoisonError};
13use std::time::Duration;
14#[cfg(any(feature = "postgres", feature = "sqlite"))]
15use std::time::Instant;
16
17use crate::backend;
18use crate::database::Pool;
19use crate::schema::Statement;
20use crate::transaction::{Target, Transaction};
21use crate::{Database, Driver, Error, Value};
22
23/// One migration: its name, and what applies it (`up`) and reverts it (`down`). Either fixed SQL
24/// ([`Migration::new`], files) or schema statements ([`Migration::schema`]).
25#[derive(Debug, Clone)]
26pub struct Migration {
27    /// The folder name, such as `20261108000000_create_users`.
28    pub name: String,
29    /// The contents of `up.sql`. Empty for a [`schema`](Self::schema) migration, which carries
30    /// statements instead.
31    pub up: String,
32    /// The contents of `down.sql`. Empty for a [`schema`](Self::schema) migration.
33    pub down: String,
34    /// The statements of a schema migration, rendered when the migrator runs it.
35    schema: Option<SchemaBody>,
36}
37
38#[derive(Debug, Clone)]
39struct SchemaBody {
40    up: Vec<Statement>,
41    down: Vec<Statement>,
42}
43
44impl Migration {
45    /// A migration from its name and SQL, for code that builds migrations itself.
46    pub fn new(name: impl Into<String>, up: impl Into<String>, down: impl Into<String>) -> Self {
47        Self {
48            name: name.into(),
49            up: up.into(),
50            down: down.into(),
51            schema: None,
52        }
53    }
54
55    /// A migration written with the schema builder. The statements are rendered for the connected
56    /// database when the migrator runs or rolls it back, so one migration serves every database.
57    /// A statement that cannot be rendered fails the migration with that error, before any of its
58    /// SQL runs. The migration keeps the same transaction and batch behaviour as an SQL one.
59    ///
60    /// A [`schema::statement`](crate::schema::statement) runs its SQL as written. A statement
61    /// limited with [`Statement::only`] to another database family renders nothing, so it does
62    /// not run there.
63    pub fn schema(name: impl Into<String>, up: Vec<Statement>, down: Vec<Statement>) -> Self {
64        Self {
65            name: name.into(),
66            up: String::new(),
67            down: String::new(),
68            schema: Some(SchemaBody { up, down }),
69        }
70    }
71
72    /// What `run` executes, for `driver`: the SQL scripts, in order.
73    fn up_scripts(&self, driver: Driver) -> Result<Vec<String>, Error> {
74        match &self.schema {
75            Some(body) => render(&body.up, driver),
76            None => Ok(vec![self.up.clone()]),
77        }
78    }
79
80    /// What `rollback` executes, for `driver`.
81    fn down_scripts(&self, driver: Driver) -> Result<Vec<String>, Error> {
82        match &self.schema {
83            Some(body) => render(&body.down, driver),
84            None => Ok(vec![self.down.clone()]),
85        }
86    }
87}
88
89fn render(statements: &[Statement], driver: Driver) -> Result<Vec<String>, Error> {
90    let mut sql = Vec::new();
91    for statement in statements {
92        sql.extend(statement.try_to_sql(driver)?);
93    }
94    Ok(sql)
95}
96
97/// Whether a migration has run. `Applied` carries the batch it ran in.
98#[derive(Debug, Clone, Copy, PartialEq, Eq)]
99pub enum State {
100    Applied { batch: i64 },
101    Pending,
102}
103
104/// One line of [`Migrator::status`].
105#[derive(Debug, Clone, PartialEq, Eq)]
106pub struct Status {
107    pub name: String,
108    pub state: State,
109}
110
111/// The migrations of one app, sorted by name. Build it with [`Migrator::from_dir`] or
112/// [`migrations!`](crate::migrations).
113#[derive(Debug, Clone)]
114pub struct Migrator {
115    migrations: Vec<Migration>,
116    lock_timeout: Duration,
117}
118
119/// The lock that serialises `run` and `rollback` across processes is named after the database,
120/// so two databases on one server do not wait for each other.
121const LOCK_PREFIX: &str = "orbit_migrations";
122
123/// How long `run` and `rollback` wait for another process's lock before giving up.
124const DEFAULT_LOCK_TIMEOUT: Duration = Duration::from_secs(10);
125
126/// How often a waiting lock is tried again, where the database has no blocking lock call.
127#[cfg(any(feature = "postgres", feature = "sqlite"))]
128const LOCK_POLL: Duration = Duration::from_millis(50);
129
130/// The table that records the applied migrations, on every database.
131const TABLE: &str = "__diesel_schema_migrations";
132
133/// The migrations that are held in this process, by lock name. Used by SQLite, which has no
134/// named lock. See `sqlite_lock`.
135#[cfg(feature = "sqlite")]
136static SQLITE_HELD: Mutex<BTreeSet<String>> = Mutex::new(BTreeSet::new());
137
138impl Migrator {
139    /// Sorts the migrations by name. A name must start with a 14-digit timestamp and an
140    /// underscore, as `20261108000000_create_users` does. Two migrations with the same name, or
141    /// the same timestamp, are an error.
142    pub fn new(mut migrations: Vec<Migration>) -> Result<Self, Error> {
143        for migration in &migrations {
144            if !has_timestamp(&migration.name) {
145                return Err(Error::InvalidMigrationName {
146                    name: migration.name.clone(),
147                });
148            }
149        }
150        migrations.sort_by(|a, b| a.name.cmp(&b.name));
151        for pair in migrations.windows(2) {
152            if pair[0].name == pair[1].name {
153                return Err(Error::DuplicateMigration {
154                    name: pair[0].name.clone(),
155                });
156            }
157            if timestamp(&pair[0].name) == timestamp(&pair[1].name) {
158                return Err(Error::DuplicateTimestamp {
159                    first: pair[0].name.clone(),
160                    second: pair[1].name.clone(),
161                });
162            }
163        }
164        Ok(Self {
165            migrations,
166            lock_timeout: DEFAULT_LOCK_TIMEOUT,
167        })
168    }
169
170    /// Sets how long `run` and `rollback` wait for another process to finish its migrations.
171    pub fn with_lock_timeout(mut self, timeout: Duration) -> Self {
172        self.lock_timeout = timeout;
173        self
174    }
175
176    /// The migrations in name order.
177    pub fn migrations(&self) -> &[Migration] {
178        &self.migrations
179    }
180
181    /// Reads every folder in `dir` as a migration, such as `migrations/20261108000000_create_users`.
182    /// Each folder needs `up.sql` and `down.sql`. Hidden entries are skipped.
183    pub fn from_dir(dir: impl AsRef<Path>) -> Result<Self, Error> {
184        let dir = dir.as_ref();
185        let entries = std::fs::read_dir(dir).map_err(|source| Error::Read {
186            path: dir.to_path_buf(),
187            source,
188        })?;
189        let mut migrations = Vec::new();
190        for entry in entries {
191            let entry = entry.map_err(|source| Error::Read {
192                path: dir.to_path_buf(),
193                source,
194            })?;
195            let name = entry.file_name().to_string_lossy().into_owned();
196            let path = entry.path();
197            if name.starts_with('.') || !path.is_dir() {
198                continue;
199            }
200            let up = read_file(&path.join("up.sql"))?;
201            let down = read_file(&path.join("down.sql"))?;
202            migrations.push(Migration::new(name, up, down));
203        }
204        Self::new(migrations)
205    }
206
207    /// Applies every pending migration, in name order, as one new batch. Each migration runs in
208    /// its own transaction. The run stops at the first failure, and the error names it. The
209    /// migrations before it stay applied in this batch.
210    ///
211    /// MySQL and SQLite do not roll back DDL, and PostgreSQL does. On MySQL a failing migration can
212    /// leave the DDL it ran before the failure in place; only its DML is rolled back. Returns the
213    /// names that were applied.
214    ///
215    /// Only one process runs migrations at a time. A second process waits up to the lock timeout,
216    /// then fails with [`Error::MigrationsLocked`]. The lock is held differently on each database;
217    /// see `docs/laravel-differences.md`.
218    pub async fn run(&self, db: &Database) -> Result<Vec<String>, Error> {
219        self.with_lock(db, self.run_unlocked(db)).await
220    }
221
222    async fn run_unlocked(&self, db: &Database) -> Result<Vec<String>, Error> {
223        ensure_table(db).await?;
224        let applied = applied(db).await?;
225        let pending: Vec<&Migration> = self
226            .migrations
227            .iter()
228            .filter(|m| !applied.contains_key(timestamp(&m.name)))
229            .collect();
230        if pending.is_empty() {
231            return Ok(Vec::new());
232        }
233
234        let max: Option<i32> = backend::scalar(
235            Target::Pool(&db.pool),
236            &format!("SELECT MAX(batch) FROM {TABLE}"),
237            &[],
238        )
239        .await?;
240        let batch = i64::from(max.unwrap_or(0)) + 1;
241        let mut names = Vec::with_capacity(pending.len());
242        for migration in pending {
243            let scripts = migration.up_scripts(db.driver())?;
244            let mut tx = Transaction::begin(&db.pool).await?;
245            for script in &scripts {
246                backend::run_script(Target::Tx(&mut tx.inner), script)
247                    .await
248                    .map_err(|source| Error::MigrationFailed {
249                        name: migration.name.clone(),
250                        source,
251                    })?;
252            }
253            let (sql, values) = prepared(
254                db.driver(),
255                &format!("INSERT INTO {TABLE} (version, batch) VALUES (?, ?)"),
256                vec![
257                    Value::Text(timestamp(&migration.name).to_owned()),
258                    Value::Int(batch),
259                ],
260            );
261            backend::execute(Target::Tx(&mut tx.inner), &sql, &values)
262                .await
263                .map_err(|source| Error::MigrationFailed {
264                    name: migration.name.clone(),
265                    source,
266                })?;
267            tx.commit().await?;
268            names.push(migration.name.clone());
269        }
270        Ok(names)
271    }
272
273    /// Lists every migration in this migrator as applied (with its batch) or pending.
274    /// Applied migrations without files here are not listed.
275    pub async fn status(&self, db: &Database) -> Result<Vec<Status>, Error> {
276        ensure_table(db).await?;
277        let applied = applied(db).await?;
278        Ok(self
279            .migrations
280            .iter()
281            .map(|m| Status {
282                name: m.name.clone(),
283                state: match applied.get(timestamp(&m.name)) {
284                    Some(&batch) => State::Applied { batch },
285                    None => State::Pending,
286                },
287            })
288            .collect())
289    }
290
291    /// Reverts the last batch, running each `down.sql` in reverse name order, each in its own
292    /// transaction. It checks that every migration in the batch has files here before it reverts
293    /// anything. Returns the names that were reverted, or nothing when no migration has run.
294    ///
295    /// Batch 0 holds the history that Diesel recorded, so a rollback that reaches batch 0 reverts
296    /// those migrations and removes their rows from Diesel's table too.
297    ///
298    /// As with [`run`](Self::run), DDL in a `down.sql` is only rolled back on PostgreSQL.
299    pub async fn rollback(&self, db: &Database) -> Result<Vec<String>, Error> {
300        self.with_lock(db, self.rollback_unlocked(db)).await
301    }
302
303    async fn rollback_unlocked(&self, db: &Database) -> Result<Vec<String>, Error> {
304        ensure_table(db).await?;
305        let last: Option<i32> = backend::scalar(
306            Target::Pool(&db.pool),
307            &format!("SELECT MAX(batch) FROM {TABLE}"),
308            &[],
309        )
310        .await?;
311        let Some(batch) = last else {
312            return Ok(Vec::new());
313        };
314        let (sql, values) = prepared(
315            db.driver(),
316            &format!("SELECT version FROM {TABLE} WHERE batch = ? ORDER BY version DESC"),
317            vec![Value::Int(i64::from(batch))],
318        );
319        let versions: Vec<(String,)> =
320            backend::fetch_all(Target::Pool(&db.pool), &sql, &values).await?;
321
322        let mut to_revert = Vec::with_capacity(versions.len());
323        for (version,) in &versions {
324            let migration = self
325                .migrations
326                .iter()
327                .find(|m| timestamp(&m.name) == version.as_str())
328                .ok_or_else(|| Error::UnknownMigration {
329                    name: version.clone(),
330                })?;
331            to_revert.push(migration);
332        }
333        let names = to_revert.iter().map(|m| m.name.clone()).collect();
334        for migration in to_revert {
335            let scripts = migration.down_scripts(db.driver())?;
336            let mut tx = Transaction::begin(&db.pool).await?;
337            for script in &scripts {
338                backend::run_script(Target::Tx(&mut tx.inner), script)
339                    .await
340                    .map_err(|source| Error::RollbackFailed {
341                        name: migration.name.clone(),
342                        source,
343                    })?;
344            }
345            let (sql, values) = prepared(
346                db.driver(),
347                &format!("DELETE FROM {TABLE} WHERE version = ?"),
348                vec![Value::Text(timestamp(&migration.name).to_owned())],
349            );
350            backend::execute(Target::Tx(&mut tx.inner), &sql, &values)
351                .await
352                .map_err(|source| Error::RollbackFailed {
353                    name: migration.name.clone(),
354                    source,
355                })?;
356            tx.commit().await?;
357        }
358        Ok(names)
359    }
360}
361
362impl Migrator {
363    /// Brings Diesel's history into this migrator's table. Orbit and Diesel share
364    /// `__diesel_schema_migrations`, so the rows Diesel wrote are already the applied history, in
365    /// batch 0. This call does what is still needed for them: it adds the `batch` column to a
366    /// table that has only Diesel's columns. No `up.sql` runs, and no row is added or changed.
367    /// It returns nothing, because there is nothing new to record. It is safe to call again.
368    #[deprecated(
369        since = "0.1.1",
370        note = "not needed any more: run, status and rollback upgrade the migrations table \
371                themselves. This returns an empty list and will be removed in a later release"
372    )]
373    pub async fn import_diesel_history(&self, db: &Database) -> Result<Vec<String>, Error> {
374        self.with_lock(db, import_unlocked(db)).await
375    }
376
377    /// The versions in the migrations table that have no migration in this migrator, in order.
378    /// They include versions that Orbit recorded when a folder has since been removed. Empty when
379    /// the table does not exist. Only reads the table.
380    pub async fn unmatched_diesel_versions(&self, db: &Database) -> Result<Vec<String>, Error> {
381        let Some(versions) = recorded_versions(db).await? else {
382            return Ok(Vec::new());
383        };
384        let mut unmatched: Vec<String> = versions
385            .into_iter()
386            .filter(|version| {
387                !self
388                    .migrations
389                    .iter()
390                    .any(|m| timestamp(&m.name) == version.as_str())
391            })
392            .collect();
393        unmatched.sort();
394        Ok(unmatched)
395    }
396}
397
398/// Embeds migrations in the binary at build time. `dir` is relative to the crate root, and each
399/// name is a folder in it with `up.sql` and `down.sql`. The names are listed by hand, and a
400/// missing file is a build error. Returns a `Result<Migrator, Error>`.
401///
402/// ```ignore
403/// let migrator = orbit_orm::migrations!("migrations", "20261108000000_create_users")?;
404/// ```
405#[macro_export]
406macro_rules! migrations {
407    ($dir:literal, $($name:literal),* $(,)?) => {
408        $crate::Migrator::new(vec![
409            $(
410                $crate::Migration::new(
411                    $name,
412                    include_str!(concat!(env!("CARGO_MANIFEST_DIR"), "/", $dir, "/", $name, "/up.sql")),
413                    include_str!(concat!(env!("CARGO_MANIFEST_DIR"), "/", $dir, "/", $name, "/down.sql")),
414                ),
415            )*
416        ])
417    };
418}
419
420impl Migrator {
421    /// Runs `body` while holding the migrations lock for the database. See each driver's lock
422    /// function for how it is held.
423    async fn with_lock<T>(
424        &self,
425        db: &Database,
426        body: impl Future<Output = Result<T, Error>>,
427    ) -> Result<T, Error> {
428        match &db.pool {
429            #[cfg(feature = "mysql")]
430            Pool::MySql(pool) => mysql_lock(pool, self.lock_timeout, body).await,
431            #[cfg(feature = "postgres")]
432            Pool::Postgres(pool) => postgres_lock(pool, self.lock_timeout, body).await,
433            #[cfg(feature = "sqlite")]
434            Pool::Sqlite(pool) => sqlite_lock(pool, self.lock_timeout, body).await,
435        }
436    }
437}
438
439/// MySQL: a named lock from `GET_LOCK`, held by one pooled connection, so it lasts for the whole
440/// run and is released when the run ends.
441#[cfg(feature = "mysql")]
442async fn mysql_lock<T>(
443    pool: &sqlx::MySqlPool,
444    timeout: Duration,
445    body: impl Future<Output = Result<T, Error>>,
446) -> Result<T, Error> {
447    let name = format!(
448        "{LOCK_PREFIX}:{}",
449        pool.connect_options().get_database().unwrap_or("")
450    );
451    let mut conn = pool.acquire().await?;
452    let got: Option<i64> = sqlx::query_scalar("SELECT GET_LOCK(?, ?)")
453        .bind(&name)
454        .bind(timeout.as_secs() as i64)
455        .fetch_one(&mut *conn)
456        .await?;
457    if got != Some(1) {
458        return Err(Error::MigrationsLocked {
459            seconds: timeout.as_secs(),
460        });
461    }
462
463    let result = body.await;
464    let released = sqlx::query("SELECT RELEASE_LOCK(?)")
465        .bind(&name)
466        .execute(&mut *conn)
467        .await;
468    if released.is_err() {
469        // Closing the connection drops the lock on the server, so it cannot stay in the pool.
470        drop(conn.detach());
471    }
472    let value = result?;
473    released?;
474    Ok(value)
475}
476
477/// PostgreSQL: a session advisory lock, keyed by a stable hash of the database name. It is held by
478/// one pooled connection, and waiting is a `pg_try_advisory_lock` retried until the timeout.
479#[cfg(feature = "postgres")]
480async fn postgres_lock<T>(
481    pool: &sqlx::PgPool,
482    timeout: Duration,
483    body: impl Future<Output = Result<T, Error>>,
484) -> Result<T, Error> {
485    let name = format!(
486        "{LOCK_PREFIX}:{}",
487        pool.connect_options().get_database().unwrap_or("")
488    );
489    let key = advisory_key(&name);
490    let mut conn = pool.acquire().await?;
491    let deadline = Instant::now() + timeout;
492    loop {
493        let got: bool = sqlx::query_scalar("SELECT pg_try_advisory_lock($1)")
494            .bind(key)
495            .fetch_one(&mut *conn)
496            .await?;
497        if got {
498            break;
499        }
500        if Instant::now() >= deadline {
501            return Err(Error::MigrationsLocked {
502                seconds: timeout.as_secs(),
503            });
504        }
505        tokio::time::sleep(LOCK_POLL).await;
506    }
507
508    let result = body.await;
509    let released = sqlx::query_scalar::<_, bool>("SELECT pg_advisory_unlock($1)")
510        .bind(key)
511        .fetch_one(&mut *conn)
512        .await;
513    if released.is_err() {
514        // Closing the connection ends the session, which drops its advisory locks.
515        drop(conn.detach());
516    }
517    let value = result?;
518    released?;
519    Ok(value)
520}
521
522/// SQLite has no named lock, so this lock is held by the process: the name is kept in a set
523/// while the run lasts, and a second run in the same process waits for it to leave the set.
524/// Two processes are not held apart by this lock. SQLite serialises their writes, so they can
525/// both reach the migrations table, and one of them may fail on a duplicate version. See
526/// `docs/laravel-differences.md`.
527#[cfg(feature = "sqlite")]
528async fn sqlite_lock<T>(
529    pool: &sqlx::SqlitePool,
530    timeout: Duration,
531    body: impl Future<Output = Result<T, Error>>,
532) -> Result<T, Error> {
533    let name = format!(
534        "{LOCK_PREFIX}:{}",
535        pool.connect_options().get_filename().display()
536    );
537    let deadline = Instant::now() + timeout;
538    loop {
539        let taken = SQLITE_HELD
540            .lock()
541            .unwrap_or_else(PoisonError::into_inner)
542            .insert(name.clone());
543        if taken {
544            break;
545        }
546        if Instant::now() >= deadline {
547            return Err(Error::MigrationsLocked {
548                seconds: timeout.as_secs(),
549            });
550        }
551        tokio::time::sleep(LOCK_POLL).await;
552    }
553
554    let result = body.await;
555    SQLITE_HELD
556        .lock()
557        .unwrap_or_else(PoisonError::into_inner)
558        .remove(&name);
559    result
560}
561
562/// A stable 64-bit key for a PostgreSQL advisory lock: FNV-1a of the lock name. It does not change
563/// between releases, so two versions of the app agree on the lock.
564#[cfg(feature = "postgres")]
565fn advisory_key(name: &str) -> i64 {
566    let mut hash: u64 = 0xcbf2_9ce4_8422_2325;
567    for byte in name.bytes() {
568        hash ^= u64::from(byte);
569        hash = hash.wrapping_mul(0x0000_0100_0000_01b3);
570    }
571    hash as i64
572}
573
574/// The body of [`Migrator::import_diesel_history`], run under the lock.
575async fn import_unlocked(db: &Database) -> Result<Vec<String>, Error> {
576    ensure_table(db).await?;
577    Ok(Vec::new())
578}
579
580/// Whether the name starts with a 14-digit timestamp and an underscore, followed by a name.
581fn has_timestamp(name: &str) -> bool {
582    let bytes = name.as_bytes();
583    bytes.len() > 15 && bytes[..14].iter().all(u8::is_ascii_digit) && bytes[14] == b'_'
584}
585
586/// The timestamp of a name that passed `has_timestamp`. It is the `version` in the table.
587fn timestamp(name: &str) -> &str {
588    &name[..14]
589}
590
591/// The statement and values for the driver: `?` placeholders are numbered on PostgreSQL.
592fn prepared(driver: Driver, sql: &str, values: Vec<Value>) -> (String, Vec<Value>) {
593    driver.prepare(sql, &values)
594}
595
596/// Picks the SQL for the driver. Each argument is the same statement written for one database.
597#[cfg_attr(
598    not(all(feature = "mysql", feature = "postgres", feature = "sqlite")),
599    allow(unused_variables)
600)]
601fn pick<'a>(driver: Driver, mysql: &'a str, postgres: &'a str, sqlite: &'a str) -> &'a str {
602    match driver {
603        #[cfg(feature = "mysql")]
604        Driver::MySql => mysql,
605        #[cfg(feature = "postgres")]
606        Driver::Postgres => postgres,
607        #[cfg(feature = "sqlite")]
608        Driver::Sqlite => sqlite,
609    }
610}
611
612/// Creates the migrations table in Diesel's layout if it does not exist, with Orbit's `batch`
613/// column. A table that Diesel created has `version` and `run_on` only. For that table the `batch`
614/// column is added, and its existing rows get batch 0, so they count as one batch that was
615/// imported from Diesel. Nothing is dropped or renamed. Safe to run on every call.
616async fn ensure_table(db: &Database) -> Result<(), Error> {
617    let driver = db.driver();
618    let create = pick(
619        driver,
620        "CREATE TABLE IF NOT EXISTS __diesel_schema_migrations (\
621         version VARCHAR(50) NOT NULL PRIMARY KEY, \
622         run_on TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, \
623         batch INT NOT NULL DEFAULT 0)",
624        "CREATE TABLE IF NOT EXISTS __diesel_schema_migrations (\
625         version VARCHAR(50) NOT NULL PRIMARY KEY, \
626         run_on TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, \
627         batch INT NOT NULL DEFAULT 0)",
628        "CREATE TABLE IF NOT EXISTS __diesel_schema_migrations (\
629         version TEXT NOT NULL PRIMARY KEY, \
630         run_on TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, \
631         batch INTEGER NOT NULL DEFAULT 0)",
632    );
633    backend::execute(Target::Pool(&db.pool), create, &[]).await?;
634    if has_batch_column(db).await? {
635        return Ok(());
636    }
637    let added = backend::execute(
638        Target::Pool(&db.pool),
639        "ALTER TABLE __diesel_schema_migrations ADD COLUMN batch INT NOT NULL DEFAULT 0",
640        &[],
641    )
642    .await;
643    match added {
644        Ok(_) => Ok(()),
645        // Another process added the column after the check above.
646        Err(err) if is_duplicate_column(&err) => Ok(()),
647        Err(err) => Err(err.into()),
648    }
649}
650
651/// Whether the migrations table has a `batch` column.
652async fn has_batch_column(db: &Database) -> Result<bool, Error> {
653    let sql = pick(
654        db.driver(),
655        "SELECT COUNT(*) FROM information_schema.columns \
656         WHERE table_schema = DATABASE() AND table_name = '__diesel_schema_migrations' \
657         AND column_name = 'batch'",
658        "SELECT COUNT(*) FROM information_schema.columns \
659         WHERE table_schema = current_schema() AND table_name = '__diesel_schema_migrations' \
660         AND column_name = 'batch'",
661        "SELECT COUNT(*) FROM pragma_table_info('__diesel_schema_migrations') \
662         WHERE name = 'batch'",
663    );
664    let found: i64 = backend::scalar(Target::Pool(&db.pool), sql, &[]).await?;
665    Ok(found > 0)
666}
667
668/// Whether the error says that the column already exists: MySQL 1060, PostgreSQL 42701, or
669/// SQLite's "duplicate column name".
670fn is_duplicate_column(err: &sqlx::Error) -> bool {
671    err.as_database_error().is_some_and(|db_err| {
672        matches!(db_err.code().as_deref(), Some("1060" | "42701"))
673            || db_err.message().contains("duplicate column name")
674    })
675}
676
677/// The applied migrations, keyed by version, with their batch numbers.
678async fn applied(db: &Database) -> Result<HashMap<String, i64>, Error> {
679    let rows: Vec<(String, i32)> = backend::fetch_all(
680        Target::Pool(&db.pool),
681        &format!("SELECT version, batch FROM {TABLE}"),
682        &[],
683    )
684    .await?;
685    Ok(rows
686        .into_iter()
687        .map(|(version, batch)| (version, i64::from(batch)))
688        .collect())
689}
690
691/// The versions in the migrations table, or `None` when the table does not exist. Only reads the
692/// table.
693async fn recorded_versions(db: &Database) -> Result<Option<HashSet<String>>, Error> {
694    let sql = pick(
695        db.driver(),
696        "SELECT COUNT(*) FROM information_schema.tables \
697         WHERE table_schema = DATABASE() AND table_name = '__diesel_schema_migrations'",
698        "SELECT COUNT(*) FROM information_schema.tables \
699         WHERE table_schema = current_schema() AND table_name = '__diesel_schema_migrations'",
700        "SELECT COUNT(*) FROM sqlite_master \
701         WHERE type = 'table' AND name = '__diesel_schema_migrations'",
702    );
703    let found: i64 = backend::scalar(Target::Pool(&db.pool), sql, &[]).await?;
704    if found == 0 {
705        return Ok(None);
706    }
707    let versions: Vec<(String,)> = backend::fetch_all(
708        Target::Pool(&db.pool),
709        &format!("SELECT version FROM {TABLE}"),
710        &[],
711    )
712    .await?;
713    Ok(Some(versions.into_iter().map(|(v,)| v).collect()))
714}
715
716/// Reads a migration file. The error names the file.
717pub(crate) fn read_file(path: &Path) -> Result<String, Error> {
718    std::fs::read_to_string(path).map_err(|source| Error::Read {
719        path: path.to_path_buf(),
720        source,
721    })
722}
orbit_orm/migrate.rs

orbit_orm/
migrate.rs
