linkrs_storage/
migration_history.rs

1use linkrs_core::{StorageError, StorageResult};
2use serde::{Deserialize, Serialize};
3use std::collections::{HashMap, HashSet};
4use std::path::Path;
5
6/// Status of a migration execution.
7#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
8#[serde(rename_all = "snake_case")]
9pub enum MigrationStatus {
10    Applied,
11    RolledBack,
12    Failed,
13}
14
15impl std::fmt::Display for MigrationStatus {
16    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
17        match self {
18            MigrationStatus::Applied => write!(f, "applied"),
19            MigrationStatus::RolledBack => write!(f, "rolled_back"),
20            MigrationStatus::Failed => write!(f, "failed"),
21        }
22    }
23}
24
25/// One row of the `migration_history` table.
26#[derive(Debug, Clone, Serialize, Deserialize)]
27pub struct MigrationHistoryRecord {
28    pub id: u64,
29    pub space: String,
30    pub label: String,
31    pub is_edge: bool,
32    pub from_version: u64,
33    pub to_version: u64,
34    pub plan_hash: String,
35    pub safety_level: String,
36    pub steps_count: usize,
37    pub rows_migrated: u64,
38    pub status: MigrationStatus,
39    pub applied_at: u64,
40    pub completed_at: Option<u64>,
41    pub error_message: Option<String>,
42}
43
44impl MigrationHistoryRecord {
45    #[allow(clippy::too_many_arguments)]
46    pub fn new(
47        space: String,
48        label: String,
49        is_edge: bool,
50        from_version: u64,
51        to_version: u64,
52        plan_hash: String,
53        safety_level: String,
54        steps_count: usize,
55        rows_migrated: u64,
56        status: MigrationStatus,
57    ) -> Self {
58        let now = std::time::SystemTime::now()
59            .duration_since(std::time::UNIX_EPOCH)
60            .map(|d| d.as_millis() as u64)
61            .unwrap_or(0);
62        Self {
63            id: 0,
64            space,
65            label,
66            is_edge,
67            from_version,
68            to_version,
69            plan_hash,
70            safety_level,
71            steps_count,
72            rows_migrated,
73            status,
74            applied_at: now,
75            completed_at: Some(now),
76            error_message: None,
77        }
78    }
79}
80
81/// In-memory manager for migration history. Persisted as JSON.
82#[derive(Debug, Clone, Serialize, Deserialize, Default)]
83pub struct MigrationHistoryManager {
84    records: Vec<MigrationHistoryRecord>,
85    next_id: u64,
86    #[serde(skip)]
87    index: HashMap<(String, String, bool, u64), usize>,
88}
89
90impl MigrationHistoryManager {
91    pub fn new() -> Self {
92        Self {
93            records: Vec::new(),
94            next_id: 1,
95            index: HashMap::new(),
96        }
97    }
98
99    fn rebuild_index(&mut self) {
100        self.index.clear();
101        for (idx, r) in self.records.iter().enumerate() {
102            self.index.insert(
103                (r.space.clone(), r.label.clone(), r.is_edge, r.to_version),
104                idx,
105            );
106        }
107        if let Some(max_id) = self.records.iter().map(|r| r.id).max() {
108            self.next_id = max_id + 1;
109        } else {
110            self.next_id = 1;
111        }
112    }
113
114    pub fn record(&mut self, mut rec: MigrationHistoryRecord) -> StorageResult<()> {
115        let key = (
116            rec.space.clone(),
117            rec.label.clone(),
118            rec.is_edge,
119            rec.to_version,
120        );
121        if self.index.contains_key(&key) {
122            return Err(StorageError::already_exists(format!(
123                "migration_history unique violation: space={} label={} is_edge={} to_version={}",
124                rec.space, rec.label, rec.is_edge, rec.to_version
125            )));
126        }
127        rec.id = self.next_id;
128        self.next_id += 1;
129        let idx = self.records.len();
130        self.records.push(rec);
131        self.index.insert(key, idx);
132        Ok(())
133    }
134
135    pub fn list(&self, space: &str, label: &str, is_edge: bool) -> Vec<MigrationHistoryRecord> {
136        self.records
137            .iter()
138            .filter(|r| r.space == space && r.label == label && r.is_edge == is_edge)
139            .cloned()
140            .collect()
141    }
142
143    pub fn list_all(&self) -> Vec<MigrationHistoryRecord> {
144        self.records.clone()
145    }
146
147    pub fn get_applied_versions(&self, space: &str, label: &str, is_edge: bool) -> HashSet<u64> {
148        self.records
149            .iter()
150            .filter(|r| {
151                r.space == space
152                    && r.label == label
153                    && r.is_edge == is_edge
154                    && r.status == MigrationStatus::Applied
155            })
156            .map(|r| r.to_version)
157            .collect()
158    }
159
160    pub fn get_applied_versions_sorted(&self, space: &str, label: &str, is_edge: bool) -> Vec<u64> {
161        let mut v: Vec<u64> = self
162            .get_applied_versions(space, label, is_edge)
163            .into_iter()
164            .collect();
165        v.sort_unstable();
166        v
167    }
168
169    /// Check a plan's version range against already-applied versions.
170    ///
171    /// Pure function over a sorted applied list so both the storage engine
172    /// and the migration executor share one chain rule: forward-only ranges,
173    /// no re-applying an applied target, and no gaps (a non-empty chain must
174    /// continue exactly where it left off).
175    pub fn check_chain(
176        applied_sorted: &[u64],
177        space: &str,
178        label: &str,
179        is_edge: bool,
180        from_version: u64,
181        to_version: u64,
182    ) -> StorageResult<()> {
183        if to_version <= from_version {
184            return Err(StorageError::invalid_input(format!(
185                "non-forward migration range v{from_version} -> v{to_version} for {space}/{label}"
186            )));
187        }
188        if applied_sorted.contains(&to_version) {
189            return Err(StorageError::already_exists(format!(
190                "migration target v{to_version} for {space}/{label} is already applied"
191            )));
192        }
193        let mut ascending = applied_sorted.to_vec();
194        ascending.sort_unstable();
195        if let Some(&latest) = ascending.last() {
196            if from_version != latest {
197                return Err(StorageError::invalid_input(format!(
198                    "migration chain gap for {space}/{label} (is_edge={is_edge}): latest applied is v{latest}, plan starts at v{from_version}"
199                )));
200            }
201        }
202        Ok(())
203    }
204
205    /// Instance form of [`Self::check_chain`] over this manager's records.
206    pub fn validate_version_chain(
207        &self,
208        space: &str,
209        label: &str,
210        is_edge: bool,
211        from_version: u64,
212        to_version: u64,
213    ) -> StorageResult<()> {
214        Self::check_chain(
215            &self.get_applied_versions_sorted(space, label, is_edge),
216            space,
217            label,
218            is_edge,
219            from_version,
220            to_version,
221        )
222    }
223
224    pub fn find(
225        &self,
226        space: &str,
227        label: &str,
228        is_edge: bool,
229        to_version: u64,
230    ) -> Option<MigrationHistoryRecord> {
231        let key = (space.to_string(), label.to_string(), is_edge, to_version);
232        self.index
233            .get(&key)
234            .and_then(|&idx| self.records.get(idx).cloned())
235    }
236
237    pub fn save_to_file(&self, path: &Path) -> StorageResult<()> {
238        if let Some(parent) = path.parent() {
239            std::fs::create_dir_all(parent).map_err(|e| StorageError::io_error(e.to_string()))?;
240        }
241        let json = serde_json::to_string_pretty(&self.records)
242            .map_err(|e| StorageError::serialize_error(e.to_string()))?;
243        std::fs::write(path, json).map_err(|e| StorageError::io_error(e.to_string()))?;
244        Ok(())
245    }
246
247    pub fn load_from_file(path: &Path) -> StorageResult<Self> {
248        if !path.exists() {
249            return Ok(Self::new());
250        }
251        let data =
252            std::fs::read_to_string(path).map_err(|e| StorageError::io_error(e.to_string()))?;
253        if data.trim().is_empty() {
254            return Ok(Self::new());
255        }
256        let records: Vec<MigrationHistoryRecord> = serde_json::from_str(&data)
257            .map_err(|e| StorageError::deserialize_error(e.to_string()))?;
258        let mut mgr = Self {
259            records,
260            next_id: 1,
261            index: HashMap::new(),
262        };
263        mgr.rebuild_index();
264        Ok(mgr)
265    }
266
267    pub fn clear(&mut self) {
268        self.records.clear();
269        self.index.clear();
270        self.next_id = 1;
271    }
272}
273
274#[cfg(test)]
275mod tests {
276    use super::*;
277
278    #[test]
279    fn test_record_and_query() {
280        let mut mgr = MigrationHistoryManager::new();
281        let rec = MigrationHistoryRecord::new(
282            "s".into(),
283            "l".into(),
284            false,
285            1,
286            2,
287            "hash".into(),
288            "SAFE".into(),
289            1,
290            10,
291            MigrationStatus::Applied,
292        );
293        mgr.record(rec).unwrap();
294        assert_eq!(mgr.get_applied_versions("s", "l", false).len(), 1);
295        assert!(mgr.get_applied_versions("s", "l", true).is_empty());
296    }
297
298    #[test]
299    fn test_unique_violation() {
300        let mut mgr = MigrationHistoryManager::new();
301        let rec = MigrationHistoryRecord::new(
302            "s".into(),
303            "l".into(),
304            false,
305            1,
306            2,
307            "hash".into(),
308            "SAFE".into(),
309            1,
310            10,
311            MigrationStatus::Applied,
312        );
313        mgr.record(rec.clone()).unwrap();
314        let err = mgr.record(rec).unwrap_err();
315        assert_eq!(
316            err.kind(),
317            linkrs_core::error::storage::StorageErrorKind::AlreadyExists
318        );
319    }
320
321    #[test]
322    fn test_save_load() {
323        let mut mgr = MigrationHistoryManager::new();
324        mgr.record(MigrationHistoryRecord::new(
325            "s".into(),
326            "l".into(),
327            false,
328            1,
329            2,
330            "h".into(),
331            "SAFE".into(),
332            1,
333            5,
334            MigrationStatus::Applied,
335        ))
336        .unwrap();
337        let tmp = tempfile::NamedTempFile::new().unwrap();
338        mgr.save_to_file(tmp.path()).unwrap();
339        let loaded = MigrationHistoryManager::load_from_file(tmp.path()).unwrap();
340        assert_eq!(loaded.records.len(), 1);
341        assert_eq!(loaded.get_applied_versions("s", "l", false).len(), 1);
342    }
343
344    #[test]
345    fn test_check_chain_fresh_label() {
346        assert!(MigrationHistoryManager::check_chain(&[], "s", "l", false, 1, 2).is_ok());
347    }
348
349    #[test]
350    fn test_check_chain_rejects_non_forward() {
351        assert!(MigrationHistoryManager::check_chain(&[], "s", "l", false, 2, 2).is_err());
352        assert!(MigrationHistoryManager::check_chain(&[], "s", "l", false, 3, 2).is_err());
353    }
354
355    #[test]
356    fn test_check_chain_rejects_reapply() {
357        assert!(MigrationHistoryManager::check_chain(&[2], "s", "l", false, 1, 2).is_err());
358    }
359
360    #[test]
361    fn test_check_chain_rejects_gap() {
362        assert!(MigrationHistoryManager::check_chain(&[2], "s", "l", false, 3, 4).is_err());
363        assert!(MigrationHistoryManager::check_chain(&[1, 2], "s", "l", false, 1, 3).is_err());
364    }
365
366    #[test]
367    fn test_check_chain_accepts_continuation() {
368        assert!(MigrationHistoryManager::check_chain(&[2], "s", "l", false, 2, 3).is_ok());
369        assert!(MigrationHistoryManager::check_chain(&[1, 2], "s", "l", false, 2, 5).is_ok());
370    }
371
372    #[test]
373    fn test_validate_version_chain_instance() {
374        let mut mgr = MigrationHistoryManager::new();
375        mgr.record(MigrationHistoryRecord::new(
376            "s".into(),
377            "l".into(),
378            false,
379            1,
380            2,
381            "h".into(),
382            "SAFE".into(),
383            1,
384            5,
385            MigrationStatus::Applied,
386        ))
387        .unwrap();
388        assert!(mgr.validate_version_chain("s", "l", false, 2, 3).is_ok());
389        assert!(mgr.validate_version_chain("s", "l", false, 1, 3).is_err());
390    }
391}
linkrs_storage/migration_history.rs

linkrs_storage/
migration_history.rs
