surreal_sync_runtime/checkpoint_fs/
filesystem.rs

1//! Filesystem-based checkpoint storage implementation.
2
3use anyhow::Result;
4use async_trait::async_trait;
5use chrono::Utc;
6use std::path::PathBuf;
7use surreal_sync_core::{CheckpointID, CheckpointStore, StoredCheckpoint};
8
9/// Filesystem implementation of CheckpointStore trait.
10///
11/// Stores checkpoints as JSON files in a directory.
12pub struct FilesystemStore {
13    dir: PathBuf,
14}
15
16impl FilesystemStore {
17    /// Create a new FilesystemStore with the given directory.
18    pub fn new(dir: impl Into<PathBuf>) -> Self {
19        Self { dir: dir.into() }
20    }
21
22    /// Get the directory path.
23    pub fn dir(&self) -> &PathBuf {
24        &self.dir
25    }
26}
27
28#[async_trait]
29impl CheckpointStore for FilesystemStore {
30    async fn store_checkpoint(&self, id: &CheckpointID, checkpoint_data: String) -> Result<()> {
31        tokio::fs::create_dir_all(&self.dir).await?;
32
33        let stored = StoredCheckpoint {
34            checkpoint_data,
35            database_type: id.database_type.clone(),
36            phase: id.phase.clone(),
37            created_at: Utc::now(),
38        };
39
40        let timestamp = Utc::now().to_rfc3339();
41        let filename = self
42            .dir
43            .join(format!("checkpoint_{}_{}.json", id.phase, timestamp));
44
45        tokio::fs::write(&filename, serde_json::to_string_pretty(&stored)?).await?;
46        tracing::info!("Stored checkpoint to {}", filename.display());
47        Ok(())
48    }
49
50    async fn read_checkpoint(&self, id: &CheckpointID) -> Result<Option<StoredCheckpoint>> {
51        if !tokio::fs::try_exists(&self.dir).await.unwrap_or(false) {
52            return Ok(None);
53        }
54
55        // Find latest checkpoint file for this phase
56        let prefix = format!("checkpoint_{}_", id.phase);
57        let mut latest_file = None;
58        let mut latest_time = None;
59
60        let mut entries = tokio::fs::read_dir(&self.dir).await?;
61        while let Some(entry) = entries.next_entry().await? {
62            let filename = entry.file_name().to_string_lossy().to_string();
63            if filename.starts_with(&prefix) && filename.ends_with(".json") {
64                let modified = entry.metadata().await?.modified()?;
65                if latest_time.is_none() || modified > latest_time.unwrap() {
66                    latest_time = Some(modified);
67                    latest_file = Some(entry.path());
68                }
69            }
70        }
71
72        match latest_file {
73            Some(path) => {
74                let content = tokio::fs::read_to_string(path).await?;
75                Ok(Some(serde_json::from_str(&content)?))
76            }
77            None => Ok(None),
78        }
79    }
80}
surreal_sync_runtime/checkpoint_fs/filesystem.rs

surreal_sync_runtime/checkpoint_fs/
filesystem.rs
