sofie_db/
migrate.rs

1use std::borrow::Cow;
2use std::cmp::Ordering;
3use std::collections::HashMap;
4
5use sofie_core::{Cast, Error, Fnv1a64, Result, SqlType, SqlValue};
6
7use crate::{AggregateTable, Db, Engine, File, Step};
8
9#[cfg(feature = "fresh")]
10mod fresh;
11#[cfg(feature = "migration-status")]
12mod status;
13
14#[cfg(feature = "migration-status")]
15pub use status::{MigrationState, MigrationStatus, ModuleStatus};
16
17#[derive(Clone, Copy, Debug)]
18pub struct Migrations {
19    pub(crate) module: &'static str,
20    files: &'static [File],
21    sqlite: &'static [File],
22    pub(crate) aggregates: &'static [AggregateTable],
23}
24
25impl Migrations {
26    #[must_use]
27    pub const fn new(module: &'static str, files: &'static [File]) -> Self {
28        Self {
29            module,
30            files,
31            sqlite: &[],
32            aggregates: &[],
33        }
34    }
35
36    /// Run on SQLite instead of the files of the same name, where its SQL differs:
37    /// SQLite names a module's tables after it, such as `order_order`.
38    #[must_use]
39    pub const fn sqlite(self, sqlite: &'static [File]) -> Self {
40        Self { sqlite, ..self }
41    }
42
43    #[must_use]
44    pub const fn aggregates(self, aggregates: &'static [AggregateTable]) -> Self {
45        Self { aggregates, ..self }
46    }
47
48    fn in_name_order(&self, engine: Engine) -> Result<Vec<(&'static str, Cow<'static, str>)>> {
49        let mut files = self.files.to_vec();
50        if engine == Engine::Sqlite {
51            files.retain(|file| !self.sqlite.iter().any(|own| own.name() == file.name()));
52            files.extend_from_slice(self.sqlite);
53        }
54        files.sort_unstable_by(|first, second| name_order(first.name(), second.name()));
55        files
56            .into_iter()
57            .map(|file| Ok((file.name(), normalized(file.text()?))))
58            .collect()
59    }
60}
61
62/// `10_x` runs after `9_y`; a name without a number runs after every numbered one.
63pub(crate) fn name_order(first: &str, second: &str) -> Ordering {
64    let key = |name: &str| {
65        let digits = name
66            .find(|character: char| !character.is_ascii_digit())
67            .map_or(name, |end| &name[..end]);
68        let number = digits.parse::<u64>().ok();
69        (number.is_none(), number)
70    };
71    key(first).cmp(&key(second)).then(first.cmp(second))
72}
73
74/// CRLF line ends and a byte-order mark change neither the SQL nor its checksum.
75fn normalized(sql: &'static str) -> Cow<'static, str> {
76    let sql = sql.strip_prefix('\u{feff}').unwrap_or(sql);
77    if sql.contains("\r\n") {
78        return Cow::Owned(sql.replace("\r\n", "\n"));
79    }
80    Cow::Borrowed(sql)
81}
82
83/// Detects an applied file that was edited, not tampering.
84fn checksum(text: &str) -> String {
85    let hash = Fnv1a64::new().write(text.as_bytes()).finish();
86    format!("{hash:016x}")
87}
88
89type Applied = HashMap<(String, String), String>;
90
91#[derive(Clone, Copy, Debug, PartialEq, Eq)]
92pub struct PendingMigration {
93    pub module: &'static str,
94    pub name: &'static str,
95}
96
97impl std::fmt::Display for PendingMigration {
98    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
99        write!(formatter, "{}/{}", self.module, self.name)
100    }
101}
102
103#[derive(Clone, Copy)]
104pub struct Schema<'db> {
105    pub(crate) db: &'db Db,
106}
107
108impl Schema<'_> {
109    /// An applied file that changed is refused; one named before files already
110    /// applied, such as a module's role migration, is applied all the same.
111    pub async fn migrate(self, modules: &[Migrations]) -> Result<()> {
112        let db = self.db;
113        let Some(locking) = db.driver().lock_migrations() else {
114            let each_takes = db.driver().dialect().migrations_lock();
115            return migrate_under(db, modules, Some(each_takes)).await;
116        };
117        let _lock = locking.await?;
118        migrate_under(db, modules, None).await
119    }
120
121    pub async fn first_pending(self, modules: &[Migrations]) -> Result<Option<PendingMigration>> {
122        let db = self.db;
123        let applied = if keeps_records(db).await? {
124            applied_migrations(db).await?
125        } else {
126            Applied::new()
127        };
128        let dialect = db.driver().dialect();
129        let framework = dialect.framework_migrations();
130        for migrations in std::iter::once(&framework).chain(modules) {
131            for (name, sql) in migrations.in_name_order(dialect.engine())? {
132                let module = migrations.module;
133                if !is_applied(&applied, module, name, &checksum(&sql))? {
134                    return Ok(Some(PendingMigration { module, name }));
135                }
136            }
137        }
138        Ok(None)
139    }
140}
141
142async fn keeps_records(db: &Db) -> Result<bool> {
143    let Some(catalog) = db.driver().catalog() else {
144        return Ok(true);
145    };
146    Ok(!catalog.columns("sofie", "migrations").await?.is_empty())
147}
148
149async fn migrate_under(db: &Db, modules: &[Migrations], lock: Option<&str>) -> Result<()> {
150    let dialect = db.driver().dialect();
151    let framework = dialect.framework_migrations();
152    let setup = Step::Script(dialect.migrations_setup().to_owned());
153    db.transaction(locked(lock, vec![setup])).await?;
154    let applied = applied_migrations(db).await?;
155    for migrations in std::iter::once(&framework).chain(modules) {
156        for (name, sql) in migrations.in_name_order(dialect.engine())? {
157            apply_unless_applied(db, &applied, migrations.module, name, &sql, lock).await?;
158        }
159    }
160    Ok(())
161}
162
163fn locked(lock: Option<&str>, steps: Vec<Step>) -> Vec<Step> {
164    lock.map(|sql| Step::Script(sql.to_owned()))
165        .into_iter()
166        .chain(steps)
167        .collect()
168}
169
170async fn apply_unless_applied(
171    db: &Db,
172    applied: &Applied,
173    module: &str,
174    name: &str,
175    sql: &str,
176    lock: Option<&str>,
177) -> Result<()> {
178    let sum = checksum(sql);
179    if is_applied(applied, module, name, &sum)? {
180        return Ok(());
181    }
182    let Err(error) = apply(db, module, name, sql, &sum, lock).await else {
183        return Ok(());
184    };
185    // Its record explains the failure if another process applied it meanwhile.
186    let now = applied_migrations(db).await?;
187    if is_applied(&now, module, name, &sum)? {
188        return Ok(());
189    }
190    Err(error)
191}
192
193fn record_table(db: &Db) -> String {
194    db.driver().dialect().qualified("sofie", "migrations")
195}
196
197async fn applied_migrations(db: &Db) -> Result<Applied> {
198    let sql = format!("SELECT module, name, checksum FROM {}", record_table(db));
199    let types = vec![Some(SqlType::Text); 3];
200    let applied = crate::query_decoded(db.driver(), &sql, &[], types, |row| {
201        let key = (String::decode(row)?, String::decode(row)?);
202        Ok((key, String::decode(row)?))
203    })
204    .await?;
205    Ok(applied.into_iter().collect())
206}
207
208async fn apply(
209    db: &Db,
210    module: &str,
211    name: &str,
212    sql: &str,
213    sum: &str,
214    lock: Option<&str>,
215) -> Result<()> {
216    let dialect = db.driver().dialect();
217    let insert = format!(
218        "INSERT INTO {} (module, name, checksum) VALUES ({}, {}, {})",
219        record_table(db),
220        dialect.placeholder(1),
221        dialect.placeholder(2),
222        dialect.placeholder(3)
223    );
224    let params = [module, name, sum].map(|text| SqlValue::Text(text.to_owned()));
225    let record = Step::statement(insert, params.to_vec(), None);
226    let failed = |error| Error::internal(format!("migration {module}/{name}"), error);
227    #[cfg(feature = "concurrent-indexes")]
228    let record = match lock {
229        None => match db.driver().migrate_alone(module, sql, record) {
230            Ok(running) => return running.await.map_err(failed),
231            Err(record) => record,
232        },
233        Some(_) => record,
234    };
235    let steps = vec![
236        record,
237        Step::Script(dialect.enter_schema(module)),
238        Step::Script(sql.to_owned()),
239    ];
240    db.transaction(locked(lock, steps)).await.map_err(failed)
241}
242
243fn is_applied(applied: &Applied, module: &str, name: &str, sum: &str) -> Result<bool> {
244    match applied.get(&(module.to_owned(), name.to_owned())) {
245        None => Ok(false),
246        Some(recorded) if recorded == sum => Ok(true),
247        Some(_) => Err(Error::internal_message(format!(
248            "migration {module}/{name} changed after it was applied"
249        ))),
250    }
251}
252
253#[cfg(test)]
254mod tests {
255    use super::*;
256
257    fn names_and_texts(
258        files: Vec<(&'static str, Cow<'static, str>)>,
259    ) -> Vec<(&'static str, String)> {
260        files
261            .into_iter()
262            .map(|(name, sql)| (name, sql.into_owned()))
263            .collect()
264    }
265
266    #[test]
267    fn sqlite_runs_its_own_file_where_a_module_has_one() {
268        const MIGRATIONS: Migrations = Migrations::new(
269            "order",
270            &[
271                File::new("0001_a.sql", b"pg a"),
272                File::new("0002_b.sql", b"pg b"),
273            ],
274        )
275        .sqlite(&[File::new("sqlite/0001_a.sql", b"lite a")]);
276
277        assert_eq!(
278            names_and_texts(MIGRATIONS.in_name_order(Engine::Sqlite).unwrap()),
279            [
280                ("0001_a.sql", "lite a".to_owned()),
281                ("0002_b.sql", "pg b".to_owned())
282            ]
283        );
284        assert_eq!(
285            names_and_texts(MIGRATIONS.in_name_order(Engine::Postgres).unwrap()),
286            [
287                ("0001_a.sql", "pg a".to_owned()),
288                ("0002_b.sql", "pg b".to_owned())
289            ]
290        );
291    }
292
293    #[test]
294    fn migrations_run_in_the_order_of_their_numbers() {
295        const MIGRATIONS: Migrations = Migrations::new(
296            "order",
297            &[
298                File::new("10_c.sql", b""),
299                File::new("notes.sql", b""),
300                File::new("9_b.sql", b""),
301                File::new("1_a.sql", b""),
302            ],
303        );
304
305        let names: Vec<&str> = MIGRATIONS
306            .in_name_order(Engine::Postgres)
307            .unwrap()
308            .into_iter()
309            .map(|(name, _)| name)
310            .collect();
311
312        assert_eq!(names, ["1_a.sql", "9_b.sql", "10_c.sql", "notes.sql"]);
313    }
314
315    #[test]
316    fn a_byte_order_mark_and_crlf_line_ends_change_neither_the_sql_nor_its_checksum() {
317        const MIGRATIONS: Migrations = Migrations::new(
318            "order",
319            &[
320                File::new("0001_lf.sql", b"CREATE TABLE a ();\nCREATE TABLE b ();\n"),
321                File::new(
322                    "0002_crlf.sql",
323                    "\u{feff}CREATE TABLE a ();\r\nCREATE TABLE b ();\r\n".as_bytes(),
324                ),
325            ],
326        );
327
328        let files = MIGRATIONS.in_name_order(Engine::Postgres).unwrap();
329
330        assert_eq!(files[1].1, files[0].1);
331        assert_eq!(checksum(&files[1].1), checksum(&files[0].1));
332    }
333
334    #[test]
335    fn checksum_changes_with_the_text_and_is_fnv_1a_of_it() {
336        assert_eq!(checksum("create table a ()"), checksum("create table a ()"));
337        assert_ne!(checksum("create table a ()"), checksum("create table b ()"));
338        assert_eq!(checksum(""), "cbf29ce484222325");
339    }
340}
sofie_db/migrate.rs

sofie_db/
migrate.rs
