appcore_sync_sqlite/
checkpoint.rs

1// =============================================================================
2//        #######
3//     ###       ###     F: checkpoint.rs
4//    ##   ## ##   ##    P: AppCore-Runtime
5//         ## ##
6//                       C: 2026/08/26 00:00:00 by dnettoRaw
7//    ##   ## ##   ##    U: 2026/08/26 00:00:00 by dnettoRaw
8//      ###########      S: 2.0.0
9// =============================================================================
10
11use crate::{SqliteSyncError, SqliteSyncStore};
12use appcore_sync::{SyncCheckpointStore, SyncError, SyncResult};
13use rusqlite::{params, OptionalExtension, TransactionBehavior};
14
15const MAX_PEER_ID_BYTES: usize = 256;
16
17/// SQLite-backed per-peer synchronization checkpoints.
18#[derive(Debug, Clone)]
19pub struct SqliteSyncCheckpointStore {
20    store: SqliteSyncStore,
21}
22
23impl SqliteSyncCheckpointStore {
24    pub(crate) fn new(store: SqliteSyncStore) -> Self {
25        Self { store }
26    }
27}
28
29impl SyncCheckpointStore for SqliteSyncCheckpointStore {
30    fn get_checkpoint(&self, peer_id: &str) -> SyncResult<Option<(u64, String)>> {
31        validate_peer_id(peer_id)?;
32        self.store
33            .with_connection(|connection| {
34                connection
35                    .query_row(
36                        "SELECT sequence, batch_hash FROM appcore_sync_checkpoint
37                         WHERE peer_id = ?1",
38                        [peer_id],
39                        |row| Ok((row.get::<_, i64>(0)?, row.get::<_, String>(1)?)),
40                    )
41                    .optional()
42                    .map_err(SqliteSyncError::database)?
43                    .map(|(sequence, hash)| {
44                        u64::try_from(sequence)
45                            .map(|sequence| (sequence, hash))
46                            .map_err(|_| SqliteSyncError::CorruptRecord("checkpoint"))
47                    })
48                    .transpose()
49            })
50            .map_err(SqliteSyncError::sync)
51    }
52
53    fn set_checkpoint(&self, peer_id: &str, sequence: u64, hash: &str) -> SyncResult<()> {
54        validate_peer_id(peer_id)?;
55        validate_hash(hash)?;
56        let sequence_u64 = sequence;
57        let sequence = i64::try_from(sequence).map_err(|_| {
58            SyncError::ReplicationFailed("checkpoint capacity exceeded".to_string())
59        })?;
60        self.store
61            .with_connection(|connection| {
62                let transaction = connection
63                    .transaction_with_behavior(TransactionBehavior::Immediate)
64                    .map_err(SqliteSyncError::database)?;
65                let existing: Option<(i64, String)> = transaction
66                    .query_row(
67                        "SELECT sequence, batch_hash FROM appcore_sync_checkpoint WHERE peer_id = ?1",
68                        [peer_id],
69                        |row| Ok((row.get(0)?, row.get(1)?)),
70                    )
71                    .optional()
72                    .map_err(SqliteSyncError::database)?;
73                if let Some((current_sequence, current_hash)) = existing {
74                    if sequence < current_sequence
75                        || (sequence == current_sequence && hash != current_hash)
76                    {
77                        return Err(SqliteSyncError::CorruptRecord("checkpoint conflict"));
78                    }
79                    if sequence == current_sequence {
80                        return Ok(());
81                    }
82                } else if checkpoint_count(&transaction)? >= self.store.config().max_checkpoints {
83                    return Err(SqliteSyncError::CapacityExceeded("checkpoint"));
84                }
85                transaction
86                    .execute(
87                        "INSERT INTO appcore_sync_checkpoint(peer_id, sequence, batch_hash)
88                         VALUES (?1, ?2, ?3)
89                         ON CONFLICT(peer_id) DO UPDATE SET
90                         sequence = excluded.sequence, batch_hash = excluded.batch_hash",
91                        params![peer_id, sequence, hash],
92                    )
93                    .map_err(SqliteSyncError::database)?;
94                transaction.commit().map_err(SqliteSyncError::database)
95            })
96            .map_err(|error| match error {
97                SqliteSyncError::CorruptRecord("checkpoint conflict") => {
98                    SyncError::InvalidSequence(sequence_u64)
99                }
100                other => other.sync(),
101            })
102    }
103}
104
105pub(crate) fn validate_peer_id(peer_id: &str) -> SyncResult<()> {
106    if peer_id.is_empty()
107        || peer_id.len() > MAX_PEER_ID_BYTES
108        || !peer_id.chars().all(|character| {
109            character.is_ascii_alphanumeric() || matches!(character, '.' | '_' | ':' | '-')
110        })
111    {
112        return Err(SyncError::InvalidPeerId);
113    }
114    Ok(())
115}
116
117pub(crate) fn validate_hash(hash: &str) -> SyncResult<()> {
118    if hash.is_empty() || (hash.len() == 64 && hash.bytes().all(|byte| byte.is_ascii_hexdigit())) {
119        Ok(())
120    } else {
121        Err(SyncError::ReplicationFailed(
122            "invalid checkpoint hash".to_string(),
123        ))
124    }
125}
126
127fn checkpoint_count(connection: &rusqlite::Connection) -> Result<usize, SqliteSyncError> {
128    let count: i64 = connection
129        .query_row("SELECT COUNT(*) FROM appcore_sync_checkpoint", [], |row| {
130            row.get(0)
131        })
132        .map_err(SqliteSyncError::database)?;
133    usize::try_from(count).map_err(|_| SqliteSyncError::CorruptRecord("checkpoint count"))
134}
appcore_sync_sqlite/checkpoint.rs

appcore_sync_sqlite/
checkpoint.rs
