1use crate::history::{history_version, Applied};
4use cufflink_types::manifest::validate::MAX_MIGRATION_VERSION;
5use cufflink_types::SqlMigration;
6use siphasher::sip::SipHasher13;
7use std::collections::BTreeMap;
8use std::hash::{Hash, Hasher};
9
10pub fn checksum(migration: &SqlMigration) -> u64 {
14 let mut hasher = SipHasher13::new();
15 migration.name.hash(&mut hasher);
16 history_version(migration.version).hash(&mut hasher);
17 migration.sql.as_str().hash(&mut hasher);
18 hasher.finish()
19}
20
21#[derive(Debug, Clone, PartialEq, Eq)]
22pub struct Plan {
23 pub pending: Vec<SqlMigration>,
25 pub verified: usize,
27 pub baselined: Vec<i32>,
30}
31
32#[derive(Debug, Clone, thiserror::Error, PartialEq, Eq)]
33pub enum PlanError {
34 #[error(
35 "migration V{version} diverges from the one applied: applied '{applied_name}' \
36 (checksum {applied_checksum}), local '{local_name}' (checksum {local_checksum})"
37 )]
38 Divergent {
39 version: i32,
40 applied_name: String,
41 local_name: String,
42 applied_checksum: u64,
43 local_checksum: u64,
44 },
45 #[error("applied migration V{version}__{name} has no local file")]
46 Missing { version: i32, name: String },
47 #[error(
48 "migration V{version} is not applied and precedes V{max_applied}, which is; \
49 a new migration takes a version above every applied one"
50 )]
51 OutOfOrder { version: u32, max_applied: i32 },
52 #[error("migration version {version} is declared twice")]
53 Duplicate { version: u32 },
54 #[error("migration version {version} is above {MAX_MIGRATION_VERSION}")]
55 VersionOutOfRange { version: u32 },
56 #[error("history row of version {version} is unreadable: {reason}")]
57 BadHistoryRow { version: i32, reason: String },
58}
59
60fn by_version(local: &[SqlMigration]) -> Result<BTreeMap<i64, &SqlMigration>, PlanError> {
61 local.iter().try_fold(BTreeMap::new(), |mut found, m| {
62 if m.version > MAX_MIGRATION_VERSION {
63 return Err(PlanError::VersionOutOfRange { version: m.version });
64 }
65 match found.insert(i64::from(m.version), m) {
66 Some(_) => Err(PlanError::Duplicate { version: m.version }),
67 None => Ok(found),
68 }
69 })
70}
71
72enum Standing {
73 Verified,
74 Baselined,
75}
76
77fn standing(
78 applied: &Applied,
79 local: Option<&SqlMigration>,
80 baseline: Option<u32>,
81) -> Result<Standing, PlanError> {
82 let under_baseline =
83 baseline.is_some_and(|baseline| i64::from(applied.version) <= i64::from(baseline));
84 match local.map(|local| (local, checksum(local))) {
85 Some((local, sum)) if local.name == applied.name && sum == applied.checksum => {
86 Ok(Standing::Verified)
87 }
88 _ if under_baseline => Ok(Standing::Baselined),
89 Some((local, sum)) => Err(PlanError::Divergent {
90 version: applied.version,
91 applied_name: applied.name.clone(),
92 local_name: local.name.clone(),
93 applied_checksum: applied.checksum,
94 local_checksum: sum,
95 }),
96 None => Err(PlanError::Missing {
97 version: applied.version,
98 name: applied.name.clone(),
99 }),
100 }
101}
102
103pub fn plan(
107 applied: &[Applied],
108 local: &[SqlMigration],
109 baseline: Option<u32>,
110) -> Result<Plan, PlanError> {
111 let files = by_version(local)?;
112 let (verified, baselined) =
113 applied
114 .iter()
115 .try_fold((0usize, Vec::new()), |(verified, mut baselined), row| {
116 let file = files.get(&i64::from(row.version)).copied();
117 standing(row, file, baseline).map(|standing| match standing {
118 Standing::Verified => (verified + 1, baselined),
119 Standing::Baselined => {
120 baselined.push(row.version);
121 (verified, baselined)
122 }
123 })
124 })?;
125 let max_applied = applied.iter().map(|row| row.version).max();
126 let is_applied = |version: i64| applied.iter().any(|row| i64::from(row.version) == version);
127 let pending: Vec<SqlMigration> = files
128 .iter()
129 .filter(|(version, _)| !is_applied(**version))
130 .map(|(version, file)| match max_applied {
131 Some(max_applied) if *version < i64::from(max_applied) => Err(PlanError::OutOfOrder {
132 version: file.version,
133 max_applied,
134 }),
135 _ => Ok((*file).clone()),
136 })
137 .collect::<Result<_, _>>()?;
138 Ok(Plan {
139 pending,
140 verified,
141 baselined,
142 })
143}
144
145#[cfg(test)]
146mod tests {
147 use super::*;
148
149 fn file(version: u32, name: &str, sql: &str) -> SqlMigration {
150 SqlMigration {
151 version,
152 name: name.into(),
153 sql: sql.into(),
154 no_transaction: false,
155 }
156 }
157
158 fn row(file: &SqlMigration) -> Applied {
159 Applied {
160 version: file.version as i32,
161 name: file.name.clone(),
162 checksum: checksum(file),
163 applied_on: "2024-05-01T09:30:00Z".into(),
164 }
165 }
166
167 fn foreign(version: i32) -> Applied {
168 Applied {
169 version,
170 name: format!("deferred_{version}"),
171 checksum: 7,
172 applied_on: "2024-05-01T09:30:00Z".into(),
173 }
174 }
175
176 fn loan_service() -> Vec<SqlMigration> {
177 vec![
178 file(
179 1,
180 "initial_tables",
181 "CREATE TABLE loans (id UUID PRIMARY KEY);\n",
182 ),
183 file(
184 6,
185 "add_loan_kind",
186 "ALTER TABLE loans ADD COLUMN kind TEXT;\r\n",
187 ),
188 file(
189 7,
190 "renewal_index",
191 "CREATE INDEX loans_renewal ON loans (id);",
192 ),
193 ]
194 }
195
196 fn versions(plan: &Plan) -> Vec<u32> {
197 plan.pending.iter().map(|m| m.version).collect()
198 }
199
200 #[test]
201 fn the_checksum_is_refinerys() {
202 for migration in loan_service().into_iter().chain([
203 file(2_147_483_647, "last", "SELECT 1;"),
204 file(0, "zero", "\u{feff}-- cufflink:no-transaction\nSELECT 'é';"),
205 file(
206 8,
207 "windows_line_endings",
208 "CREATE TABLE loans (\r\n id uuid PRIMARY KEY\r\n);\r\n",
209 ),
210 ]) {
211 let refinery =
212 refinery_core::Migration::unapplied(&migration.file_stem(), &migration.sql)
213 .expect("a refinery migration");
214 assert_eq!(checksum(&migration), refinery.checksum(), "{migration:?}");
215 }
216 }
217
218 #[test]
219 fn a_fresh_history_applies_everything_in_order() {
220 let mut local = loan_service();
221 local.reverse();
222 let plan = plan(&[], &local, None).expect("plan");
223 assert_eq!(versions(&plan), [1, 6, 7]);
224 assert_eq!((plan.verified, plan.baselined), (0, vec![]));
225 }
226
227 #[test]
228 fn a_history_is_continued_after_a_gap() {
229 let local = loan_service();
230 let applied = [row(&local[0]), row(&local[1])];
231 let plan = plan(&applied, &local, None).expect("plan");
232 assert_eq!(versions(&plan), [7]);
233 assert_eq!((plan.verified, plan.baselined), (2, vec![]));
234 let done = [row(&local[0]), row(&local[1]), row(&local[2])];
235 assert_eq!(
236 super::plan(&done, &local, None),
237 Ok(Plan {
238 pending: vec![],
239 verified: 3,
240 baselined: vec![]
241 })
242 );
243 }
244
245 #[test]
246 fn a_changed_name_or_checksum_diverges() {
247 let local = loan_service();
248 let renamed = Applied {
249 name: "add_kind".into(),
250 ..row(&local[1])
251 };
252 assert_eq!(
253 plan(&[row(&local[0]), renamed], &local, None),
254 Err(PlanError::Divergent {
255 version: 6,
256 applied_name: "add_kind".into(),
257 local_name: "add_loan_kind".into(),
258 applied_checksum: checksum(&local[1]),
259 local_checksum: checksum(&local[1]),
260 })
261 );
262 let edited = Applied {
263 checksum: 42,
264 ..row(&local[1])
265 };
266 let error = plan(&[row(&local[0]), edited], &local, None).expect_err("diverges");
267 assert_eq!(
268 error.to_string(),
269 format!(
270 "migration V6 diverges from the one applied: applied 'add_loan_kind' \
271 (checksum 42), local 'add_loan_kind' (checksum {})",
272 checksum(&local[1])
273 )
274 );
275 }
276
277 #[test]
278 fn an_applied_migration_needs_its_file_above_the_baseline() {
279 let local = loan_service();
280 let applied = [row(&local[0]), foreign(3), row(&local[1])];
281 assert_eq!(
282 plan(&applied, &local, None),
283 Err(PlanError::Missing {
284 version: 3,
285 name: "deferred_3".into()
286 })
287 );
288 assert_eq!(
289 plan(&applied, &local, Some(2)),
290 Err(PlanError::Missing {
291 version: 3,
292 name: "deferred_3".into()
293 })
294 );
295 let plan = plan(&applied, &local, Some(3)).expect("plan");
296 assert_eq!(versions(&plan), [7]);
297 assert_eq!((plan.verified, plan.baselined), (2, vec![3]));
298 }
299
300 #[test]
301 fn a_history_under_the_baseline_is_accepted_unverified() {
302 let applied: Vec<Applied> = (1..=15).map(foreign).collect();
303 let local: Vec<SqlMigration> = (14..=16)
304 .map(|v| file(v, &format!("report_{v}"), "SELECT 1;"))
305 .collect();
306 let plan = plan(&applied, &local, Some(15)).expect("plan");
307 assert_eq!(versions(&plan), [16]);
308 assert_eq!(plan.verified, 0);
309 assert_eq!(plan.baselined, (1..=15).collect::<Vec<i32>>());
310 }
311
312 #[test]
313 fn a_new_file_under_an_applied_version_is_out_of_order() {
314 let mut local = loan_service();
315 let applied = [row(&local[0]), row(&local[1])];
316 local.push(file(4, "late", "SELECT 1;"));
317 assert_eq!(
318 plan(&applied, &local, None),
319 Err(PlanError::OutOfOrder {
320 version: 4,
321 max_applied: 6
322 })
323 );
324 assert_eq!(
325 plan(&applied, &local, Some(6)),
326 Err(PlanError::OutOfOrder {
327 version: 4,
328 max_applied: 6
329 })
330 );
331 }
332
333 #[test]
334 fn line_endings_are_part_of_what_is_checksummed() {
335 assert_ne!(
336 checksum(&file(8, "loans", "SELECT 1;\r\nSELECT 2;\r\n")),
337 checksum(&file(8, "loans", "SELECT 1;\nSELECT 2;\n")),
338 "a checkout that rewrites line endings diverges, as it does under refinery"
339 );
340 }
341
342 #[test]
343 fn a_version_the_history_cannot_hold_is_refused() {
344 let mut local = loan_service();
345 local.push(file(2_147_483_648, "too_far", "SELECT 1;"));
346 assert_eq!(
347 plan(&[], &local, None),
348 Err(PlanError::VersionOutOfRange {
349 version: 2_147_483_648
350 })
351 );
352 }
353
354 #[test]
355 fn a_version_declared_twice_is_refused() {
356 let mut local = loan_service();
357 local.push(file(6, "again", "SELECT 1;"));
358 assert_eq!(
359 plan(&[], &local, None),
360 Err(PlanError::Duplicate { version: 6 })
361 );
362 }
363}