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

appcore_sync_sqlite/
checkpoint.rs
