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, _) in migrations.in_name_order(dialect.engine())? {
132                let key = (migrations.module.to_owned(), name.to_owned());
133                if !applied.contains_key(&key) {
134                    let module = migrations.module;
135                    return Ok(Some(PendingMigration { module, name }));
136                }
137            }
138        }
139        Ok(None)
140    }
141}
142
143async fn keeps_records(db: &Db) -> Result<bool> {
144    let Some(catalog) = db.driver().catalog() else {
145        return Ok(true);
146    };
147    Ok(!catalog.columns("sofie", "migrations").await?.is_empty())
148}
149
150async fn migrate_under(db: &Db, modules: &[Migrations], lock: Option<&str>) -> Result<()> {
151    let dialect = db.driver().dialect();
152    let framework = dialect.framework_migrations();
153    let setup = Step::Script(dialect.migrations_setup().to_owned());
154    db.transaction(locked(lock, vec![setup])).await?;
155    let applied = applied_migrations(db).await?;
156    for migrations in std::iter::once(&framework).chain(modules) {
157        for (name, sql) in migrations.in_name_order(dialect.engine())? {
158            apply_unless_applied(db, &applied, migrations.module, name, &sql, lock).await?;
159        }
160    }
161    Ok(())
162}
163
164fn locked(lock: Option<&str>, steps: Vec<Step>) -> Vec<Step> {
165    lock.map(|sql| Step::Script(sql.to_owned()))
166        .into_iter()
167        .chain(steps)
168        .collect()
169}
170
171async fn apply_unless_applied(
172    db: &Db,
173    applied: &Applied,
174    module: &str,
175    name: &str,
176    sql: &str,
177    lock: Option<&str>,
178) -> Result<()> {
179    let sum = checksum(sql);
180    if is_applied(applied, module, name, &sum)? {
181        return Ok(());
182    }
183    let Err(error) = apply(db, module, name, sql, &sum, lock).await else {
184        return Ok(());
185    };
186    // Its record explains the failure if another process applied it meanwhile.
187    let now = applied_migrations(db).await?;
188    if is_applied(&now, module, name, &sum)? {
189        return Ok(());
190    }
191    Err(error)
192}
193
194fn record_table(db: &Db) -> String {
195    db.driver().dialect().qualified("sofie", "migrations")
196}
197
198async fn applied_migrations(db: &Db) -> Result<Applied> {
199    let sql = format!("SELECT module, name, checksum FROM {}", record_table(db));
200    let types = vec![Some(SqlType::Text); 3];
201    let applied = crate::query_decoded(db.driver(), &sql, &[], types, |row| {
202        let key = (String::decode(row)?, String::decode(row)?);
203        Ok((key, String::decode(row)?))
204    })
205    .await?;
206    Ok(applied.into_iter().collect())
207}
208
209async fn apply(
210    db: &Db,
211    module: &str,
212    name: &str,
213    sql: &str,
214    sum: &str,
215    lock: Option<&str>,
216) -> Result<()> {
217    let dialect = db.driver().dialect();
218    let insert = format!(
219        "INSERT INTO {} (module, name, checksum) VALUES ({}, {}, {})",
220        record_table(db),
221        dialect.placeholder(1),
222        dialect.placeholder(2),
223        dialect.placeholder(3)
224    );
225    let params = [module, name, sum].map(|text| SqlValue::Text(text.to_owned()));
226    let record = Step::statement(insert, params.to_vec(), None);
227    let failed = |error| Error::internal(format!("migration {module}/{name}"), error);
228    #[cfg(feature = "concurrent-indexes")]
229    let record = match lock {
230        None => match db.driver().migrate_alone(module, sql, record) {
231            Ok(running) => return running.await.map_err(failed),
232            Err(record) => record,
233        },
234        Some(_) => record,
235    };
236    let steps = vec![
237        record,
238        Step::Script(dialect.enter_schema(module)),
239        Step::Script(sql.to_owned()),
240    ];
241    db.transaction(locked(lock, steps)).await.map_err(failed)
242}
243
244fn is_applied(applied: &Applied, module: &str, name: &str, sum: &str) -> Result<bool> {
245    match applied.get(&(module.to_owned(), name.to_owned())) {
246        None => Ok(false),
247        Some(recorded) if recorded == sum => Ok(true),
248        Some(_) => Err(Error::internal_message(format!(
249            "migration {module}/{name} changed after it was applied"
250        ))),
251    }
252}
253
254#[cfg(test)]
255mod tests {
256    use super::*;
257
258    fn names_and_texts(
259        files: Vec<(&'static str, Cow<'static, str>)>,
260    ) -> Vec<(&'static str, String)> {
261        files
262            .into_iter()
263            .map(|(name, sql)| (name, sql.into_owned()))
264            .collect()
265    }
266
267    #[test]
268    fn sqlite_runs_its_own_file_where_a_module_has_one() {
269        const MIGRATIONS: Migrations = Migrations::new(
270            "order",
271            &[
272                File::new("0001_a.sql", b"pg a"),
273                File::new("0002_b.sql", b"pg b"),
274            ],
275        )
276        .sqlite(&[File::new("sqlite/0001_a.sql", b"lite a")]);
277
278        assert_eq!(
279            names_and_texts(MIGRATIONS.in_name_order(Engine::Sqlite).unwrap()),
280            [
281                ("0001_a.sql", "lite a".to_owned()),
282                ("0002_b.sql", "pg b".to_owned())
283            ]
284        );
285        assert_eq!(
286            names_and_texts(MIGRATIONS.in_name_order(Engine::Postgres).unwrap()),
287            [
288                ("0001_a.sql", "pg a".to_owned()),
289                ("0002_b.sql", "pg b".to_owned())
290            ]
291        );
292    }
293
294    #[test]
295    fn migrations_run_in_the_order_of_their_numbers() {
296        const MIGRATIONS: Migrations = Migrations::new(
297            "order",
298            &[
299                File::new("10_c.sql", b""),
300                File::new("notes.sql", b""),
301                File::new("9_b.sql", b""),
302                File::new("1_a.sql", b""),
303            ],
304        );
305
306        let names: Vec<&str> = MIGRATIONS
307            .in_name_order(Engine::Postgres)
308            .unwrap()
309            .into_iter()
310            .map(|(name, _)| name)
311            .collect();
312
313        assert_eq!(names, ["1_a.sql", "9_b.sql", "10_c.sql", "notes.sql"]);
314    }
315
316    #[test]
317    fn a_byte_order_mark_and_crlf_line_ends_change_neither_the_sql_nor_its_checksum() {
318        const MIGRATIONS: Migrations = Migrations::new(
319            "order",
320            &[
321                File::new("0001_lf.sql", b"CREATE TABLE a ();\nCREATE TABLE b ();\n"),
322                File::new(
323                    "0002_crlf.sql",
324                    "\u{feff}CREATE TABLE a ();\r\nCREATE TABLE b ();\r\n".as_bytes(),
325                ),
326            ],
327        );
328
329        let files = MIGRATIONS.in_name_order(Engine::Postgres).unwrap();
330
331        assert_eq!(files[1].1, files[0].1);
332        assert_eq!(checksum(&files[1].1), checksum(&files[0].1));
333    }
334
335    #[test]
336    fn checksum_changes_with_the_text_and_is_fnv_1a_of_it() {
337        assert_eq!(checksum("create table a ()"), checksum("create table a ()"));
338        assert_ne!(checksum("create table a ()"), checksum("create table b ()"));
339        assert_eq!(checksum(""), "cbf29ce484222325");
340    }
341}
sofie_db/migrate.rs

sofie_db/
migrate.rs
