persistant_migrate/
lib.rs

1// SPDX-License-Identifier: MIT OR Apache-2.0
2//! The logic behind the `persistant-migrate` command: reading a directory of migrations and comparing
3//! it with a database ledger. The commands themselves are in `main.rs`.
4//!
5//! A directory holds `NNN_name.up.sql` files (with `NNN_name.down.sql` partners) or plain
6//! `NNN_name.sql`. The ledger is sqlx's own `_sqlx_migrations`, so a database can be handled by this
7//! tool or by `sqlx migrate run` interchangeably. **Down files are read and checked for pairing, never
8//! run**: [`Migrates`](persistant_relational::Migrates) has no undo.
9
10use persistant_relational::Migration;
11use persistant_relational::catalog::{self, Snapshot};
12use sha2::{Digest, Sha384};
13use std::path::Path;
14
15/// Why a directory could not be turned into migrations.
16#[derive(Debug)]
17pub enum LoadError {
18    /// The directory or a file in it could not be read or parsed.
19    Read(String),
20    /// A `.down.sql` file has no `.up.sql` partner of the same version.
21    UnpairedDown {
22        /// The version.
23        version: i64,
24    },
25    /// The migrations cannot be run as they stand (see [`persistant_sqlx::FromSqlxError`]).
26    Unsupported(persistant_sqlx::FromSqlxError),
27}
28
29impl std::fmt::Display for LoadError {
30    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
31        match self {
32            LoadError::Read(m) => write!(f, "reading the migrations: {m}"),
33            LoadError::UnpairedDown { version } => {
34                write!(f, "migration {version} has a .down.sql but no .up.sql")
35            }
36            LoadError::Unsupported(e) => write!(f, "{e}"),
37        }
38    }
39}
40
41impl std::error::Error for LoadError {}
42
43/// The migrations in `dir`, in version order. Each down file must pair with an up file.
44pub async fn load_dir(dir: &Path) -> Result<Vec<Migration>, LoadError> {
45    let entries = std::fs::read_dir(dir).map_err(|e| LoadError::Read(e.to_string()))?;
46    let mut ups = Vec::new();
47    let mut downs = Vec::new();
48    for entry in entries {
49        let name = entry
50            .map_err(|e| LoadError::Read(e.to_string()))?
51            .file_name()
52            .to_string_lossy()
53            .into_owned();
54        let version = name.split('_').next().and_then(|v| v.parse::<i64>().ok());
55        if let Some(v) = version {
56            if name.ends_with(".down.sql") {
57                downs.push(v);
58            } else if name.ends_with(".sql") {
59                ups.push(v);
60            }
61        }
62    }
63    if let Some(&version) = downs.iter().find(|d| !ups.contains(d)) {
64        return Err(LoadError::UnpairedDown { version });
65    }
66    let migrator = sqlx::migrate::Migrator::new(dir)
67        .await
68        .map_err(|e| LoadError::Read(e.to_string()))?;
69    persistant_sqlx::migrations_from_sqlx(&migrator).map_err(LoadError::Unsupported)
70}
71
72/// The checksum the ledger holds for `m`: SHA-384 of its text.
73pub fn checksum(m: &Migration) -> Vec<u8> {
74    Sha384::digest(m.sql.as_bytes()).to_vec()
75}
76
77/// A snapshot read from the migration files alone, with no database: the migrations' text, in version
78/// order, is read as DDL in `dialect` (`postgres`, `mysql`, `mariadb` or `sqlite`) by UniSQL's
79/// `schema_from_ddl`. Returns the snapshot and the statements UniSQL could not read (head and
80/// reason); a skipped statement can only leave a column unregistered, never invent one.
81///
82/// This is **not** the catalog's answer: only `CREATE TABLE`, `ALTER TABLE .. ADD|DROP|ALTER COLUMN`
83/// and `DROP TABLE` are applied, types keep the spelling the migration wrote, and no key or default
84/// the DDL does not state is assumed. Its `meta` says `source = ddl` so [`catalog::differences`]
85/// never compares it with a catalog snapshot as if they were the same kind.
86pub fn snapshot_from_files(
87    migrations: &[Migration],
88    dialect: &str,
89) -> Result<(Snapshot, Vec<String>), String> {
90    let mut ordered: Vec<&Migration> = migrations.iter().collect();
91    ordered.sort_by_key(|m| m.version);
92    let mut ddl = String::new();
93    for m in &ordered {
94        ddl.push_str(&m.sql);
95        if !m.sql.ends_with(char::from(10)) {
96            ddl.push(char::from(10));
97        }
98    }
99    let read = unisql::schema_from_ddl(&ddl, dialect).map_err(|e| e.to_string())?;
100    let ledger: Vec<(i64, Vec<u8>)> = ordered.iter().map(|m| (m.version, checksum(m))).collect();
101    let mut meta = catalog::migration_meta(&ledger);
102    meta.insert("dialect".to_string(), dialect.to_string());
103    meta.insert("source".to_string(), "ddl".to_string());
104    let mut snapshot = Snapshot::new(read.schema);
105    snapshot.meta = meta;
106    Ok((snapshot, read.skipped))
107}
108
109/// Where one migration stands against a database.
110#[derive(Debug, Clone, PartialEq, Eq)]
111pub enum State {
112    /// Applied, and the text still matches.
113    Applied,
114    /// In the directory, not yet applied.
115    Pending,
116    /// Applied, but the text in the directory is not what was applied.
117    Changed,
118    /// Applied in the database, absent from the directory: the database is newer than this program.
119    Unknown,
120}
121
122/// One line of a status report.
123#[derive(Debug, Clone, PartialEq, Eq)]
124pub struct Line {
125    /// The version.
126    pub version: i64,
127    /// The description (empty for an [`State::Unknown`] version).
128    pub description: String,
129    /// Where it stands.
130    pub state: State,
131}
132
133/// Compare `migrations` with the ledger rows `(version, checksum)`.
134pub fn status(migrations: &[Migration], ledger: &[(i64, Vec<u8>)]) -> Vec<Line> {
135    let mut out: Vec<Line> = migrations
136        .iter()
137        .map(|m| {
138            let state = match ledger.iter().find(|(v, _)| *v == m.version) {
139                None => State::Pending,
140                Some((_, c)) if *c == checksum(m) => State::Applied,
141                Some(_) => State::Changed,
142            };
143            Line {
144                version: m.version,
145                description: m.description.clone(),
146                state,
147            }
148        })
149        .collect();
150    out.extend(
151        ledger
152            .iter()
153            .filter(|(v, _)| !migrations.iter().any(|m| m.version == *v))
154            .map(|(v, _)| Line {
155                version: *v,
156                description: String::new(),
157                state: State::Unknown,
158            }),
159    );
160    out.sort_by_key(|l| l.version);
161    out
162}
163
164/// Where a database URL points, as it is safe to print: the password is dropped.
165#[derive(Debug, Clone, PartialEq, Eq)]
166pub struct Target {
167    /// The URL without its password or query string.
168    pub display: String,
169    /// Whether the host is this machine (loopback, a unix socket, or an SQLite file).
170    pub local: bool,
171}
172
173/// Describe `url`. A URL that cannot be read is reported as **not** local, so the guard errs closed.
174pub fn describe_target(url: &str) -> Target {
175    if url.starts_with("sqlite:") {
176        let path = url.trim_start_matches("sqlite:").trim_start_matches("//");
177        return Target {
178            display: format!("sqlite:{path}"),
179            local: true,
180        };
181    }
182    let Some((scheme, rest)) = url.split_once("://") else {
183        return Target {
184            display: "(unreadable URL)".to_string(),
185            local: false,
186        };
187    };
188    let (rest, query) = rest.split_once('?').unwrap_or((rest, ""));
189    let (authority, db) = rest.split_once('/').unwrap_or((rest, ""));
190    let (user, hostport) = match authority.rsplit_once('@') {
191        Some((cred, hp)) => (cred.split(':').next().unwrap_or(""), hp),
192        None => ("", authority),
193    };
194    let host_from_authority = if let Some(v6) = hostport.strip_prefix('[') {
195        v6.split(']').next().unwrap_or("")
196    } else {
197        hostport.split(':').next().unwrap_or("")
198    };
199    // A `host=` query parameter overrides the authority (PostgreSQL socket and multi-host URLs).
200    let host = query
201        .split('&')
202        .find_map(|kv| kv.strip_prefix("host="))
203        .unwrap_or(host_from_authority);
204    let local = host.is_empty()
205        || host.starts_with('/')
206        || host == "localhost"
207        || host == "::1"
208        || host.starts_with("127.");
209    let at = if user.is_empty() {
210        String::new()
211    } else {
212        format!("{user}@")
213    };
214    let shown_host = if host == host_from_authority {
215        hostport.to_string()
216    } else {
217        format!("{hostport} (host={host})")
218    };
219    Target {
220        display: format!("{scheme}://{at}{shown_host}/{db}"),
221        local,
222    }
223}
persistant_migrate/lib.rs

persistant_migrate/
lib.rs
