datumesh_store/
migrate.rs

1//! Schema migrations: plain SQL files (`migrations/*.sql`, embedded), applied in order with
2//! a tracking table, and a schema check that compares the live database's catalog with the
3//! catalog the migrations produce in a scratch database.
4//!
5//! Each function works through a [`PgStore`], which connects to an empty database without
6//! touching it: `migrate::apply(&PgStore::connect(url)?)` creates the schema. The SQLite
7//! backend's migrations are [`sqlite`] (since 0.4.0):
8//! `migrate::sqlite::apply(&SqliteStore::open(path)?)`.
9
10use postgres::Client;
11
12use crate::error::{Db, Error, Result};
13use crate::PgStore;
14
15/// One migration.
16#[derive(Clone, Copy, Debug, PartialEq, Eq)]
17pub struct Migration {
18    /// Its number; applied in increasing order.
19    pub version: u32,
20    /// Its file name.
21    pub name: &'static str,
22    /// Its SQL.
23    pub sql: &'static str,
24}
25
26impl Migration {
27    /// SHA-256 of the SQL, recorded when applied: a migration edited after it was applied is
28    /// refused.
29    pub fn checksum(&self) -> String {
30        datumesh_core::sha256_hex(self.sql.as_bytes())
31    }
32}
33
34/// Every migration, in order.
35pub const MIGRATIONS: &[Migration] = &[
36    Migration { version: 1, name: "0001_initial.sql", sql: include_str!("../migrations/0001_initial.sql") },
37    Migration {
38        version: 2,
39        name: "0002_computed_versions.sql",
40        sql: include_str!("../migrations/0002_computed_versions.sql"),
41    },
42    Migration { version: 3, name: "0003_slices.sql", sql: include_str!("../migrations/0003_slices.sql") },
43    Migration {
44        version: 4,
45        name: "0004_receipt_values.sql",
46        sql: include_str!("../migrations/0004_receipt_values.sql"),
47    },
48];
49
50const TRACKING: &str = "CREATE TABLE IF NOT EXISTS datumesh_schema_migrations (
51    version integer PRIMARY KEY,
52    name text NOT NULL,
53    checksum text NOT NULL,
54    applied_at timestamptz NOT NULL DEFAULT now()
55)";
56
57/// Serialises migrators of one database (a session-level advisory lock).
58const LOCK: &str = "SELECT pg_advisory_lock(hashtextextended('datumesh-store:migrate', 0))";
59const UNLOCK: &str = "SELECT pg_advisory_unlock(hashtextextended('datumesh-store:migrate', 0))";
60
61/// Applies the migrations not yet applied, each in its own transaction, and returns their
62/// versions (empty when the schema is current). Concurrent callers are serialised (a
63/// session-level advisory lock); a migration whose recorded checksum differs from its SQL is
64/// an error. Holds one of the store's pooled connections throughout, so the lock and the
65/// tracking table's transactions are on one session.
66pub fn apply(store: &PgStore) -> Result<Vec<u32>> {
67    let mut conn = store.conn()?;
68    apply_client(&mut conn.client)
69}
70
71fn apply_client(client: &mut Client) -> Result<Vec<u32>> {
72    client.batch_execute(LOCK).db()?;
73    let result = apply_locked(client);
74    let unlocked = client.batch_execute(UNLOCK).db();
75    let applied = result?;
76    unlocked?;
77    for v in &applied {
78        tracing::info!(target: "datumesh_store", version = v, "migration applied");
79    }
80    Ok(applied)
81}
82
83fn apply_locked(client: &mut Client) -> Result<Vec<u32>> {
84    client.batch_execute(TRACKING).db()?;
85    let rows = client.query("SELECT version, checksum FROM datumesh_schema_migrations", &[]).db()?;
86    let mut done = Vec::new();
87    for r in &rows {
88        let v: i32 = r.try_get(0).db()?;
89        let c: String = r.try_get(1).db()?;
90        done.push((v, c));
91    }
92    let mut applied = Vec::new();
93    for m in MIGRATIONS {
94        let version =
95            i32::try_from(m.version).map_err(|_| Error::Migration(format!("{}: version too large", m.name)))?;
96        match done.iter().find(|(v, _)| *v == version) {
97            Some((_, c)) if *c == m.checksum() => continue,
98            Some(_) => {
99                return Err(Error::Migration(format!(
100                    "{} was applied with another checksum: the file changed after it was applied",
101                    m.name
102                )))
103            }
104            None => {}
105        }
106        let mut t = client.transaction().db()?;
107        t.batch_execute(m.sql).db()?;
108        t.execute(
109            "INSERT INTO datumesh_schema_migrations (version, name, checksum) VALUES ($1, $2, $3)",
110            &[&version, &m.name, &m.checksum()],
111        )
112        .db()?;
113        t.commit().db()?;
114        applied.push(m.version);
115    }
116    if let Some((v, _)) = done.iter().find(|(v, _)| !MIGRATIONS.iter().any(|m| i64::from(m.version) == i64::from(*v))) {
117        return Err(Error::Migration(format!("the database has migration {v}, which this version does not know")));
118    }
119    Ok(applied)
120}
121
122/// The catalog of the current schema as sorted lines: relations, columns (position, type,
123/// nullability, default, collation, generation), constraints, indexes, triggers (with their
124/// enabled state), functions (a hash of their definition), and the recorded migrations.
125pub fn catalog(store: &PgStore) -> Result<Vec<String>> {
126    catalog_client(&mut store.conn()?.client)
127}
128
129fn catalog_client(client: &mut Client) -> Result<Vec<String>> {
130    let q = r#"
131WITH ns AS (SELECT oid FROM pg_namespace WHERE nspname = current_schema())
132SELECT 'relation ' || c.relkind::text || ' ' || c.relname
133  FROM pg_class c WHERE c.relnamespace = (SELECT oid FROM ns) AND c.relkind IN ('r', 'p', 'v', 'm', 'S', 'f')
134UNION ALL
135SELECT 'column ' || c.relname || '.' || a.attname || ' #' || a.attnum || ' ' || format_type(a.atttypid, a.atttypmod)
136       || CASE WHEN a.attnotnull THEN ' not null' ELSE '' END
137       || coalesce(' default ' || pg_get_expr(d.adbin, d.adrelid), '')
138       || coalesce(' collate ' || co.collname, '')
139       || CASE WHEN a.attgenerated::text <> '' THEN ' generated ' || a.attgenerated::text ELSE '' END
140       || CASE WHEN a.attidentity::text <> '' THEN ' identity ' || a.attidentity::text ELSE '' END
141  FROM pg_attribute a
142  JOIN pg_class c ON c.oid = a.attrelid
143  LEFT JOIN pg_attrdef d ON d.adrelid = a.attrelid AND d.adnum = a.attnum
144  LEFT JOIN pg_collation co ON co.oid = a.attcollation AND a.attcollation <> 0
145       AND co.oid <> (SELECT oid FROM pg_collation WHERE collname = 'default')
146 WHERE c.relnamespace = (SELECT oid FROM ns) AND c.relkind IN ('r', 'p', 'v', 'm', 'f')
147   AND a.attnum > 0 AND NOT a.attisdropped
148UNION ALL
149SELECT 'constraint ' || c.relname || '.' || k.conname || ' ' || pg_get_constraintdef(k.oid)
150  FROM pg_constraint k JOIN pg_class c ON c.oid = k.conrelid
151 WHERE k.connamespace = (SELECT oid FROM ns)
152UNION ALL
153SELECT 'index ' || c.relname || ' ' || pg_get_indexdef(i.indexrelid)
154  FROM pg_index i JOIN pg_class c ON c.oid = i.indexrelid
155 WHERE c.relnamespace = (SELECT oid FROM ns)
156UNION ALL
157SELECT 'trigger ' || c.relname || '.' || t.tgname || ' enabled ' || t.tgenabled::text || ' ' || pg_get_triggerdef(t.oid)
158  FROM pg_trigger t JOIN pg_class c ON c.oid = t.tgrelid
159 WHERE c.relnamespace = (SELECT oid FROM ns) AND NOT t.tgisinternal
160UNION ALL
161SELECT 'function ' || p.oid::regprocedure::text || ' ' || md5(pg_get_functiondef(p.oid))
162  FROM pg_proc p WHERE p.pronamespace = (SELECT oid FROM ns) AND p.prokind IN ('f', 'p')
163"#;
164    let mut lines = Vec::new();
165    for r in client.query(q, &[]).db()? {
166        lines.push(r.try_get::<_, String>(0).db()?);
167    }
168    let tracked: bool =
169        client.query_one("SELECT to_regclass('datumesh_schema_migrations') IS NOT NULL", &[]).db()?.try_get(0).db()?;
170    if tracked {
171        for r in client.query("SELECT version, name, checksum FROM datumesh_schema_migrations", &[]).db()? {
172            let (v, n, c): (i32, String, String) = (r.try_get(0).db()?, r.try_get(1).db()?, r.try_get(2).db()?);
173            lines.push(format!("migration {v} {n} {c}"));
174        }
175    }
176    lines.sort();
177    Ok(lines)
178}
179
180/// Compares the live database's schema with what the migrations produce: applies every
181/// migration to `scratch` (a store on an empty database the caller creates and drops),
182/// then diffs the two catalogs. Returns the differences, `- <line>` for what the live
183/// database lacks and `+ <line>` for what it has in addition; empty when they agree.
184pub fn check(live: &PgStore, scratch: &PgStore) -> Result<Vec<String>> {
185    let mut sc = scratch.conn()?;
186    let objects: i64 = sc
187        .client
188        .query_one(
189            "SELECT count(*) FROM pg_class WHERE relnamespace = (SELECT oid FROM pg_namespace WHERE nspname = current_schema())",
190            &[],
191        ).db()?
192        .try_get(0).db()?;
193    if objects != 0 {
194        return Err(Error::Migration("the scratch database is not empty".into()));
195    }
196    apply_client(&mut sc.client)?;
197    let want = catalog_client(&mut sc.client)?;
198    drop(sc);
199    let have = catalog(live)?;
200    let mut diff: Vec<String> = want.iter().filter(|l| !have.contains(l)).map(|l| format!("- {l}")).collect();
201    diff.extend(have.iter().filter(|l| !want.contains(l)).map(|l| format!("+ {l}")));
202    Ok(diff)
203}
204
205/// SQLite migrations (since 0.4.0): the same contract as the PostgreSQL ones, through a
206/// [`SqliteStore`](crate::SqliteStore). `migrations/sqlite/*.sql`, embedded, applied in
207/// order with the same tracking table; [`check`] compares the live file's schema with
208/// what the migrations produce in a scratch database in memory.
209#[cfg(feature = "sqlite")]
210pub mod sqlite {
211    use rusqlite::Connection;
212
213    use super::Migration;
214    use crate::error::{Db, Error, Result};
215    use crate::SqliteStore;
216
217    /// Every SQLite migration, in order. One so far: the schema of the PostgreSQL
218    /// migrations 1-4 at once (there is no earlier SQLite schema to migrate from).
219    pub const MIGRATIONS: &[Migration] = &[Migration {
220        version: 1,
221        name: "0001_initial.sql",
222        sql: include_str!("../migrations/sqlite/0001_initial.sql"),
223    }];
224
225    const TRACKING: &str = "CREATE TABLE IF NOT EXISTS datumesh_schema_migrations (
226    version INTEGER PRIMARY KEY,
227    name TEXT NOT NULL,
228    checksum TEXT NOT NULL,
229    applied_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
230) STRICT";
231
232    /// Applies the migrations not yet applied and returns their versions (empty when the
233    /// schema is current). All of it is one write transaction (`BEGIN IMMEDIATE`; SQLite's
234    /// schema changes are transactional), so concurrent callers, in this process or
235    /// others, are serialised by the database's write lock, and a failure leaves nothing
236    /// half applied. A migration whose recorded checksum differs from its SQL is an error.
237    pub fn apply(store: &SqliteStore) -> Result<Vec<u32>> {
238        let c = store.raw()?;
239        let applied = apply_conn(&c)?;
240        for v in &applied {
241            tracing::info!(target: "datumesh_store", version = v, "migration applied");
242        }
243        Ok(applied)
244    }
245
246    fn apply_conn(c: &Connection) -> Result<Vec<u32>> {
247        crate::sqlite::write(c, |c| {
248            c.execute_batch(TRACKING).db()?;
249            let mut done: Vec<(i64, String)> = Vec::new();
250            {
251                let mut st = c.prepare("SELECT version, checksum FROM datumesh_schema_migrations").db()?;
252                let rows = st.query_map([], |r| Ok((r.get(0)?, r.get(1)?))).db()?;
253                for r in rows {
254                    done.push(r.db()?);
255                }
256            }
257            let mut applied = Vec::new();
258            for m in MIGRATIONS {
259                match done.iter().find(|(v, _)| *v == i64::from(m.version)) {
260                    Some((_, c)) if *c == m.checksum() => continue,
261                    Some(_) => {
262                        return Err(Error::Migration(format!(
263                            "{} was applied with another checksum: the file changed after it was applied",
264                            m.name
265                        )))
266                    }
267                    None => {}
268                }
269                c.execute_batch(m.sql).db()?;
270                c.execute(
271                    "INSERT INTO datumesh_schema_migrations (version, name, checksum) VALUES (?1, ?2, ?3)",
272                    rusqlite::params![m.version, m.name, m.checksum()],
273                )
274                .db()?;
275                applied.push(m.version);
276            }
277            if let Some((v, _)) = done.iter().find(|(v, _)| !MIGRATIONS.iter().any(|m| i64::from(m.version) == *v)) {
278                return Err(Error::Migration(format!(
279                    "the database has migration {v}, which this version does not know"
280                )));
281            }
282            Ok(applied)
283        })
284    }
285
286    /// The schema as sorted lines: each table, index and trigger with its SQL (whitespace
287    /// collapsed), the indexes SQLite makes for keys, and the recorded migrations.
288    pub fn catalog(store: &SqliteStore) -> Result<Vec<String>> {
289        catalog_conn(&*store.raw()?)
290    }
291
292    fn catalog_conn(c: &Connection) -> Result<Vec<String>> {
293        let mut lines = Vec::new();
294        {
295            let mut st = c
296                .prepare(
297                    "SELECT type, name, tbl_name, sql FROM sqlite_schema
298                      WHERE name NOT LIKE 'sqlite_stat%' ORDER BY type, name",
299                )
300                .db()?;
301            let rows = st
302                .query_map([], |r| {
303                    Ok((
304                        r.get::<_, String>(0)?,
305                        r.get::<_, String>(1)?,
306                        r.get::<_, String>(2)?,
307                        r.get::<_, Option<String>>(3)?,
308                    ))
309                })
310                .db()?;
311            for r in rows {
312                let (ty, name, table, sql) = r.db()?;
313                let sql = sql.map(|s| s.split_whitespace().collect::<Vec<_>>().join(" ")).unwrap_or_default();
314                lines.push(format!("{ty} {name} on {table}: {sql}"));
315            }
316        }
317        let tracked: bool = c
318            .query_row(
319                "SELECT EXISTS (SELECT 1 FROM sqlite_schema WHERE type = 'table' AND name = 'datumesh_schema_migrations')",
320                [],
321                |r| r.get(0),
322            )
323            .db()?;
324        if tracked {
325            let mut st = c.prepare("SELECT version, name, checksum FROM datumesh_schema_migrations").db()?;
326            let rows =
327                st.query_map([], |r| Ok((r.get::<_, i64>(0)?, r.get::<_, String>(1)?, r.get::<_, String>(2)?))).db()?;
328            for r in rows {
329                let (v, n, c) = r.db()?;
330                lines.push(format!("migration {v} {n} {c}"));
331            }
332        }
333        lines.sort();
334        Ok(lines)
335    }
336
337    /// [`check`] of the file at `path`, opened with
338    /// [`SqliteStore::open_existing`]: a path with no file is an error (`no such database
339    /// file`) and nothing is created. Since 0.4.0.
340    pub fn check_file(path: impl AsRef<std::path::Path>) -> Result<Vec<String>> {
341        check(&SqliteStore::open_existing(path)?)
342    }
343
344    /// Compares the live file's schema with what the migrations produce (a store from
345    /// [`SqliteStore::open_existing`] checks without creating or changing anything;
346    /// [`SqliteStore::open`] creates a missing file): applies every
347    /// migration to a scratch database in memory, then diffs the two catalogs. Returns the
348    /// differences, `- <line>` for what the live database lacks and `+ <line>` for what it
349    /// has in addition; empty when they agree.
350    pub fn check(live: &SqliteStore) -> Result<Vec<String>> {
351        let scratch = Connection::open_in_memory().db()?;
352        apply_conn(&scratch)?;
353        let mut want = catalog_conn(&scratch)?;
354        let mut have = catalog(live)?;
355        want.sort();
356        have.sort();
357        let mut diff: Vec<String> = want.iter().filter(|l| !have.contains(l)).map(|l| format!("- {l}")).collect();
358        diff.extend(have.iter().filter(|l| !want.contains(l)).map(|l| format!("+ {l}")));
359        Ok(diff)
360    }
361}
datumesh_store/migrate.rs

datumesh_store/
migrate.rs
