a3s_orm/drivers/sqlite/
migration.rs

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

a3s_orm/drivers/sqlite/
migration.rs
