surreal_sync_core/checkpoint/
manager.rs

1//! Generic sync manager for checkpoint operations.
2
3use crate::checkpoint::{
4    store::CheckpointStore, Checkpoint, CheckpointID, StoredCheckpoint, SyncPhase,
5};
6
7/// Manager for handling sync operations with checkpoint tracking.
8///
9/// The `SyncManager` is generic over the checkpoint store type `S`,
10/// providing storage-agnostic checkpoint management.
11pub struct SyncManager<S: CheckpointStore> {
12    store: S,
13    emit_checkpoints: bool,
14}
15
16impl<S: CheckpointStore> SyncManager<S> {
17    /// Create a new sync manager with the given store.
18    ///
19    /// By default, checkpoint emission is enabled.
20    pub fn new(store: S) -> Self {
21        Self {
22            store,
23            emit_checkpoints: true,
24        }
25    }
26
27    /// Create a new sync manager with checkpoint emission disabled.
28    pub fn new_without_emit(store: S) -> Self {
29        Self {
30            store,
31            emit_checkpoints: false,
32        }
33    }
34
35    /// Set whether to emit checkpoints.
36    pub fn with_emit_checkpoints(mut self, emit: bool) -> Self {
37        self.emit_checkpoints = emit;
38        self
39    }
40
41    /// Check if checkpoint emission is enabled.
42    pub fn emit_checkpoints(&self) -> bool {
43        self.emit_checkpoints
44    }
45
46    /// Get a reference to the underlying store.
47    pub fn store(&self) -> &S {
48        &self.store
49    }
50
51    /// Emit checkpoint for any database-specific checkpoint type.
52    pub async fn emit_checkpoint<C: Checkpoint>(
53        &self,
54        checkpoint: &C,
55        phase: SyncPhase,
56    ) -> anyhow::Result<()> {
57        if !self.emit_checkpoints {
58            return Ok(());
59        }
60
61        let id = CheckpointID {
62            database_type: C::DATABASE_TYPE.to_string(),
63            phase: phase.as_str().to_string(),
64        };
65
66        let checkpoint_data = serde_json::to_string(checkpoint)?;
67        self.store.store_checkpoint(&id, checkpoint_data).await?;
68
69        tracing::info!(
70            "Emitted {} checkpoint: {}",
71            phase,
72            checkpoint.to_cli_string()
73        );
74
75        Ok(())
76    }
77
78    /// Read and parse checkpoint into database-specific type.
79    pub async fn read_checkpoint<C: Checkpoint>(&self, phase: SyncPhase) -> anyhow::Result<C> {
80        let id = CheckpointID {
81            database_type: C::DATABASE_TYPE.to_string(),
82            phase: phase.as_str().to_string(),
83        };
84
85        let stored = self
86            .store
87            .read_checkpoint(&id)
88            .await?
89            .ok_or_else(|| anyhow::anyhow!("No checkpoint found for phase: {phase}"))?;
90
91        if stored.database_type != C::DATABASE_TYPE {
92            return Err(anyhow::anyhow!(
93                "Checkpoint database type mismatch: expected '{}', found '{}'",
94                C::DATABASE_TYPE,
95                stored.database_type
96            ));
97        }
98
99        Ok(serde_json::from_str(&stored.checkpoint_data)?)
100    }
101}
102
103/// A no-op checkpoint store for when checkpoint storage is disabled.
104pub struct NullStore;
105
106#[async_trait::async_trait]
107impl CheckpointStore for NullStore {
108    async fn store_checkpoint(
109        &self,
110        _id: &CheckpointID,
111        _checkpoint_data: String,
112    ) -> anyhow::Result<()> {
113        Ok(())
114    }
115
116    async fn read_checkpoint(
117        &self,
118        _id: &CheckpointID,
119    ) -> anyhow::Result<Option<StoredCheckpoint>> {
120        Ok(None)
121    }
122}
123
124/// Type alias for SyncManager with disabled checkpointing.
125pub type NullSyncManager = SyncManager<NullStore>;
126
127impl NullSyncManager {
128    /// Create a sync manager that does nothing (for disabled checkpoint storage).
129    pub fn disabled() -> Self {
130        SyncManager::new_without_emit(NullStore)
131    }
132}
surreal_sync_core/checkpoint/manager.rs

surreal_sync_core/checkpoint/
manager.rs
