feat(migration): add function to undo migrations

Right now, there is only a `run` function to programatically run the
migrations, which is great, but nothing to run the migrations down.

This function adds the possibility to undo the migrations until a
specific version (could say -1 or 0 to remove them all).

With this feature, it's now possible, in the end to end or integration
tests to run the migrations and undo them between each test set and
therefore test the migrations themselves.

This is the kind of feature that some ORM have like sequelize in nodejs
that allow you to undo migrations programatically.

Reference to the doc:
https://sequelize.org/v7/manual/migrations.html#undoing-migrations

Signed-off-by: Jérémie Drouet <[email protected]>

Author: jdrouet

sqlx-core/src/migrate/migrator.rs

@@ -137,4 +137,64 @@ impl Migrator {
 
         Ok(())
     }
+
+    /// Run down migrations against the database until a specific version.
+    ///
+    /// # Examples
+    ///
+    /// ```rust,no_run
+    /// # use sqlx_core::migrate::MigrateError;
+    /// # #[cfg(feature = "sqlite")]
+    /// # fn main() -> Result<(), MigrateError> {
+    /// #     sqlx_rt::block_on(async move {
+    /// # use sqlx_core::migrate::Migrator;
+    /// let m = Migrator::new(std::path::Path::new("./migrations")).await?;
+    /// let pool = sqlx_core::sqlite::SqlitePoolOptions::new().connect("sqlite::memory:").await?;
+    /// m.undo(&pool, 4).await
+    /// #     })
+    /// # }
+    /// ```
+    pub async fn undo<'a, A>(&self, migrator: A, target: i64) -> Result<(), MigrateError>
+    where
+        A: Acquire<'a>,
+        <A::Connection as Deref>::Target: Migrate,
+    {
+        let mut conn = migrator.acquire().await?;
+
+        // lock the database for exclusive access by the migrator
+        conn.lock().await?;
+
+        // creates [_migrations] table only if needed
+        // eventually this will likely migrate previous versions of the table
+        conn.ensure_migrations_table().await?;
+
+        let version = conn.dirty_version().await?;
+        if let Some(version) = version {
+            return Err(MigrateError::Dirty(version));
+        }
+
+        let applied_migrations = conn.list_applied_migrations().await?;
+        validate_applied_migrations(&applied_migrations, self)?;
+
+        let applied_migrations: HashMap<_, _> = applied_migrations
+            .into_iter()
+            .map(|m| (m.version, m))
+            .collect();
+
+        for migration in self
+            .iter()
+            .rev()
+            .filter(|m| m.migration_type.is_down_migration())
+            .filter(|m| applied_migrations.contains_key(&m.version))
+            .filter(|m| m.version > target)
+        {
+            conn.apply(migration).await?;
+        }
+
+        // unlock the migrator to allow other migrators to run
+        // but do nothing as we already migrated
+        conn.unlock().await?;
+
+        Ok(())
+    }
 }