a3s_orm/drivers/sqlite/
migration.rs

1use async_trait::async_trait;
2use tokio_rusqlite::rusqlite;
3
4use crate::{
5    pending_migrations, AppliedMigration, MigrationBackend, MigrationReport, PreparedMigration,
6};
7
8use super::{SqliteExecutor, SqliteMigrationError};
9
10const CREATE_TABLE: &str = "
11    create table if not exists a3s_orm_migrations (
12        version text primary key,
13        name text not null,
14        checksum text not null,
15        applied_at text not null default current_timestamp
16    )";
17
18#[async_trait]
19impl MigrationBackend for SqliteExecutor {
20    type Error = SqliteMigrationError;
21
22    async fn apply(
23        &self,
24        migrations: &[PreparedMigration],
25    ) -> Result<MigrationReport, Self::Error> {
26        let _guard = self.transaction_lock.lock().await;
27        let migrations = migrations.to_vec();
28        let outcome = self
29            .connection
30            .call(move |connection| {
31                connection.execute_batch("BEGIN IMMEDIATE")?;
32                let result = migrate(connection, &migrations);
33                match result {
34                    Ok(Ok(report)) => {
35                        connection.execute_batch("COMMIT")?;
36                        Ok(Ok(report))
37                    }
38                    Ok(Err(error)) => {
39                        let _ = connection.execute_batch("ROLLBACK");
40                        Ok(Err(error))
41                    }
42                    Err(error) => {
43                        let _ = connection.execute_batch("ROLLBACK");
44                        Err(error)
45                    }
46                }
47            })
48            .await
49            .map_err(crate::SqliteError::from)?;
50        outcome.map_err(|error| match error {
51            MigrationFailure::Validation(error) => SqliteMigrationError::Migration(error),
52            MigrationFailure::Apply { version, source } => {
53                SqliteMigrationError::Apply { version, source }
54            }
55        })
56    }
57}
58
59enum MigrationFailure {
60    Validation(crate::MigrationError),
61    Apply {
62        version: String,
63        source: rusqlite::Error,
64    },
65}
66
67fn migrate(
68    connection: &rusqlite::Connection,
69    migrations: &[PreparedMigration],
70) -> rusqlite::Result<Result<MigrationReport, MigrationFailure>> {
71    connection.execute_batch(CREATE_TABLE)?;
72    let mut statement =
73        connection.prepare("select version, checksum from a3s_orm_migrations order by version")?;
74    let applied = statement
75        .query_map([], |row| {
76            Ok(AppliedMigration {
77                version: row.get(0)?,
78                checksum: row.get(1)?,
79            })
80        })?
81        .collect::<rusqlite::Result<Vec<_>>>()?;
82    drop(statement);
83    let pending = match pending_migrations(&applied, migrations) {
84        Ok(pending) => pending,
85        Err(error) => return Ok(Err(MigrationFailure::Validation(error))),
86    };
87    let mut versions = Vec::with_capacity(pending.len());
88    for migration in pending {
89        if let Err(source) = connection.execute_batch(migration.up_sql()) {
90            return Ok(Err(MigrationFailure::Apply {
91                version: migration.version().to_owned(),
92                source,
93            }));
94        }
95        connection.execute(
96            "insert into a3s_orm_migrations (version, name, checksum) values (?1, ?2, ?3)",
97            rusqlite::params![migration.version(), migration.name(), migration.checksum()],
98        )?;
99        versions.push(migration.version().to_owned());
100    }
101    Ok(Ok(MigrationReport { applied: versions }))
102}
a3s_orm/drivers/sqlite/migration.rs

a3s_orm/drivers/sqlite/
migration.rs
