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                set_in_transaction(
68                    &transaction,
69                    peer_id,
70                    sequence,
71                    hash,
72                    self.store.config().max_checkpoints,
73                )?;
74                transaction.commit().map_err(SqliteSyncError::database)
75            })
76            .map_err(|error| match error {
77                SqliteSyncError::CorruptRecord("checkpoint conflict") => {
78                    SyncError::InvalidSequence(sequence_u64)
79                }
80                other => other.sync(),
81            })
82    }
83}
84
85pub(crate) fn validate_peer_id(peer_id: &str) -> SyncResult<()> {
86    if peer_id.is_empty()
87        || peer_id.len() > MAX_PEER_ID_BYTES
88        || !peer_id.chars().all(|character| {
89            character.is_ascii_alphanumeric() || matches!(character, '.' | '_' | ':' | '-')
90        })
91    {
92        return Err(SyncError::InvalidPeerId);
93    }
94    Ok(())
95}
96
97pub(crate) fn validate_hash(hash: &str) -> SyncResult<()> {
98    if hash.is_empty() || (hash.len() == 64 && hash.bytes().all(|byte| byte.is_ascii_hexdigit())) {
99        Ok(())
100    } else {
101        Err(SyncError::ReplicationFailed(
102            "invalid checkpoint hash".to_string(),
103        ))
104    }
105}
106
107fn checkpoint_count(connection: &rusqlite::Connection) -> Result<usize, SqliteSyncError> {
108    let count: i64 = connection
109        .query_row("SELECT COUNT(*) FROM appcore_sync_checkpoint", [], |row| {
110            row.get(0)
111        })
112        .map_err(SqliteSyncError::database)?;
113    usize::try_from(count).map_err(|_| SqliteSyncError::CorruptRecord("checkpoint count"))
114}
115
116pub(crate) fn set_in_transaction(
117    transaction: &rusqlite::Transaction<'_>,
118    peer_id: &str,
119    sequence: i64,
120    hash: &str,
121    max_checkpoints: usize,
122) -> crate::SqliteSyncResult<()> {
123    let existing: Option<(i64, String)> = transaction
124        .query_row(
125            "SELECT sequence, batch_hash FROM appcore_sync_checkpoint WHERE peer_id = ?1",
126            [peer_id],
127            |row| Ok((row.get(0)?, row.get(1)?)),
128        )
129        .optional()
130        .map_err(SqliteSyncError::database)?;
131    if let Some((current_sequence, current_hash)) = existing {
132        if sequence < current_sequence || (sequence == current_sequence && hash != current_hash) {
133            return Err(SqliteSyncError::CorruptRecord("checkpoint conflict"));
134        }
135        if sequence == current_sequence {
136            return Ok(());
137        }
138    } else if checkpoint_count(transaction)? >= max_checkpoints {
139        return Err(SqliteSyncError::CapacityExceeded("checkpoint"));
140    }
141    transaction
142        .execute(
143            "INSERT INTO appcore_sync_checkpoint(peer_id, sequence, batch_hash)
144                         VALUES (?1, ?2, ?3)
145                         ON CONFLICT(peer_id) DO UPDATE SET
146                         sequence = excluded.sequence, batch_hash = excluded.batch_hash",
147            params![peer_id, sequence, hash],
148        )
149        .map_err(SqliteSyncError::database)?;
150    Ok(())
151}
appcore_sync_sqlite/checkpoint.rs

appcore_sync_sqlite/
checkpoint.rs
