a3s_orm/migration/
runner.rs

1use std::collections::HashSet;
2
3use super::{
4    AppliedMigration, Migration, MigrationBackend, MigrationError, MigrationLedger,
5    MigrationRunError, PreparedMigration,
6};
7
8#[derive(Clone, Debug, Default, PartialEq, Eq)]
9pub struct MigrationReport {
10    pub applied: Vec<String>,
11}
12
13impl MigrationReport {
14    pub fn is_up_to_date(&self) -> bool {
15        self.applied.is_empty()
16    }
17}
18
19pub struct Migrator<B> {
20    backend: B,
21}
22
23impl<B> Migrator<B> {
24    pub const fn new(backend: B) -> Self {
25        Self { backend }
26    }
27
28    pub fn backend(&self) -> &B {
29        &self.backend
30    }
31}
32
33impl<B: MigrationBackend> Migrator<B> {
34    pub async fn run(
35        &self,
36        migrations: impl IntoIterator<Item = Migration>,
37    ) -> Result<MigrationReport, MigrationRunError<B::Error>> {
38        let migrations = prepare(migrations)?;
39        self.backend
40            .apply(&migrations)
41            .await
42            .map_err(MigrationRunError::Backend)
43    }
44}
45
46impl<B: MigrationLedger> Migrator<B> {
47    /// Verify that every supplied migration is present with its exact
48    /// checksum, without locking or mutating the database.
49    ///
50    /// Additional database migrations are admitted so an older serving
51    /// process can run during an expand-compatible rolling upgrade. Callers
52    /// remain responsible for deciding which schema versions are compatible.
53    pub async fn verify_required(
54        &self,
55        migrations: impl IntoIterator<Item = Migration>,
56    ) -> Result<(), MigrationRunError<B::Error>> {
57        let required = prepare(migrations)?;
58        let applied = self
59            .backend
60            .applied_migrations()
61            .await
62            .map_err(MigrationRunError::Backend)?;
63        verify_required(&applied, &required)?;
64        Ok(())
65    }
66}
67
68fn verify_required(
69    applied: &[AppliedMigration],
70    required: &[PreparedMigration],
71) -> Result<(), MigrationError> {
72    for required_migration in required {
73        let Some(applied_migration) = applied
74            .iter()
75            .find(|migration| migration.version == required_migration.version())
76        else {
77            return Err(MigrationError::MissingAppliedMigration(
78                required_migration.version().to_owned(),
79            ));
80        };
81        if applied_migration.checksum != required_migration.checksum() {
82            return Err(MigrationError::ChecksumMismatch {
83                version: required_migration.version().to_owned(),
84                applied_checksum: applied_migration.checksum.clone(),
85                source_checksum: required_migration.checksum().to_owned(),
86            });
87        }
88    }
89    Ok(())
90}
91
92fn prepare(
93    migrations: impl IntoIterator<Item = Migration>,
94) -> Result<Vec<PreparedMigration>, MigrationError> {
95    let mut migrations = migrations.into_iter().collect::<Vec<_>>();
96    for migration in &migrations {
97        validate(migration)?;
98    }
99    migrations.sort_by(|left, right| left.version().cmp(right.version()));
100    let mut versions = HashSet::with_capacity(migrations.len());
101    for migration in &migrations {
102        if !versions.insert(migration.version().to_owned()) {
103            return Err(MigrationError::DuplicateVersion(
104                migration.version().to_owned(),
105            ));
106        }
107    }
108    Ok(migrations
109        .into_iter()
110        .map(PreparedMigration::prepare)
111        .collect())
112}
113
114fn validate(migration: &Migration) -> Result<(), MigrationError> {
115    if migration.version().is_empty() {
116        return Err(MigrationError::EmptyVersion);
117    }
118    if !migration
119        .version()
120        .bytes()
121        .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b'-'))
122    {
123        return Err(MigrationError::InvalidVersion(
124            migration.version().to_owned(),
125        ));
126    }
127    if migration.name().trim().is_empty() {
128        return Err(MigrationError::EmptyName {
129            version: migration.version().to_owned(),
130        });
131    }
132    if migration.up_sql().trim().is_empty() {
133        return Err(MigrationError::EmptySql {
134            version: migration.version().to_owned(),
135        });
136    }
137    Ok(())
138}
a3s_orm/migration/runner.rs

a3s_orm/migration/
runner.rs
