appcore_sync_sqlite/
log.rs

1// =============================================================================
2//        #######
3//     ###       ###     F: log.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::{
13    InMemoryReplicationLog, ReplicationLog, ReplicationSnapshot, SyncError, SyncResult,
14    REPLICATION_LOG_FORMAT_V1,
15};
16use rusqlite::{params, OptionalExtension, Transaction, TransactionBehavior};
17use sha2::{Digest, Sha256};
18
19pub(crate) const MAX_REPLICATION_RECORD_BYTES: usize = 1024 * 1024;
20
21/// SQLite-backed implementation of the AppCore replication-log contract.
22#[derive(Debug, Clone)]
23pub struct SqliteReplicationLog {
24    store: SqliteSyncStore,
25}
26
27impl SqliteReplicationLog {
28    pub(crate) fn new(store: SqliteSyncStore) -> Self {
29        Self { store }
30    }
31
32    /// Reads one bounded page after the supplied zero-based log offset.
33    pub fn events_page(&self, index: usize, max_records: usize) -> SyncResult<Vec<Vec<u8>>> {
34        if max_records == 0 || max_records > self.store.config().max_read_records {
35            return Err(capacity_error("read record"));
36        }
37        self.store
38            .with_connection(|connection| {
39                let length = count_records(connection)?;
40                if index > length {
41                    return Err(SqliteSyncError::CorruptRecord("log index"));
42                }
43                let start = i64::try_from(index)
44                    .map_err(|_| SqliteSyncError::CapacityExceeded("log index"))?;
45                let limit = i64::try_from(max_records)
46                    .map_err(|_| SqliteSyncError::CapacityExceeded("read record"))?;
47                let mut statement = connection
48                    .prepare(
49                        "SELECT payload FROM appcore_replication_log
50                         WHERE log_index > ?1 ORDER BY log_index LIMIT ?2",
51                    )
52                    .map_err(SqliteSyncError::database)?;
53                let rows = statement
54                    .query_map(params![start, limit], |row| row.get::<_, Vec<u8>>(0))
55                    .map_err(SqliteSyncError::database)?;
56                let mut payloads = Vec::with_capacity(max_records.min(length - index));
57                let mut bytes = 0usize;
58                for row in rows {
59                    let payload = row.map_err(SqliteSyncError::database)?;
60                    bytes = bytes
61                        .checked_add(payload.len())
62                        .ok_or(SqliteSyncError::CapacityExceeded("read byte"))?;
63                    if bytes > self.store.config().max_read_bytes {
64                        return Err(SqliteSyncError::CapacityExceeded("read byte"));
65                    }
66                    payloads.push(payload);
67                }
68                Ok(payloads)
69            })
70            .map_err(SqliteSyncError::sync)
71    }
72
73    fn append_record(&self, payload: Vec<u8>, sequence: u64) -> SyncResult<usize> {
74        validate_payload(&payload)?;
75        self.store
76            .with_connection(|connection| {
77                let transaction = connection
78                    .transaction_with_behavior(TransactionBehavior::Immediate)
79                    .map_err(SqliteSyncError::database)?;
80                if let Some(existing) = existing_sequence(&transaction, sequence)? {
81                    return if existing.1 == payload {
82                        Ok(existing.0)
83                    } else {
84                        Err(SqliteSyncError::CorruptRecord("sequence conflict"))
85                    };
86                }
87                let previous_hash: String = transaction
88                    .query_row(
89                        "SELECT record_hash FROM appcore_replication_log
90                         ORDER BY log_index DESC LIMIT 1",
91                        [],
92                        |row| row.get(0),
93                    )
94                    .optional()
95                    .map_err(SqliteSyncError::database)?
96                    .unwrap_or_default();
97                let record_hash = record_hash(&previous_hash, sequence, &payload);
98                transaction
99                    .execute(
100                        "INSERT INTO appcore_replication_log
101                         (source_sequence, payload, previous_hash, record_hash)
102                         VALUES (?1, ?2, ?3, ?4)",
103                        params![
104                            sequence_to_i64(sequence)?,
105                            payload,
106                            previous_hash,
107                            record_hash
108                        ],
109                    )
110                    .map_err(SqliteSyncError::database)?;
111                let index = usize::try_from(transaction.last_insert_rowid())
112                    .map_err(|_| SqliteSyncError::CapacityExceeded("log index"))?;
113                transaction.commit().map_err(SqliteSyncError::database)?;
114                Ok(index)
115            })
116            .map_err(|error| match error {
117                SqliteSyncError::CorruptRecord("sequence conflict") => {
118                    SyncError::SequenceConflict(sequence)
119                }
120                other => other.sync(),
121            })
122    }
123
124    fn snapshot_records(&self) -> SyncResult<Vec<(u64, Vec<u8>)>> {
125        self.store
126            .with_connection(|connection| {
127                let mut statement = connection
128                    .prepare(
129                        "SELECT source_sequence, payload FROM appcore_replication_log
130                         ORDER BY log_index",
131                    )
132                    .map_err(SqliteSyncError::database)?;
133                let rows = statement
134                    .query_map([], |row| {
135                        Ok((row.get::<_, i64>(0)?, row.get::<_, Vec<u8>>(1)?))
136                    })
137                    .map_err(SqliteSyncError::database)?;
138                let mut records = Vec::new();
139                let mut bytes = 0usize;
140                for row in rows {
141                    let (sequence, payload) = row.map_err(SqliteSyncError::database)?;
142                    bytes = bytes
143                        .checked_add(payload.len())
144                        .ok_or(SqliteSyncError::CapacityExceeded("snapshot byte"))?;
145                    if bytes as u64 > self.store.config().max_database_bytes {
146                        return Err(SqliteSyncError::CapacityExceeded("snapshot byte"));
147                    }
148                    records.push((
149                        u64::try_from(sequence)
150                            .map_err(|_| SqliteSyncError::CorruptRecord("sequence"))?,
151                        payload,
152                    ));
153                }
154                Ok(records)
155            })
156            .map_err(SqliteSyncError::sync)
157    }
158}
159
160impl ReplicationLog for SqliteReplicationLog {
161    fn append(&mut self, record: Vec<u8>) -> SyncResult<usize> {
162        self.append_record(record, 0)
163    }
164
165    fn append_with_sequence(&mut self, record: Vec<u8>, sequence: u64) -> SyncResult<usize> {
166        self.append_record(record, sequence)
167    }
168
169    fn event_at_sequence(&self, sequence: u64) -> SyncResult<Option<Vec<u8>>> {
170        if sequence == 0 {
171            return Ok(None);
172        }
173        self.store
174            .with_connection(|connection| {
175                connection
176                    .query_row(
177                        "SELECT payload FROM appcore_replication_log WHERE source_sequence = ?1",
178                        [sequence_to_i64(sequence)?],
179                        |row| row.get(0),
180                    )
181                    .optional()
182                    .map_err(SqliteSyncError::database)
183            })
184            .map_err(SqliteSyncError::sync)
185    }
186
187    fn events_since(&self, index: usize) -> SyncResult<Vec<Vec<u8>>> {
188        let length = self.len()?;
189        if length.saturating_sub(index) > self.store.config().max_read_records {
190            return Err(capacity_error("read record"));
191        }
192        self.events_page(index, self.store.config().max_read_records)
193    }
194
195    fn last_index(&self) -> SyncResult<usize> {
196        self.len()
197    }
198
199    fn len(&self) -> SyncResult<usize> {
200        self.store
201            .with_connection(|connection| count_records(connection))
202            .map_err(SqliteSyncError::sync)
203    }
204
205    fn is_empty(&self) -> SyncResult<bool> {
206        self.len().map(|length| length == 0)
207    }
208
209    fn create_snapshot(&self) -> SyncResult<ReplicationSnapshot> {
210        let mut memory = InMemoryReplicationLog::new();
211        for (sequence, payload) in self.snapshot_records()? {
212            let _ = memory.append_with_sequence(payload, sequence)?;
213        }
214        memory.create_snapshot()
215    }
216
217    fn restore_snapshot(&mut self, snapshot: &ReplicationSnapshot) -> SyncResult<()> {
218        let mut validated = InMemoryReplicationLog::new();
219        validated.restore_snapshot(snapshot)?;
220        let records = snapshot.records.clone();
221        self.store
222            .with_connection(|connection| {
223                let transaction = connection
224                    .transaction_with_behavior(TransactionBehavior::Immediate)
225                    .map_err(SqliteSyncError::database)?;
226                transaction
227                    .execute("DELETE FROM appcore_replication_log", [])
228                    .map_err(SqliteSyncError::database)?;
229                let mut previous_hash = String::new();
230                for (offset, record) in records.iter().enumerate() {
231                    validate_payload(&record.payload)
232                        .map_err(|_| SqliteSyncError::CapacityExceeded("replication record"))?;
233                    let hash = record_hash(&previous_hash, record.sequence, &record.payload);
234                    transaction
235                        .execute(
236                            "INSERT INTO appcore_replication_log
237                             (log_index, source_sequence, payload, previous_hash, record_hash)
238                             VALUES (?1, ?2, ?3, ?4, ?5)",
239                            params![
240                                i64::try_from(offset + 1).map_err(|_| {
241                                    SqliteSyncError::CapacityExceeded("log index")
242                                })?,
243                                sequence_to_i64(record.sequence)?,
244                                &record.payload,
245                                previous_hash,
246                                hash
247                            ],
248                        )
249                        .map_err(SqliteSyncError::database)?;
250                    previous_hash = hash;
251                }
252                transaction.commit().map_err(SqliteSyncError::database)
253            })
254            .map_err(SqliteSyncError::sync)
255    }
256}
257
258fn existing_sequence(
259    transaction: &Transaction<'_>,
260    sequence: u64,
261) -> Result<Option<(usize, Vec<u8>)>, SqliteSyncError> {
262    if sequence == 0 {
263        return Ok(None);
264    }
265    transaction
266        .query_row(
267            "SELECT log_index, payload FROM appcore_replication_log WHERE source_sequence = ?1",
268            [sequence_to_i64(sequence)?],
269            |row| Ok((row.get::<_, i64>(0)?, row.get(1)?)),
270        )
271        .optional()
272        .map_err(SqliteSyncError::database)?
273        .map(|(index, payload)| {
274            usize::try_from(index)
275                .map(|index| (index, payload))
276                .map_err(|_| SqliteSyncError::CorruptRecord("log index"))
277        })
278        .transpose()
279}
280
281fn count_records(connection: &rusqlite::Connection) -> Result<usize, SqliteSyncError> {
282    let count: i64 = connection
283        .query_row("SELECT COUNT(*) FROM appcore_replication_log", [], |row| {
284            row.get(0)
285        })
286        .map_err(SqliteSyncError::database)?;
287    usize::try_from(count).map_err(|_| SqliteSyncError::CorruptRecord("log count"))
288}
289
290fn validate_payload(payload: &[u8]) -> SyncResult<()> {
291    if payload.len() > MAX_REPLICATION_RECORD_BYTES {
292        return Err(capacity_error("replication record"));
293    }
294    Ok(())
295}
296
297fn sequence_to_i64(sequence: u64) -> Result<i64, SqliteSyncError> {
298    i64::try_from(sequence).map_err(|_| SqliteSyncError::CapacityExceeded("sequence"))
299}
300
301fn capacity_error(resource: &'static str) -> SyncError {
302    SqliteSyncError::CapacityExceeded(resource).sync()
303}
304
305pub(crate) fn record_hash(previous_hash: &str, sequence: u64, payload: &[u8]) -> String {
306    let mut hasher = Sha256::new();
307    hasher.update(REPLICATION_LOG_FORMAT_V1.as_bytes());
308    hasher.update((previous_hash.len() as u64).to_be_bytes());
309    hasher.update(previous_hash.as_bytes());
310    hasher.update(sequence.to_be_bytes());
311    hasher.update((payload.len() as u64).to_be_bytes());
312    hasher.update(payload);
313    let digest = hasher.finalize();
314    let mut output = String::with_capacity(digest.len() * 2);
315    for byte in digest {
316        use std::fmt::Write as _;
317        let _ = write!(output, "{byte:02x}");
318    }
319    output
320}
appcore_sync_sqlite/log.rs

appcore_sync_sqlite/
log.rs
