surreal_sync_runtime/checkpoint_fs/
reader.rs

1//! Blocking filesystem checkpoint file reader.
2
3use std::marker::PhantomData;
4use surreal_sync_core::{Checkpoint, CheckpointFile, StoredCheckpoint, SyncPhase};
5
6/// Manager for reading checkpoint files directly from filesystem.
7pub struct CheckpointFileReader {
8    dir: std::path::PathBuf,
9    _marker: PhantomData<()>,
10}
11
12impl CheckpointFileReader {
13    /// Create a new checkpoint file reader for the given directory.
14    pub fn new(dir: impl Into<std::path::PathBuf>) -> Self {
15        Self {
16            dir: dir.into(),
17            _marker: PhantomData,
18        }
19    }
20
21    /// Read the latest checkpoint file for a given phase.
22    pub fn read_latest_checkpoint(&self, phase: SyncPhase) -> anyhow::Result<CheckpointFile> {
23        let mut latest_file = None;
24        let mut latest_time = None;
25
26        for entry in std::fs::read_dir(&self.dir)? {
27            let entry = entry?;
28            let path = entry.path();
29            if let Some(filename) = path.file_name() {
30                let filename_str = filename.to_string_lossy();
31                if filename_str.starts_with(&format!("checkpoint_{}_", phase.as_str())) {
32                    let metadata = entry.metadata()?;
33                    let modified = metadata.modified()?;
34                    if latest_time.is_none() || modified > latest_time.unwrap() {
35                        latest_time = Some(modified);
36                        latest_file = Some(path);
37                    }
38                }
39            }
40        }
41
42        let file_path =
43            latest_file.ok_or_else(|| anyhow::anyhow!("No checkpoint found for phase: {phase}"))?;
44
45        let content = std::fs::read_to_string(file_path)?;
46
47        let stored: StoredCheckpoint = serde_json::from_str(&content)?;
48        let checkpoint: serde_json::Value = serde_json::from_str(&stored.checkpoint_data)?;
49        let file = CheckpointFile {
50            database_type: stored.database_type,
51            checkpoint,
52            phase,
53            created_at: stored.created_at,
54        };
55
56        Ok(file)
57    }
58
59    /// Read and parse checkpoint into database-specific type.
60    pub fn read_checkpoint<C: Checkpoint>(&self, phase: SyncPhase) -> anyhow::Result<C> {
61        let file = self.read_latest_checkpoint(phase)?;
62        file.parse::<C>()
63    }
64}
surreal_sync_runtime/checkpoint_fs/reader.rs

surreal_sync_runtime/checkpoint_fs/
reader.rs
