cufflink_migrate/
scan.rs

1//! The migration files of a directory.
2
3use cufflink_types::manifest::validate::{is_migration_name, MAX_MIGRATION_VERSION};
4use cufflink_types::SqlMigration;
5use std::path::{Path, PathBuf};
6
7#[derive(Debug, Clone, thiserror::Error, PartialEq, Eq)]
8pub enum ScanError {
9    #[error("cannot read {}: {reason}", path.display())]
10    Unreadable { path: PathBuf, reason: String },
11    #[error("{file}: is not named V{{version}}__{{name}}.sql")]
12    BadName { file: String },
13    #[error(
14        "{file}: unversioned (U) migrations are not supported; name it V{{version}}__{{name}}.sql"
15    )]
16    Unversioned { file: String },
17    #[error("{file}: version {version} is above {MAX_MIGRATION_VERSION}")]
18    VersionTooLarge { file: String, version: u64 },
19    #[error("{file}: is not UTF-8")]
20    NotUtf8 { file: String },
21    #[error("{second}: version {version} is already taken by {first}")]
22    Duplicate {
23        version: u32,
24        first: String,
25        second: String,
26    },
27}
28
29/// The version and name of `V{version}__{name}.sql`.
30fn parse_file_name(file: &str) -> Result<(u32, String), ScanError> {
31    let bad = || ScanError::BadName {
32        file: file.to_string(),
33    };
34    let stem = file.strip_suffix(".sql").ok_or_else(bad)?;
35    let (prefix, name) = stem.split_once("__").ok_or_else(bad)?;
36    let digits = match prefix.split_at_checked(1) {
37        Some(("V", digits)) => Ok(digits),
38        Some(("U", digits)) if is_version(digits) => Err(ScanError::Unversioned {
39            file: file.to_string(),
40        }),
41        _ => Err(bad()),
42    }?;
43    if !is_version(digits) || !is_migration_name(name) {
44        return Err(bad());
45    }
46    let version = digits.parse::<u64>().unwrap_or(u64::MAX);
47    u32::try_from(version)
48        .ok()
49        .filter(|version| *version <= MAX_MIGRATION_VERSION)
50        .map(|version| (version, name.to_string()))
51        .ok_or_else(|| ScanError::VersionTooLarge {
52            file: file.to_string(),
53            version,
54        })
55}
56
57fn is_version(digits: &str) -> bool {
58    !digits.is_empty() && digits.bytes().all(|b| b.is_ascii_digit())
59}
60
61fn unreadable(path: &Path) -> impl Fn(std::io::Error) -> ScanError + '_ {
62    move |e| ScanError::Unreadable {
63        path: path.to_path_buf(),
64        reason: e.to_string(),
65    }
66}
67
68/// A migration and the file it was read from.
69#[derive(Debug, Clone, PartialEq, Eq)]
70pub struct MigrationFile {
71    pub path: PathBuf,
72    pub migration: SqlMigration,
73}
74
75fn read(path: &Path, file: &str) -> Result<SqlMigration, ScanError> {
76    let (version, name) = parse_file_name(file)?;
77    let sql = String::from_utf8(std::fs::read(path).map_err(unreadable(path))?).map_err(|_| {
78        ScanError::NotUtf8 {
79            file: file.to_string(),
80        }
81    })?;
82    Ok(SqlMigration {
83        version,
84        name,
85        no_transaction: SqlMigration::declares_no_transaction(&sql),
86        sql,
87    })
88}
89
90/// The `.sql` files directly in `dir`, by name.
91fn sql_files(dir: &Path) -> Result<Vec<(PathBuf, String)>, ScanError> {
92    let mut files: Vec<(PathBuf, String)> = std::fs::read_dir(dir)
93        .map_err(unreadable(dir))?
94        .map(|entry| entry.map(|entry| entry.path()).map_err(unreadable(dir)))
95        .collect::<Result<Vec<_>, _>>()?
96        .into_iter()
97        .filter(|path| path.is_file())
98        .filter_map(|path| {
99            let file = path.file_name()?.to_string_lossy().into_owned();
100            file.ends_with(".sql").then_some((path, file))
101        })
102        .collect();
103    files.sort();
104    Ok(files)
105}
106
107fn file_name(path: &Path) -> String {
108    path.file_name()
109        .map(|name| name.to_string_lossy().into_owned())
110        .unwrap_or_default()
111}
112
113/// The migration files of `dir`, ascending. Every `.sql` file there is a
114/// migration; anything else is left alone.
115pub fn scan_files(dir: &Path) -> Result<Vec<MigrationFile>, ScanError> {
116    let mut found: Vec<MigrationFile> = sql_files(dir)?
117        .into_iter()
118        .map(|(path, file)| read(&path, &file).map(|migration| MigrationFile { path, migration }))
119        .collect::<Result<_, _>>()?;
120    found.sort_by_key(|file| file.migration.version);
121    match found
122        .windows(2)
123        .find(|pair| pair[0].migration.version == pair[1].migration.version)
124    {
125        Some(pair) => Err(ScanError::Duplicate {
126            version: pair[0].migration.version,
127            first: file_name(&pair[0].path),
128            second: file_name(&pair[1].path),
129        }),
130        None => Ok(found),
131    }
132}
133
134/// The migrations of `dir`, ascending, each with the exact text of its
135/// file.
136pub fn scan_dir(dir: &Path) -> Result<Vec<SqlMigration>, ScanError> {
137    scan_files(dir).map(|files| files.into_iter().map(|file| file.migration).collect())
138}
139
140#[cfg(test)]
141mod tests {
142    use super::*;
143
144    fn dir(files: &[(&str, &[u8])]) -> tempfile::TempDir {
145        let dir = tempfile::tempdir().expect("temp dir");
146        for (name, bytes) in files {
147            std::fs::write(dir.path().join(name), bytes).expect("write");
148        }
149        dir
150    }
151
152    #[test]
153    fn files_are_read_in_version_order_with_their_exact_text() {
154        let crlf = "ALTER TABLE loans ADD COLUMN kind TEXT;\r\n\r\n";
155        let bom =
156            "\u{feff}-- cufflink:no-transaction\nCREATE INDEX CONCURRENTLY i ON loans (id);\n";
157        let dir = dir(&[
158            ("V10__later.sql", b"SELECT 10;"),
159            ("V6__add_loan_kind.sql", crlf.as_bytes()),
160            ("V1__initial_tables.sql", b"CREATE TABLE loans ();\n"),
161            ("V007__renewal_index.sql", bom.as_bytes()),
162            ("README.md", b"not a migration"),
163            ("loans.sql.bak", b"nor this"),
164        ]);
165        std::fs::create_dir(dir.path().join("archive.sql")).expect("subdirectory");
166        let found = scan_dir(dir.path()).expect("scan");
167        assert_eq!(
168            found
169                .iter()
170                .map(|m| (m.version, m.name.as_str(), m.no_transaction))
171                .collect::<Vec<_>>(),
172            [
173                (1, "initial_tables", false),
174                (6, "add_loan_kind", false),
175                (7, "renewal_index", true),
176                (10, "later", false)
177            ]
178        );
179        assert_eq!(found[1].sql, crlf);
180        assert_eq!(found[2].sql, bom);
181        assert_eq!(
182            scan_dir(tempfile::tempdir().expect("dir").path()),
183            Ok(vec![])
184        );
185    }
186
187    #[test]
188    fn a_file_that_is_not_a_migration_is_refused_by_name() {
189        for (file, expected) in [
190            (
191                "U1__repeatable.sql",
192                "U1__repeatable.sql: unversioned (U) migrations are not supported; \
193                 name it V{version}__{name}.sql",
194            ),
195            (
196                "initial.sql",
197                "initial.sql: is not named V{version}__{name}.sql",
198            ),
199            (
200                "V1_initial.sql",
201                "V1_initial.sql: is not named V{version}__{name}.sql",
202            ),
203            (
204                "Vx__initial.sql",
205                "Vx__initial.sql: is not named V{version}__{name}.sql",
206            ),
207            (
208                "V-1__initial.sql",
209                "V-1__initial.sql: is not named V{version}__{name}.sql",
210            ),
211            (
212                "V1__bad-name.sql",
213                "V1__bad-name.sql: is not named V{version}__{name}.sql",
214            ),
215            (
216                "V1__.sql",
217                "V1__.sql: is not named V{version}__{name}.sql",
218            ),
219            (
220                "V2147483648__big.sql",
221                "V2147483648__big.sql: version 2147483648 is above 2147483647",
222            ),
223            (
224                "V99999999999999999999999__huge.sql",
225                "V99999999999999999999999__huge.sql: version 18446744073709551615 is above 2147483647",
226            ),
227        ] {
228            let dir = dir(&[(file, b"SELECT 1;")]);
229            assert_eq!(
230                scan_dir(dir.path()).expect_err(file).to_string(),
231                expected
232            );
233        }
234    }
235
236    #[test]
237    fn a_version_belongs_to_one_file() {
238        let dir = dir(&[
239            ("V1__initial.sql", b"SELECT 1;"),
240            ("V01__again.sql", b"SELECT 1;"),
241        ]);
242        assert_eq!(
243            scan_dir(dir.path()),
244            Err(ScanError::Duplicate {
245                version: 1,
246                first: "V01__again.sql".into(),
247                second: "V1__initial.sql".into()
248            })
249        );
250    }
251
252    #[test]
253    fn unreadable_input_is_an_error() {
254        let dir = dir(&[("V1__latin1.sql", &[0x53, 0x45, 0x4c, 0xe9][..])]);
255        assert_eq!(
256            scan_dir(dir.path()),
257            Err(ScanError::NotUtf8 {
258                file: "V1__latin1.sql".into()
259            })
260        );
261        let missing = dir.path().join("absent");
262        assert!(matches!(
263            scan_dir(&missing),
264            Err(ScanError::Unreadable { path, .. }) if path == missing
265        ));
266    }
267}
cufflink_migrate/scan.rs

cufflink_migrate/
scan.rs
