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)]
92pub struct Schema<'db> {
93    pub(crate) db: &'db Db,
94}
95
96impl Schema<'_> {
97    /// An applied file that changed is refused; one named before files already
98    /// applied, such as a module's role migration, is applied all the same.
99    pub async fn migrate(self, modules: &[Migrations]) -> Result<()> {
100        let db = self.db;
101        let Some(locking) = db.driver().lock_migrations() else {
102            let each_takes = db.driver().dialect().migrations_lock();
103            return migrate_under(db, modules, Some(each_takes)).await;
104        };
105        let _lock = locking.await?;
106        migrate_under(db, modules, None).await
107    }
108}
109
110async fn migrate_under(db: &Db, modules: &[Migrations], lock: Option<&str>) -> Result<()> {
111    let dialect = db.driver().dialect();
112    let framework = dialect.framework_migrations();
113    let setup = Step::Script(dialect.migrations_setup().to_owned());
114    db.transaction(locked(lock, vec![setup])).await?;
115    let applied = applied_migrations(db).await?;
116    for migrations in std::iter::once(&framework).chain(modules) {
117        for (name, sql) in migrations.in_name_order(dialect.engine())? {
118            apply_unless_applied(db, &applied, migrations.module, name, &sql, lock).await?;
119        }
120    }
121    Ok(())
122}
123
124fn locked(lock: Option<&str>, steps: Vec<Step>) -> Vec<Step> {
125    lock.map(|sql| Step::Script(sql.to_owned()))
126        .into_iter()
127        .chain(steps)
128        .collect()
129}
130
131async fn apply_unless_applied(
132    db: &Db,
133    applied: &Applied,
134    module: &str,
135    name: &str,
136    sql: &str,
137    lock: Option<&str>,
138) -> Result<()> {
139    let sum = checksum(sql);
140    if is_applied(applied, module, name, &sum)? {
141        return Ok(());
142    }
143    let Err(error) = apply(db, module, name, sql, &sum, lock).await else {
144        return Ok(());
145    };
146    // Its record explains the failure if another process applied it meanwhile.
147    let now = applied_migrations(db).await?;
148    if is_applied(&now, module, name, &sum)? {
149        return Ok(());
150    }
151    Err(error)
152}
153
154fn record_table(db: &Db) -> String {
155    db.driver().dialect().qualified("sofie", "migrations")
156}
157
158async fn applied_migrations(db: &Db) -> Result<Applied> {
159    let sql = format!("SELECT module, name, checksum FROM {}", record_table(db));
160    let types = vec![Some(SqlType::Text); 3];
161    let applied = crate::query_decoded(db.driver(), &sql, &[], types, |row| {
162        let key = (String::decode(row)?, String::decode(row)?);
163        Ok((key, String::decode(row)?))
164    })
165    .await?;
166    Ok(applied.into_iter().collect())
167}
168
169async fn apply(
170    db: &Db,
171    module: &str,
172    name: &str,
173    sql: &str,
174    sum: &str,
175    lock: Option<&str>,
176) -> Result<()> {
177    let dialect = db.driver().dialect();
178    let insert = format!(
179        "INSERT INTO {} (module, name, checksum) VALUES ({}, {}, {})",
180        record_table(db),
181        dialect.placeholder(1),
182        dialect.placeholder(2),
183        dialect.placeholder(3)
184    );
185    let params = [module, name, sum].map(|text| SqlValue::Text(text.to_owned()));
186    let record = Step::statement(insert, params.to_vec(), None);
187    let failed = |error| Error::internal(format!("migration {module}/{name}"), error);
188    #[cfg(feature = "concurrent-indexes")]
189    let record = match lock {
190        None => match db.driver().migrate_alone(module, sql, record) {
191            Ok(running) => return running.await.map_err(failed),
192            Err(record) => record,
193        },
194        Some(_) => record,
195    };
196    let steps = vec![
197        record,
198        Step::Script(dialect.enter_schema(module)),
199        Step::Script(sql.to_owned()),
200    ];
201    db.transaction(locked(lock, steps)).await.map_err(failed)
202}
203
204fn is_applied(applied: &Applied, module: &str, name: &str, sum: &str) -> Result<bool> {
205    match applied.get(&(module.to_owned(), name.to_owned())) {
206        None => Ok(false),
207        Some(recorded) if recorded == sum => Ok(true),
208        Some(_) => Err(Error::internal_message(format!(
209            "migration {module}/{name} changed after it was applied"
210        ))),
211    }
212}
213
214#[cfg(test)]
215mod tests {
216    use super::*;
217
218    fn names_and_texts(
219        files: Vec<(&'static str, Cow<'static, str>)>,
220    ) -> Vec<(&'static str, String)> {
221        files
222            .into_iter()
223            .map(|(name, sql)| (name, sql.into_owned()))
224            .collect()
225    }
226
227    #[test]
228    fn sqlite_runs_its_own_file_where_a_module_has_one() {
229        const MIGRATIONS: Migrations = Migrations::new(
230            "order",
231            &[
232                File::new("0001_a.sql", b"pg a"),
233                File::new("0002_b.sql", b"pg b"),
234            ],
235        )
236        .sqlite(&[File::new("sqlite/0001_a.sql", b"lite a")]);
237
238        assert_eq!(
239            names_and_texts(MIGRATIONS.in_name_order(Engine::Sqlite).unwrap()),
240            [
241                ("0001_a.sql", "lite a".to_owned()),
242                ("0002_b.sql", "pg b".to_owned())
243            ]
244        );
245        assert_eq!(
246            names_and_texts(MIGRATIONS.in_name_order(Engine::Postgres).unwrap()),
247            [
248                ("0001_a.sql", "pg a".to_owned()),
249                ("0002_b.sql", "pg b".to_owned())
250            ]
251        );
252    }
253
254    #[test]
255    fn migrations_run_in_the_order_of_their_numbers() {
256        const MIGRATIONS: Migrations = Migrations::new(
257            "order",
258            &[
259                File::new("10_c.sql", b""),
260                File::new("notes.sql", b""),
261                File::new("9_b.sql", b""),
262                File::new("1_a.sql", b""),
263            ],
264        );
265
266        let names: Vec<&str> = MIGRATIONS
267            .in_name_order(Engine::Postgres)
268            .unwrap()
269            .into_iter()
270            .map(|(name, _)| name)
271            .collect();
272
273        assert_eq!(names, ["1_a.sql", "9_b.sql", "10_c.sql", "notes.sql"]);
274    }
275
276    #[test]
277    fn a_byte_order_mark_and_crlf_line_ends_change_neither_the_sql_nor_its_checksum() {
278        const MIGRATIONS: Migrations = Migrations::new(
279            "order",
280            &[
281                File::new("0001_lf.sql", b"CREATE TABLE a ();\nCREATE TABLE b ();\n"),
282                File::new(
283                    "0002_crlf.sql",
284                    "\u{feff}CREATE TABLE a ();\r\nCREATE TABLE b ();\r\n".as_bytes(),
285                ),
286            ],
287        );
288
289        let files = MIGRATIONS.in_name_order(Engine::Postgres).unwrap();
290
291        assert_eq!(files[1].1, files[0].1);
292        assert_eq!(checksum(&files[1].1), checksum(&files[0].1));
293    }
294
295    #[test]
296    fn checksum_changes_with_the_text_and_is_fnv_1a_of_it() {
297        assert_eq!(checksum("create table a ()"), checksum("create table a ()"));
298        assert_ne!(checksum("create table a ()"), checksum("create table b ()"));
299        assert_eq!(checksum(""), "cbf29ce484222325");
300    }
301}
sofie_db/migrate.rs

sofie_db/
migrate.rs
