appcore_sync_sqlite/
outbox.rs

1// =============================================================================
2//        #######
3//     ###       ###     F: outbox.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::{SyncError, SyncMessage, SyncOutbox, SyncResult};
13use rusqlite::{params, OptionalExtension, TransactionBehavior};
14
15const MAX_BATCH_ID_BYTES: usize = 1_024;
16
17/// SQLite-backed bounded synchronization outbox.
18#[derive(Debug, Clone)]
19pub struct SqliteSyncOutbox {
20    store: SqliteSyncStore,
21}
22
23impl SqliteSyncOutbox {
24    pub(crate) fn new(store: SqliteSyncStore) -> Self {
25        Self { store }
26    }
27
28    fn read_messages(&self, limit: usize) -> SyncResult<Vec<SyncMessage>> {
29        self.store
30            .with_connection(|connection| {
31                let limit = i64::try_from(limit)
32                    .map_err(|_| SqliteSyncError::CapacityExceeded("outbox"))?;
33                let mut statement = connection
34                    .prepare(
35                        "SELECT position, encoded FROM appcore_sync_outbox
36                         ORDER BY position LIMIT ?1",
37                    )
38                    .map_err(SqliteSyncError::database)?;
39                let rows = statement
40                    .query_map([limit], |row| {
41                        Ok((row.get::<_, i64>(0)?, row.get::<_, Vec<u8>>(1)?))
42                    })
43                    .map_err(SqliteSyncError::database)?;
44                let mut messages = Vec::new();
45                let mut bytes = 0usize;
46                for row in rows {
47                    let (_position, encoded) = row.map_err(SqliteSyncError::database)?;
48                    bytes = bytes
49                        .checked_add(encoded.len())
50                        .ok_or(SqliteSyncError::CapacityExceeded("outbox byte"))?;
51                    if bytes as u64 > self.store.config().max_database_bytes {
52                        return Err(SqliteSyncError::CapacityExceeded("outbox byte"));
53                    }
54                    let message = serde_json::from_slice(&encoded)
55                        .map_err(|_| SqliteSyncError::CorruptRecord("outbox"))?;
56                    messages.push(message);
57                }
58                Ok(messages)
59            })
60            .map_err(SqliteSyncError::sync)
61    }
62}
63
64impl SyncOutbox for SqliteSyncOutbox {
65    fn try_enqueue(&self, message: SyncMessage, max_len: usize) -> SyncResult<bool> {
66        validate_batch_id(&message.batch_id)?;
67        let encoded = serde_json::to_vec(&message)
68            .map_err(|_| SyncError::InvalidSyncMessage("outbox serialization failed"))?;
69        if encoded.len() > self.store.config().max_outbox_record_bytes {
70            return Err(SqliteSyncError::CapacityExceeded("outbox record").sync());
71        }
72        let limit = max_len.min(self.store.config().max_outbox_entries);
73        self.store
74            .with_connection(|connection| {
75                let transaction = connection
76                    .transaction_with_behavior(TransactionBehavior::Immediate)
77                    .map_err(SqliteSyncError::database)?;
78                let existing: Option<Vec<u8>> = transaction
79                    .query_row(
80                        "SELECT encoded FROM appcore_sync_outbox WHERE batch_id = ?1",
81                        [&message.batch_id],
82                        |row| row.get(0),
83                    )
84                    .optional()
85                    .map_err(SqliteSyncError::database)?;
86                if let Some(existing) = existing {
87                    return if existing == encoded {
88                        Ok(false)
89                    } else {
90                        Err(SqliteSyncError::CorruptRecord("outbox conflict"))
91                    };
92                }
93                let count = count_entries(&transaction)?;
94                if count >= limit {
95                    return Ok(false);
96                }
97                let inserted = transaction
98                    .execute(
99                        "INSERT INTO appcore_sync_outbox(batch_id, encoded)
100                         VALUES (?1, ?2)",
101                        params![message.batch_id, encoded],
102                    )
103                    .map_err(SqliteSyncError::database)?;
104                transaction.commit().map_err(SqliteSyncError::database)?;
105                Ok(inserted == 1)
106            })
107            .map_err(|error| match error {
108                SqliteSyncError::CorruptRecord("outbox conflict") => {
109                    SyncError::InvalidSyncMessage("outbox batch conflict")
110                }
111                other => other.sync(),
112            })
113    }
114
115    fn front(&self) -> SyncResult<Option<SyncMessage>> {
116        Ok(self.read_messages(1)?.into_iter().next())
117    }
118
119    fn acknowledge_front(&self, batch_id: &str) -> SyncResult<()> {
120        validate_batch_id(batch_id)?;
121        self.store
122            .with_connection(|connection| {
123                let transaction = connection
124                    .transaction_with_behavior(TransactionBehavior::Immediate)
125                    .map_err(SqliteSyncError::database)?;
126                let front: Option<(i64, String)> = transaction
127                    .query_row(
128                        "SELECT position, batch_id FROM appcore_sync_outbox
129                         ORDER BY position LIMIT 1",
130                        [],
131                        |row| Ok((row.get(0)?, row.get(1)?)),
132                    )
133                    .optional()
134                    .map_err(SqliteSyncError::database)?;
135                let Some((position, current_id)) = front else {
136                    return Err(SqliteSyncError::CorruptRecord("outbox acknowledgement"));
137                };
138                if current_id != batch_id {
139                    return Err(SqliteSyncError::CorruptRecord("outbox acknowledgement"));
140                }
141                transaction
142                    .execute(
143                        "DELETE FROM appcore_sync_outbox WHERE position = ?1",
144                        [position],
145                    )
146                    .map_err(SqliteSyncError::database)?;
147                transaction.commit().map_err(SqliteSyncError::database)
148            })
149            .map_err(|error| match error {
150                SqliteSyncError::CorruptRecord("outbox acknowledgement") => {
151                    SyncError::InvalidSyncMessage("outbox acknowledgement mismatch")
152                }
153                other => other.sync(),
154            })
155    }
156
157    fn messages(&self) -> SyncResult<Vec<SyncMessage>> {
158        let length = self.len()?;
159        if length > self.store.config().max_outbox_entries {
160            return Err(SqliteSyncError::CapacityExceeded("outbox").sync());
161        }
162        self.read_messages(length)
163    }
164
165    fn len(&self) -> SyncResult<usize> {
166        self.store
167            .with_connection(|connection| count_entries(connection))
168            .map_err(SqliteSyncError::sync)
169    }
170}
171
172fn count_entries(connection: &rusqlite::Connection) -> Result<usize, SqliteSyncError> {
173    let count: i64 = connection
174        .query_row("SELECT COUNT(*) FROM appcore_sync_outbox", [], |row| {
175            row.get(0)
176        })
177        .map_err(SqliteSyncError::database)?;
178    usize::try_from(count).map_err(|_| SqliteSyncError::CorruptRecord("outbox count"))
179}
180
181pub(crate) fn validate_batch_id(batch_id: &str) -> SyncResult<()> {
182    if batch_id.is_empty()
183        || batch_id.len() > MAX_BATCH_ID_BYTES
184        || batch_id.chars().any(char::is_control)
185    {
186        return Err(SyncError::InvalidSyncMessage("invalid outbox batch id"));
187    }
188    Ok(())
189}
appcore_sync_sqlite/outbox.rs

appcore_sync_sqlite/
outbox.rs
