♻️ Major code restructuring

src/db/migration.rs

@@ -0,0 +1,32 @@
+//pub mod cli;
+pub mod connection;
+pub mod manager;
+pub mod migrator;
+pub mod prelude;
+pub mod seaql_migrations;
+//pub mod util;
+
+pub use connection::*;
+pub use manager::*;
+//pub use migrator::*;
+
+//pub use async_trait;
+//pub use sea_orm;
+//pub use sea_orm::sea_query;
+use sea_orm::DbErr;
+
+pub trait MigrationName {
+    fn name(&self) -> &str;
+}
+
+/// The migration definition
+#[async_trait::async_trait]
+pub trait MigrationTrait: MigrationName + Send + Sync {
+    /// Define actions to perform when applying the migration
+    async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr>;
+
+    /// Define actions to perform when rolling back the migration
+    async fn down(&self, _manager: &SchemaManager) -> Result<(), DbErr> {
+        Err(DbErr::Migration("We Don't Do That Here".to_owned()))
+    }
+}

src/db/migration/connection.rs

@@ -0,0 +1,148 @@
+use futures::Future;
+use sea_orm::{
+    AccessMode, ConnectionTrait, DatabaseConnection, DatabaseTransaction, DbBackend, DbErr,
+    ExecResult, IsolationLevel, QueryResult, Statement, TransactionError, TransactionTrait,
+};
+use std::pin::Pin;
+
+pub enum SchemaManagerConnection<'c> {
+    Connection(&'c DatabaseConnection),
+    Transaction(&'c DatabaseTransaction),
+}
+
+#[async_trait::async_trait]
+impl<'c> ConnectionTrait for SchemaManagerConnection<'c> {
+    fn get_database_backend(&self) -> DbBackend {
+        match self {
+            SchemaManagerConnection::Connection(conn) => conn.get_database_backend(),
+            SchemaManagerConnection::Transaction(trans) => trans.get_database_backend(),
+        }
+    }
+
+    async fn execute(&self, stmt: Statement) -> Result<ExecResult, DbErr> {
+        match self {
+            SchemaManagerConnection::Connection(conn) => conn.execute(stmt).await,
+            SchemaManagerConnection::Transaction(trans) => trans.execute(stmt).await,
+        }
+    }
+
+    async fn execute_unprepared(&self, sql: &str) -> Result<ExecResult, DbErr> {
+        match self {
+            SchemaManagerConnection::Connection(conn) => conn.execute_unprepared(sql).await,
+            SchemaManagerConnection::Transaction(trans) => trans.execute_unprepared(sql).await,
+        }
+    }
+
+    async fn query_one(&self, stmt: Statement) -> Result<Option<QueryResult>, DbErr> {
+        match self {
+            SchemaManagerConnection::Connection(conn) => conn.query_one(stmt).await,
+            SchemaManagerConnection::Transaction(trans) => trans.query_one(stmt).await,
+        }
+    }
+
+    async fn query_all(&self, stmt: Statement) -> Result<Vec<QueryResult>, DbErr> {
+        match self {
+            SchemaManagerConnection::Connection(conn) => conn.query_all(stmt).await,
+            SchemaManagerConnection::Transaction(trans) => trans.query_all(stmt).await,
+        }
+    }
+
+    fn is_mock_connection(&self) -> bool {
+        match self {
+            SchemaManagerConnection::Connection(conn) => conn.is_mock_connection(),
+            SchemaManagerConnection::Transaction(trans) => trans.is_mock_connection(),
+        }
+    }
+}
+
+#[async_trait::async_trait]
+impl<'c> TransactionTrait for SchemaManagerConnection<'c> {
+    async fn begin(&self) -> Result<DatabaseTransaction, DbErr> {
+        match self {
+            SchemaManagerConnection::Connection(conn) => conn.begin().await,
+            SchemaManagerConnection::Transaction(trans) => trans.begin().await,
+        }
+    }
+
+    async fn begin_with_config(
+        &self,
+        isolation_level: Option<IsolationLevel>,
+        access_mode: Option<AccessMode>,
+    ) -> Result<DatabaseTransaction, DbErr> {
+        match self {
+            SchemaManagerConnection::Connection(conn) => {
+                conn.begin_with_config(isolation_level, access_mode).await
+            }
+            SchemaManagerConnection::Transaction(trans) => {
+                trans.begin_with_config(isolation_level, access_mode).await
+            }
+        }
+    }
+
+    async fn transaction<F, T, E>(&self, callback: F) -> Result<T, TransactionError<E>>
+    where
+        F: for<'a> FnOnce(
+                &'a DatabaseTransaction,
+            ) -> Pin<Box<dyn Future<Output = Result<T, E>> + Send + 'a>>
+            + Send,
+        T: Send,
+        E: std::error::Error + Send,
+    {
+        match self {
+            SchemaManagerConnection::Connection(conn) => conn.transaction(callback).await,
+            SchemaManagerConnection::Transaction(trans) => trans.transaction(callback).await,
+        }
+    }
+
+    async fn transaction_with_config<F, T, E>(
+        &self,
+        callback: F,
+        isolation_level: Option<IsolationLevel>,
+        access_mode: Option<AccessMode>,
+    ) -> Result<T, TransactionError<E>>
+    where
+        F: for<'a> FnOnce(
+                &'a DatabaseTransaction,
+            ) -> Pin<Box<dyn Future<Output = Result<T, E>> + Send + 'a>>
+            + Send,
+        T: Send,
+        E: std::error::Error + Send,
+    {
+        match self {
+            SchemaManagerConnection::Connection(conn) => {
+                conn.transaction_with_config(callback, isolation_level, access_mode)
+                    .await
+            }
+            SchemaManagerConnection::Transaction(trans) => {
+                trans
+                    .transaction_with_config(callback, isolation_level, access_mode)
+                    .await
+            }
+        }
+    }
+}
+
+pub trait IntoSchemaManagerConnection<'c>: Send
+where
+    Self: 'c,
+{
+    fn into_schema_manager_connection(self) -> SchemaManagerConnection<'c>;
+}
+
+impl<'c> IntoSchemaManagerConnection<'c> for SchemaManagerConnection<'c> {
+    fn into_schema_manager_connection(self) -> SchemaManagerConnection<'c> {
+        self
+    }
+}
+
+impl<'c> IntoSchemaManagerConnection<'c> for &'c DatabaseConnection {
+    fn into_schema_manager_connection(self) -> SchemaManagerConnection<'c> {
+        SchemaManagerConnection::Connection(self)
+    }
+}
+
+impl<'c> IntoSchemaManagerConnection<'c> for &'c DatabaseTransaction {
+    fn into_schema_manager_connection(self) -> SchemaManagerConnection<'c> {
+        SchemaManagerConnection::Transaction(self)
+    }
+}

src/db/migration/manager.rs

@@ -0,0 +1,160 @@
+use super::{IntoSchemaManagerConnection, SchemaManagerConnection};
+use sea_orm::sea_query::{
+    extension::postgres::{TypeAlterStatement, TypeCreateStatement, TypeDropStatement},
+    ForeignKeyCreateStatement, ForeignKeyDropStatement, IndexCreateStatement, IndexDropStatement,
+    TableAlterStatement, TableCreateStatement, TableDropStatement, TableRenameStatement,
+    TableTruncateStatement,
+};
+use sea_orm::{ConnectionTrait, DbBackend, DbErr, StatementBuilder};
+use sea_schema::{mysql::MySql, postgres::Postgres, probe::SchemaProbe, sqlite::Sqlite};
+
+/// Helper struct for writing migration scripts in migration file
+pub struct SchemaManager<'c> {
+    conn: SchemaManagerConnection<'c>,
+}
+
+impl<'c> SchemaManager<'c> {
+    pub fn new<T>(conn: T) -> Self
+    where
+        T: IntoSchemaManagerConnection<'c>,
+    {
+        Self {
+            conn: conn.into_schema_manager_connection(),
+        }
+    }
+
+    pub async fn exec_stmt<S>(&self, stmt: S) -> Result<(), DbErr>
+    where
+        S: StatementBuilder,
+    {
+        let builder = self.conn.get_database_backend();
+        self.conn.execute(builder.build(&stmt)).await.map(|_| ())
+    }
+
+    pub fn get_database_backend(&self) -> DbBackend {
+        self.conn.get_database_backend()
+    }
+
+    pub fn get_connection(&self) -> &SchemaManagerConnection<'c> {
+        &self.conn
+    }
+}
+
+/// Schema Creation
+impl<'c> SchemaManager<'c> {
+    pub async fn create_table(&self, stmt: TableCreateStatement) -> Result<(), DbErr> {
+        self.exec_stmt(stmt).await
+    }
+
+    pub async fn create_index(&self, stmt: IndexCreateStatement) -> Result<(), DbErr> {
+        self.exec_stmt(stmt).await
+    }
+
+    pub async fn create_foreign_key(&self, stmt: ForeignKeyCreateStatement) -> Result<(), DbErr> {
+        self.exec_stmt(stmt).await
+    }
+
+    pub async fn create_type(&self, stmt: TypeCreateStatement) -> Result<(), DbErr> {
+        self.exec_stmt(stmt).await
+    }
+}
+
+/// Schema Mutation
+impl<'c> SchemaManager<'c> {
+    pub async fn alter_table(&self, stmt: TableAlterStatement) -> Result<(), DbErr> {
+        self.exec_stmt(stmt).await
+    }
+
+    pub async fn drop_table(&self, stmt: TableDropStatement) -> Result<(), DbErr> {
+        self.exec_stmt(stmt).await
+    }
+
+    pub async fn rename_table(&self, stmt: TableRenameStatement) -> Result<(), DbErr> {
+        self.exec_stmt(stmt).await
+    }
+
+    pub async fn truncate_table(&self, stmt: TableTruncateStatement) -> Result<(), DbErr> {
+        self.exec_stmt(stmt).await
+    }
+
+    pub async fn drop_index(&self, stmt: IndexDropStatement) -> Result<(), DbErr> {
+        self.exec_stmt(stmt).await
+    }
+
+    pub async fn drop_foreign_key(&self, stmt: ForeignKeyDropStatement) -> Result<(), DbErr> {
+        self.exec_stmt(stmt).await
+    }
+
+    pub async fn alter_type(&self, stmt: TypeAlterStatement) -> Result<(), DbErr> {
+        self.exec_stmt(stmt).await
+    }
+
+    pub async fn drop_type(&self, stmt: TypeDropStatement) -> Result<(), DbErr> {
+        self.exec_stmt(stmt).await
+    }
+}
+
+/// Schema Inspection
+impl<'c> SchemaManager<'c> {
+    pub async fn has_table<T>(&self, table: T) -> Result<bool, DbErr>
+    where
+        T: AsRef<str>,
+    {
+        let stmt = match self.conn.get_database_backend() {
+            DbBackend::MySql => MySql::has_table(table),
+            DbBackend::Postgres => Postgres::has_table(table),
+            DbBackend::Sqlite => Sqlite::has_table(table),
+        };
+
+        let builder = self.conn.get_database_backend();
+        let res = self
+            .conn
+            .query_one(builder.build(&stmt))
+            .await?
+            .ok_or_else(|| DbErr::Custom("Failed to check table exists".to_owned()))?;
+
+        res.try_get("", "has_table")
+    }
+
+    pub async fn has_column<T, C>(&self, table: T, column: C) -> Result<bool, DbErr>
+    where
+        T: AsRef<str>,
+        C: AsRef<str>,
+    {
+        let stmt = match self.conn.get_database_backend() {
+            DbBackend::MySql => MySql::has_column(table, column),
+            DbBackend::Postgres => Postgres::has_column(table, column),
+            DbBackend::Sqlite => Sqlite::has_column(table, column),
+        };
+
+        let builder = self.conn.get_database_backend();
+        let res = self
+            .conn
+            .query_one(builder.build(&stmt))
+            .await?
+            .ok_or_else(|| DbErr::Custom("Failed to check column exists".to_owned()))?;
+
+        res.try_get("", "has_column")
+    }
+
+    pub async fn has_index<T, I>(&self, table: T, index: I) -> Result<bool, DbErr>
+    where
+        T: AsRef<str>,
+        I: AsRef<str>,
+    {
+        let stmt = match self.conn.get_database_backend() {
+            DbBackend::MySql => MySql::has_index(table, index),
+            DbBackend::Postgres => Postgres::has_index(table, index),
+            DbBackend::Sqlite => Sqlite::has_index(table, index),
+        };
+
+        let builder = self.conn.get_database_backend();
+        let res = self
+            .conn
+            .query_one(builder.build(&stmt))
+            .await?
+            .ok_or_else(|| DbErr::Custom("Failed to check index exists".to_owned()))?;
+
+        res.try_get("", "has_index")
+    }
+}

src/db/migration/migrator.rs

@@ -0,0 +1,589 @@
+use futures::Future;
+use std::collections::HashSet;
+use std::fmt::Display;
+use std::pin::Pin;
+use std::time::SystemTime;
+use tracing::info;
+
+use sea_orm::sea_query::{
+    self, extension::postgres::Type, Alias, Expr, ForeignKey, IntoIden, JoinType, Order, Query,
+    SelectStatement, SimpleExpr, Table,
+};
+use sea_orm::{
+    ActiveModelTrait, ActiveValue, Condition, ConnectionTrait, DbBackend, DbErr, DeriveIden,
+    DynIden, EntityTrait, FromQueryResult, Iterable, QueryFilter, Schema, Statement,
+    TransactionTrait,
+};
+use sea_schema::{mysql::MySql, postgres::Postgres, probe::SchemaProbe, sqlite::Sqlite};
+
+use super::{seaql_migrations, IntoSchemaManagerConnection, MigrationTrait, SchemaManager};
+
+#[derive(Copy, Clone, Debug, PartialEq, Eq)]
+/// Status of migration
+pub enum MigrationStatus {
+    /// Not yet applied
+    Pending,
+    /// Applied
+    Applied,
+}
+
+impl Display for MigrationStatus {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        let status = match self {
+            MigrationStatus::Pending => "Pending",
+            MigrationStatus::Applied => "Applied",
+        };
+        write!(f, "{status}")
+    }
+}
+
+pub struct Migration {
+    migration: Box<dyn MigrationTrait>,
+    status: MigrationStatus,
+}
+
+impl Migration {
+    /// Get migration name from MigrationName trait implementation
+    pub fn name(&self) -> &str {
+        self.migration.name()
+    }
+
+    /// Get migration status
+    pub fn status(&self) -> MigrationStatus {
+        self.status
+    }
+}
+
+/// Performing migrations on a database
+#[async_trait::async_trait]
+pub trait MigratorTrait: Send {
+    /// Vector of migrations in time sequence
+    fn migrations() -> Vec<Box<dyn MigrationTrait>>;
+
+    /// Name of the migration table, it is `seaql_migrations` by default
+    fn migration_table_name() -> DynIden {
+        seaql_migrations::Entity.into_iden()
+    }
+
+    /// Get list of migrations wrapped in `Migration` struct
+    fn get_migration_files() -> Vec<Migration> {
+        Self::migrations()
+            .into_iter()
+            .map(|migration| Migration {
+                migration,
+                status: MigrationStatus::Pending,
+            })
+            .collect()
+    }
+
+    /// Get list of applied migrations from database
+    async fn get_migration_models<C>(db: &C) -> Result<Vec<seaql_migrations::Model>, DbErr>
+    where
+        C: ConnectionTrait,
+    {
+        Self::install(db).await?;
+        let stmt = Query::select()
+            .table_name(Self::migration_table_name())
+            .columns(seaql_migrations::Column::iter().map(IntoIden::into_iden))
+            .order_by(seaql_migrations::Column::Version, Order::Asc)
+            .to_owned();
+        let builder = db.get_database_backend();
+        seaql_migrations::Model::find_by_statement(builder.build(&stmt))
+            .all(db)
+            .await
+    }
+
+    /// Get list of migrations with status
+    async fn get_migration_with_status<C>(db: &C) -> Result<Vec<Migration>, DbErr>
+    where
+        C: ConnectionTrait,
+    {
+        Self::install(db).await?;
+        let mut migration_files = Self::get_migration_files();
+        let migration_models = Self::get_migration_models(db).await?;
+
+        let migration_in_db: HashSet<String> = migration_models
+            .into_iter()
+            .map(|model| model.version)
+            .collect();
+        let migration_in_fs: HashSet<String> = migration_files
+            .iter()
+            .map(|file| file.migration.name().to_string())
+            .collect();
+
+        let pending_migrations = &migration_in_fs - &migration_in_db;
+        for migration_file in migration_files.iter_mut() {
+            if !pending_migrations.contains(migration_file.migration.name()) {
+                migration_file.status = MigrationStatus::Applied;
+            }
+        }
+        /*
+        let missing_migrations_in_fs = &migration_in_db - &migration_in_fs;
+        let errors: Vec<String> = missing_migrations_in_fs
+            .iter()
+            .map(|missing_migration| {
+                format!("Migration file of version '{missing_migration}' is missing, this migration has been applied but its file is missing")
+            }).collect();
+
+        if !errors.is_empty() {
+            Err(DbErr::Custom(errors.join("\n")))
+        } else { */
+        Ok(migration_files)
+        /* } */
+    }
+
+    /// Get list of pending migrations
+    async fn get_pending_migrations<C>(db: &C) -> Result<Vec<Migration>, DbErr>
+    where
+        C: ConnectionTrait,
+    {
+        Self::install(db).await?;
+        Ok(Self::get_migration_with_status(db)
+            .await?
+            .into_iter()
+            .filter(|file| file.status == MigrationStatus::Pending)
+            .collect())
+    }
+
+    /// Get list of applied migrations
+    async fn get_applied_migrations<C>(db: &C) -> Result<Vec<Migration>, DbErr>
+    where
+        C: ConnectionTrait,
+    {
+        Self::install(db).await?;
+        Ok(Self::get_migration_with_status(db)
+            .await?
+            .into_iter()
+            .filter(|file| file.status == MigrationStatus::Applied)
+            .collect())
+    }
+
+    /// Create migration table `seaql_migrations` in the database
+    async fn install<C>(db: &C) -> Result<(), DbErr>
+    where
+        C: ConnectionTrait,
+    {
+        let builder = db.get_database_backend();
+        let schema = Schema::new(builder);
+        let mut stmt = schema
+            .create_table_from_entity(seaql_migrations::Entity)
+            .table_name(Self::migration_table_name());
+        stmt.if_not_exists();
+        db.execute(builder.build(&stmt)).await.map(|_| ())
+    }
+
+    /// Check the status of all migrations
+    async fn status<C>(db: &C) -> Result<(), DbErr>
+    where
+        C: ConnectionTrait,
+    {
+        Self::install(db).await?;
+
+        info!("Checking migration status");
+
+        for Migration { migration, status } in Self::get_migration_with_status(db).await? {
+            info!("Migration '{}'... {}", migration.name(), status);
+        }
+
+        Ok(())
+    }
+
+    /// Drop all tables from the database, then reapply all migrations
+    async fn fresh<'c, C>(db: C) -> Result<(), DbErr>
+    where
+        C: IntoSchemaManagerConnection<'c>,
+    {
+        exec_with_connection::<'_, _, _>(db, move |manager| {
+            Box::pin(async move { exec_fresh::<Self>(manager).await })
+        })
+        .await
+    }
+
+    /// Rollback all applied migrations, then reapply all migrations
+    async fn refresh<'c, C>(db: C) -> Result<(), DbErr>
+    where
+        C: IntoSchemaManagerConnection<'c>,
+    {
+        exec_with_connection::<'_, _, _>(db, move |manager| {
+            Box::pin(async move {
+                exec_down::<Self>(manager, None).await?;
+                exec_up::<Self>(manager, None).await
+            })
+        })
+        .await
+    }
+
+    /// Rollback all applied migrations
+    async fn reset<'c, C>(db: C) -> Result<(), DbErr>
+    where
+        C: IntoSchemaManagerConnection<'c>,
+    {
+        exec_with_connection::<'_, _, _>(db, move |manager| {
+            Box::pin(async move { exec_down::<Self>(manager, None).await })
+        })
+        .await
+    }
+
+    /// Apply pending migrations
+    async fn up<'c, C>(db: C, steps: Option<u32>) -> Result<(), DbErr>
+    where
+        C: IntoSchemaManagerConnection<'c>,
+    {
+        exec_with_connection::<'_, _, _>(db, move |manager| {
+            Box::pin(async move { exec_up::<Self>(manager, steps).await })
+        })
+        .await
+    }
+
+    /// Rollback applied migrations
+    async fn down<'c, C>(db: C, steps: Option<u32>) -> Result<(), DbErr>
+    where
+        C: IntoSchemaManagerConnection<'c>,
+    {
+        exec_with_connection::<'_, _, _>(db, move |manager| {
+            Box::pin(async move { exec_down::<Self>(manager, steps).await })
+        })
+        .await
+    }
+}
+
+async fn exec_with_connection<'c, C, F>(db: C, f: F) -> Result<(), DbErr>
+where
+    C: IntoSchemaManagerConnection<'c>,
+    F: for<'b> Fn(
+        &'b SchemaManager<'_>,
+    ) -> Pin<Box<dyn Future<Output = Result<(), DbErr>> + Send + 'b>>,
+{
+    let db = db.into_schema_manager_connection();
+
+    match db.get_database_backend() {
+        DbBackend::Postgres => {
+            let transaction = db.begin().await?;
+            let manager = SchemaManager::new(&transaction);
+            f(&manager).await?;
+            transaction.commit().await
+        }
+        DbBackend::MySql | DbBackend::Sqlite => {
+            let manager = SchemaManager::new(db);
+            f(&manager).await
+        }
+    }
+}
+
+async fn exec_fresh<M>(manager: &SchemaManager<'_>) -> Result<(), DbErr>
+where
+    M: MigratorTrait + ?Sized,
+{
+    let db = manager.get_connection();
+
+    M::install(db).await?;
+    let db_backend = db.get_database_backend();
+
+    // Temporarily disable the foreign key check
+    if db_backend == DbBackend::Sqlite {
+        info!("Disabling foreign key check");
+        db.execute(Statement::from_string(
+            db_backend,
+            "PRAGMA foreign_keys = OFF".to_owned(),
+        ))
+        .await?;
+        info!("Foreign key check disabled");
+    }
+
+    // Drop all foreign keys
+    if db_backend == DbBackend::MySql {
+        info!("Dropping all foreign keys");
+        let stmt = query_mysql_foreign_keys(db);
+        let rows = db.query_all(db_backend.build(&stmt)).await?;
+        for row in rows.into_iter() {
+            let constraint_name: String = row.try_get("", "CONSTRAINT_NAME")?;
+            let table_name: String = row.try_get("", "TABLE_NAME")?;
+            info!(
+                "Dropping foreign key '{}' from table '{}'",
+                constraint_name, table_name
+            );
+            let mut stmt = ForeignKey::drop();
+            stmt.table(Alias::new(table_name.as_str()))
+                .name(constraint_name.as_str());
+            db.execute(db_backend.build(&stmt)).await?;
+            info!("Foreign key '{}' has been dropped", constraint_name);
+        }
+        info!("All foreign keys dropped");
+    }
+
+    // Drop all tables
+    let stmt = query_tables(db);
+    let rows = db.query_all(db_backend.build(&stmt)).await?;
+    for row in rows.into_iter() {
+        let table_name: String = row.try_get("", "table_name")?;
+        info!("Dropping table '{}'", table_name);
+        let mut stmt = Table::drop();
+        stmt.table(Alias::new(table_name.as_str()))
+            .if_exists()
+            .cascade();
+        db.execute(db_backend.build(&stmt)).await?;
+        info!("Table '{}' has been dropped", table_name);
+    }
+
+    // Drop all types
+    if db_backend == DbBackend::Postgres {
+        info!("Dropping all types");
+        let stmt = query_pg_types(db);
+        let rows = db.query_all(db_backend.build(&stmt)).await?;
+        for row in rows {
+            let type_name: String = row.try_get("", "typname")?;
+            info!("Dropping type '{}'", type_name);
+            let mut stmt = Type::drop();
+            stmt.name(Alias::new(&type_name));
+            db.execute(db_backend.build(&stmt)).await?;
+            info!("Type '{}' has been dropped", type_name);
+        }
+    }
+
+    // Restore the foreign key check
+    if db_backend == DbBackend::Sqlite {
+        info!("Restoring foreign key check");
+        db.execute(Statement::from_string(
+            db_backend,
+            "PRAGMA foreign_keys = ON".to_owned(),
+        ))
+        .await?;
+        info!("Foreign key check restored");
+    }
+
+    // Reapply all migrations
+    exec_up::<M>(manager, None).await
+}
+
+async fn exec_up<M>(manager: &SchemaManager<'_>, mut steps: Option<u32>) -> Result<(), DbErr>
+where
+    M: MigratorTrait + ?Sized,
+{
+    let db = manager.get_connection();
+
+    M::install(db).await?;
+
+    if let Some(steps) = steps {
+        info!("Applying {} pending migrations", steps);
+    } else {
+        info!("Applying all pending migrations");
+    }
+
+    let migrations = M::get_pending_migrations(db).await?.into_iter();
+    if migrations.len() == 0 {
+        info!("No pending migrations");
+    }
+    for Migration { migration, .. } in migrations {
+        if let Some(steps) = steps.as_mut() {
+            if steps == &0 {
+                break;
+            }
+            *steps -= 1;
+        }
+        info!("Applying migration '{}'", migration.name());
+        migration.up(manager).await?;
+        info!("Migration '{}' has been applied", migration.name());
+        let now = SystemTime::now()
+            .duration_since(SystemTime::UNIX_EPOCH)
+            .expect("SystemTime before UNIX EPOCH!");
+        seaql_migrations::Entity::insert(seaql_migrations::ActiveModel {
+            version: ActiveValue::Set(migration.name().to_owned()),
+            applied_at: ActiveValue::Set(now.as_secs() as i64),
+        })
+        .table_name(M::migration_table_name())
+        .exec(db)
+        .await?;
+    }
+
+    Ok(())
+}
+
+async fn exec_down<M>(manager: &SchemaManager<'_>, mut steps: Option<u32>) -> Result<(), DbErr>
+where
+    M: MigratorTrait + ?Sized,
+{
+    let db = manager.get_connection();
+
+    M::install(db).await?;
+
+    if let Some(steps) = steps {
+        info!("Rolling back {} applied migrations", steps);
+    } else {
+        info!("Rolling back all applied migrations");
+    }
+
+    let migrations = M::get_applied_migrations(db).await?.into_iter().rev();
+    if migrations.len() == 0 {
+        info!("No applied migrations");
+    }
+    for Migration { migration, .. } in migrations {
+        if let Some(steps) = steps.as_mut() {
+            if steps == &0 {
+                break;
+            }
+            *steps -= 1;
+        }
+        info!("Rolling back migration '{}'", migration.name());
+        migration.down(manager).await?;
+        info!("Migration '{}' has been rollbacked", migration.name());
+        seaql_migrations::Entity::delete_many()
+            .filter(Expr::col(seaql_migrations::Column::Version).eq(migration.name()))
+            .table_name(M::migration_table_name())
+            .exec(db)
+            .await?;
+    }
+
+    Ok(())
+}
+
+fn query_tables<C>(db: &C) -> SelectStatement
+where
+    C: ConnectionTrait,
+{
+    match db.get_database_backend() {
+        DbBackend::MySql => MySql::query_tables(),
+        DbBackend::Postgres => Postgres::query_tables(),
+        DbBackend::Sqlite => Sqlite::query_tables(),
+    }
+}
+
+fn get_current_schema<C>(db: &C) -> SimpleExpr
+where
+    C: ConnectionTrait,
+{
+    match db.get_database_backend() {
+        DbBackend::MySql => MySql::get_current_schema(),
+        DbBackend::Postgres => Postgres::get_current_schema(),
+        DbBackend::Sqlite => unimplemented!(),
+    }
+}
+
+#[derive(DeriveIden)]
+enum InformationSchema {
+    #[sea_orm(iden = "information_schema")]
+    Schema,
+    #[sea_orm(iden = "TABLE_NAME")]
+    TableName,
+    #[sea_orm(iden = "CONSTRAINT_NAME")]
+    ConstraintName,
+    TableConstraints,
+    TableSchema,
+    ConstraintType,
+}
+
+fn query_mysql_foreign_keys<C>(db: &C) -> SelectStatement
+where
+    C: ConnectionTrait,
+{
+    let mut stmt = Query::select();
+    stmt.columns([
+        InformationSchema::TableName,
+        InformationSchema::ConstraintName,
+    ])
+    .from((
+        InformationSchema::Schema,
+        InformationSchema::TableConstraints,
+    ))
+    .cond_where(
+        Condition::all()
+            .add(Expr::expr(get_current_schema(db)).equals((
+                InformationSchema::TableConstraints,
+                InformationSchema::TableSchema,
+            )))
+            .add(
+                Expr::col((
+                    InformationSchema::TableConstraints,
+                    InformationSchema::ConstraintType,
+                ))
+                .eq("FOREIGN KEY"),
+            ),
+    );
+    stmt
+}
+
+#[derive(DeriveIden)]
+enum PgType {
+    Table,
+    Typname,
+    Typnamespace,
+    Typelem,
+}
+
+#[derive(DeriveIden)]
+enum PgNamespace {
+    Table,
+    Oid,
+    Nspname,
+}
+
+fn query_pg_types<C>(db: &C) -> SelectStatement
+where
+    C: ConnectionTrait,
+{
+    let mut stmt = Query::select();
+    stmt.column(PgType::Typname)
+        .from(PgType::Table)
+        .join(
+            JoinType::LeftJoin,
+            PgNamespace::Table,
+            Expr::col((PgNamespace::Table, PgNamespace::Oid))
+                .equals((PgType::Table, PgType::Typnamespace)),
+        )
+        .cond_where(
+            Condition::all()
+                .add(
+                    Expr::expr(get_current_schema(db))
+                        .equals((PgNamespace::Table, PgNamespace::Nspname)),
+                )
+                .add(Expr::col((PgType::Table, PgType::Typelem)).eq(0)),
+        );
+    stmt
+}
+
+trait QueryTable {
+    type Statement;
+
+    fn table_name(self, table_name: DynIden) -> Self::Statement;
+}
+
+impl QueryTable for SelectStatement {
+    type Statement = SelectStatement;
+
+    fn table_name(mut self, table_name: DynIden) -> SelectStatement {
+        self.from(table_name);
+        self
+    }
+}
+
+impl QueryTable for sea_query::TableCreateStatement {
+    type Statement = sea_query::TableCreateStatement;
+
+    fn table_name(mut self, table_name: DynIden) -> sea_query::TableCreateStatement {
+        self.table(table_name);
+        self
+    }
+}
+
+impl<A> QueryTable for sea_orm::Insert<A>
+where
+    A: ActiveModelTrait,
+{
+    type Statement = sea_orm::Insert<A>;
+
+    fn table_name(mut self, table_name: DynIden) -> sea_orm::Insert<A> {
+        sea_orm::QueryTrait::query(&mut self).into_table(table_name);
+        self
+    }
+}
+
+impl<E> QueryTable for sea_orm::DeleteMany<E>
+where
+    E: EntityTrait,
+{
+    type Statement = sea_orm::DeleteMany<E>;
+
+    fn table_name(mut self, table_name: DynIden) -> sea_orm::DeleteMany<E> {
+        sea_orm::QueryTrait::query(&mut self).from_table(table_name);
+        self
+    }
+}

src/db/migration/prelude.rs

@@ -0,0 +1,13 @@
+//pub use super::cli;
+
+pub use super::connection::IntoSchemaManagerConnection;
+pub use super::connection::SchemaManagerConnection;
+pub use super::manager::SchemaManager;
+pub use super::migrator::MigratorTrait;
+pub use super::{MigrationName, MigrationTrait};
+pub use async_trait;
+pub use sea_orm;
+pub use sea_orm::sea_query;
+pub use sea_orm::sea_query::*;
+pub use sea_orm::DeriveIden;
+pub use sea_orm::DeriveMigrationName;

src/db/migration/seaql_migrations.rs

@@ -0,0 +1,14 @@
+use sea_orm::entity::prelude::*;
+
+#[derive(Clone, Debug, PartialEq, Eq, DeriveEntityModel)]
+#[sea_orm(table_name = "seaql_migrations")]
+pub struct Model {
+    #[sea_orm(primary_key, auto_increment = false)]
+    pub version: String,
+    pub applied_at: i64,
+}
+
+#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
+pub enum Relation {}
+
+impl ActiveModelBehavior for ActiveModel {}