1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144

use crate::error::WrapMigrationError;
use crate::traits::{
    verify_migrations, ASSERT_MIGRATIONS_TABLE_QUERY, GET_APPLIED_MIGRATIONS_QUERY,
    GET_LAST_APPLIED_MIGRATION_QUERY,
};
use crate::{Error, Migration, Report, Target};
use chrono::Local;

pub trait Transaction {
    type Error: std::error::Error + Send + Sync + 'static;

    fn execute(&mut self, queries: &[&str]) -> Result<usize, Self::Error>;
}

pub trait Query<T>: Transaction {
    fn query(&mut self, query: &str) -> Result<T, Self::Error>;
}

fn migrate<T: Transaction>(
    transaction: &mut T,
    migrations: Vec<Migration>,
    target: Target,
) -> Result<Report, Error> {
    let mut applied_migrations = vec![];

    for mut migration in migrations.into_iter() {
        if let Target::Version(input_target) = target {
            if input_target < migration.version() {
                log::info!("stoping at migration: {}, due to user option", input_target);
                break;
            }
        }

        log::info!("applying migration: {}", migration);
        migration.set_applied();
        let update_query = &format!(
                "INSERT INTO refinery_schema_history (version, name, applied_on, checksum) VALUES ({}, '{}', '{}', '{}')",
                // safe to call unwrap as we just converted it to applied
                migration.version(), migration.name(), migration.applied_on().unwrap().to_rfc3339(), migration.checksum());

        let sql = migration.sql().expect("sql must be Some!");
        transaction.execute(&[sql, update_query]).migration_err(
            &format!("error applying migration {}", migration),
            Some(&applied_migrations),
        )?;
        applied_migrations.push(migration);
    }
    Ok(Report::new(applied_migrations))
}

fn migrate_grouped<T: Transaction>(
    transaction: &mut T,
    migrations: Vec<Migration>,
    target: Target,
) -> Result<Report, Error> {
    let mut grouped_migrations = Vec::new();
    let mut applied_migrations = Vec::new();

    for migration in migrations.into_iter() {
        if let Target::Version(input_target) = target {
            if input_target < migration.version() {
                break;
            }
        }

        let query = format!(
            "INSERT INTO refinery_schema_history (version, name, applied_on, checksum) VALUES ({}, '{}', '{}', '{}')",
            migration.version(), migration.name(), Local::now().to_rfc3339(), migration.checksum().to_string()
        );
        let sql = migration.sql().expect("sql must be Some!").to_string();
        applied_migrations.push(migration);
        grouped_migrations.push(sql);
        grouped_migrations.push(query);
    }

    log::info!(
        "going to apply batch migrations in single transaction: {:#?}",
        applied_migrations.iter().map(ToString::to_string)
    );

    if let Target::Version(input_target) = target {
        log::info!("stoping at migration: {}, due to user option", input_target);
    }

    let refs: Vec<&str> = grouped_migrations.iter().map(AsRef::as_ref).collect();

    transaction
        .execute(refs.as_ref())
        .migration_err("error applying migrations", None)?;

    Ok(Report::new(applied_migrations))
}

pub trait Migrate: Query<Vec<Migration>>
where
    Self: Sized,
{
    fn get_last_applied_migration(&mut self) -> Result<Option<Migration>, Error> {
        let mut migrations = self
            .query(GET_LAST_APPLIED_MIGRATION_QUERY)
            .migration_err("error getting last applied migration", None)?;

        Ok(migrations.pop())
    }

    fn get_applied_migrations(&mut self) -> Result<Vec<Migration>, Error> {
        let migrations = self
            .query(GET_APPLIED_MIGRATIONS_QUERY)
            .migration_err("error getting applied migrations", None)?;

        Ok(migrations)
    }

    fn migrate(
        &mut self,
        migrations: &[Migration],
        abort_divergent: bool,
        abort_missing: bool,
        grouped: bool,
        target: Target,
    ) -> Result<Report, Error> {
        self.execute(&[ASSERT_MIGRATIONS_TABLE_QUERY])
            .migration_err("error asserting migrations table", None)?;

        let applied_migrations = self.get_applied_migrations()?;

        let migrations = verify_migrations(
            applied_migrations,
            migrations.to_vec(),
            abort_divergent,
            abort_missing,
        )?;

        if migrations.is_empty() {
            log::info!("no migrations to apply");
        }

        if grouped {
            migrate_grouped(self, migrations, target)
        } else {
            migrate(self, migrations, target)
        }
    }
}