appcore_sync_sqlite/
store.rs

1// =============================================================================
2//        #######
3//     ###       ###     F: store.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::integrity::validate_internal_records;
12use crate::schema;
13use crate::{
14    SqliteReplicationLog, SqliteSyncCheckpointStore, SqliteSyncConfig, SqliteSyncError,
15    SqliteSyncOutbox, SqliteSyncResult, SqliteSyncTombstoneStore,
16};
17use appcore_contracts::ProviderId;
18use appcore_storage::{
19    StorageCapabilityDescriptorV1, StorageCapabilityProviderV1, StorageCapabilityV1,
20};
21use parking_lot::{Condvar, Mutex};
22use rusqlite::limits::Limit;
23use rusqlite::{Connection, OpenFlags};
24use std::fmt;
25use std::fs;
26use std::path::Path;
27use std::sync::Arc;
28use std::time::{Duration, Instant};
29
30/// Result of a provider integrity inspection.
31#[derive(Debug, Clone, Copy, PartialEq, Eq)]
32pub struct SqliteSyncHealth {
33    /// Internal schema version observed by the provider.
34    pub schema_version: u32,
35    /// Database pages currently allocated.
36    pub page_count: u64,
37    /// Configured maximum page count.
38    pub max_page_count: u64,
39}
40
41struct SqliteSyncInner {
42    pool: Mutex<ConnectionPool>,
43    available: Condvar,
44    config: SqliteSyncConfig,
45}
46
47struct ConnectionPool {
48    idle: Vec<Connection>,
49    total: usize,
50}
51
52struct ConnectionGuard<'a> {
53    inner: &'a SqliteSyncInner,
54    connection: Option<Connection>,
55}
56
57/// Shared owner of one bounded SQLite sync database.
58#[derive(Clone)]
59pub struct SqliteSyncStore {
60    inner: Arc<SqliteSyncInner>,
61}
62
63impl SqliteSyncStore {
64    /// Opens, migrates and integrity-checks one provider database.
65    pub fn open(mut config: SqliteSyncConfig) -> SqliteSyncResult<Self> {
66        config.validate()?;
67        config.path = normalize_path(&config.path)?;
68        let flags = OpenFlags::SQLITE_OPEN_READ_WRITE
69            | OpenFlags::SQLITE_OPEN_CREATE
70            | OpenFlags::SQLITE_OPEN_FULL_MUTEX;
71        let mut connection =
72            Connection::open_with_flags(&config.path, flags).map_err(SqliteSyncError::database)?;
73        configure(&connection, &config)?;
74        schema::migrate(&mut connection)?;
75        integrity_check(&connection, &config)?;
76        Ok(Self {
77            inner: Arc::new(SqliteSyncInner {
78                pool: Mutex::new(ConnectionPool {
79                    idle: vec![connection],
80                    total: 1,
81                }),
82                available: Condvar::new(),
83                config,
84            }),
85        })
86    }
87
88    /// Runs an integrity check and reports bounded database usage.
89    pub fn health(&self) -> SqliteSyncResult<SqliteSyncHealth> {
90        self.with_connection(|connection| {
91            integrity_check(connection, &self.inner.config)?;
92            let schema_version = connection
93                .pragma_query_value(None, "user_version", |row| row.get(0))
94                .map_err(SqliteSyncError::database)?;
95            let page_count: i64 = connection
96                .pragma_query_value(None, "page_count", |row| row.get(0))
97                .map_err(SqliteSyncError::database)?;
98            let max_page_count: i64 = connection
99                .pragma_query_value(None, "max_page_count", |row| row.get(0))
100                .map_err(SqliteSyncError::database)?;
101            Ok(SqliteSyncHealth {
102                schema_version,
103                page_count: u64::try_from(page_count)
104                    .map_err(|_| SqliteSyncError::CorruptRecord("page count"))?,
105                max_page_count: u64::try_from(max_page_count)
106                    .map_err(|_| SqliteSyncError::CorruptRecord("page limit"))?,
107            })
108        })
109    }
110
111    /// Returns the provider's redacted configuration bounds.
112    pub fn config(&self) -> &SqliteSyncConfig {
113        &self.inner.config
114    }
115
116    /// Creates a replication-log handle backed by this database.
117    pub fn replication_log(&self) -> SqliteReplicationLog {
118        SqliteReplicationLog::new(self.clone())
119    }
120
121    /// Creates a checkpoint-store handle backed by this database.
122    pub fn checkpoint_store(&self) -> SqliteSyncCheckpointStore {
123        SqliteSyncCheckpointStore::new(self.clone())
124    }
125
126    /// Creates an outbox handle backed by this database.
127    pub fn outbox(&self) -> SqliteSyncOutbox {
128        SqliteSyncOutbox::new(self.clone())
129    }
130
131    /// Creates an opaque tombstone-store handle backed by this database.
132    pub fn tombstone_store(&self) -> SqliteSyncTombstoneStore {
133        SqliteSyncTombstoneStore::new(self.clone())
134    }
135
136    pub(crate) fn with_connection<T>(
137        &self,
138        action: impl FnOnce(&mut Connection) -> SqliteSyncResult<T>,
139    ) -> SqliteSyncResult<T> {
140        let mut connection = self.acquire_connection()?;
141        action(connection.connection_mut()?)
142    }
143
144    fn acquire_connection(&self) -> SqliteSyncResult<ConnectionGuard<'_>> {
145        let deadline = Instant::now() + Duration::from_millis(self.inner.config.busy_timeout_ms);
146        let mut pool = self.inner.pool.lock();
147        loop {
148            if let Some(connection) = pool.idle.pop() {
149                return Ok(ConnectionGuard {
150                    inner: &self.inner,
151                    connection: Some(connection),
152                });
153            }
154            if pool.total < self.inner.config.max_connections {
155                pool.total += 1;
156                drop(pool);
157                return match open_connection(&self.inner.config) {
158                    Ok(connection) => Ok(ConnectionGuard {
159                        inner: &self.inner,
160                        connection: Some(connection),
161                    }),
162                    Err(error) => {
163                        let mut pool = self.inner.pool.lock();
164                        pool.total = pool.total.saturating_sub(1);
165                        self.inner.available.notify_one();
166                        Err(error)
167                    }
168                };
169            }
170            let now = Instant::now();
171            if now >= deadline {
172                return Err(SqliteSyncError::CapacityExceeded("connection"));
173            }
174            self.inner.available.wait_for(&mut pool, deadline - now);
175        }
176    }
177}
178
179impl ConnectionGuard<'_> {
180    fn connection_mut(&mut self) -> SqliteSyncResult<&mut Connection> {
181        self.connection
182            .as_mut()
183            .ok_or(SqliteSyncError::DatabaseOperation)
184    }
185}
186
187impl Drop for ConnectionGuard<'_> {
188    fn drop(&mut self) {
189        if let Some(connection) = self.connection.take() {
190            self.inner.pool.lock().idle.push(connection);
191            self.inner.available.notify_one();
192        }
193    }
194}
195
196impl fmt::Debug for SqliteSyncStore {
197    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
198        formatter
199            .debug_struct("SqliteSyncStore")
200            .field("config", &self.inner.config)
201            .finish_non_exhaustive()
202    }
203}
204
205impl StorageCapabilityProviderV1 for SqliteSyncStore {
206    fn storage_capabilities_v1(
207        &self,
208    ) -> Result<StorageCapabilityDescriptorV1, appcore_storage::StorageCapabilityError> {
209        sqlite_sync_capability_descriptor_v1()
210    }
211}
212
213/// Returns the conservative provider-independent guarantees for SQLite sync.
214pub fn sqlite_sync_capability_descriptor_v1(
215) -> Result<StorageCapabilityDescriptorV1, appcore_storage::StorageCapabilityError> {
216    let provider_id = ProviderId::new("sqlite-sync")
217        .map_err(|_| appcore_storage::StorageCapabilityError::InvalidDescriptor)?;
218    Ok(StorageCapabilityDescriptorV1::new(
219        provider_id,
220        [
221            StorageCapabilityV1::Transactions,
222            StorageCapabilityV1::Locking,
223            StorageCapabilityV1::Snapshot,
224            StorageCapabilityV1::OnlineBackup,
225            StorageCapabilityV1::MultiProcess,
226        ],
227    ))
228}
229
230fn configure(connection: &Connection, config: &SqliteSyncConfig) -> SqliteSyncResult<()> {
231    connection
232        .busy_timeout(Duration::from_millis(config.busy_timeout_ms))
233        .map_err(SqliteSyncError::database)?;
234    connection
235        .pragma_update_and_check(None, "journal_mode", "WAL", |row| row.get::<_, String>(0))
236        .map_err(SqliteSyncError::database)
237        .and_then(|mode| {
238            if mode.eq_ignore_ascii_case("wal") {
239                Ok(())
240            } else {
241                Err(SqliteSyncError::DatabaseOperation)
242            }
243        })?;
244    connection
245        .pragma_update(None, "synchronous", "FULL")
246        .and_then(|_| connection.pragma_update(None, "foreign_keys", true))
247        .and_then(|_| connection.pragma_update(None, "trusted_schema", false))
248        .and_then(|_| connection.pragma_update(None, "wal_autocheckpoint", 1_000))
249        .map_err(SqliteSyncError::database)?;
250    let max_length = i32::try_from(config.max_outbox_record_bytes)
251        .map_err(|_| SqliteSyncError::InvalidConfiguration("outbox record bound overflow"))?;
252    connection
253        .set_limit(Limit::SQLITE_LIMIT_LENGTH, max_length)
254        .and_then(|_| connection.set_limit(Limit::SQLITE_LIMIT_ATTACHED, 0))
255        .and_then(|_| connection.set_limit(Limit::SQLITE_LIMIT_WORKER_THREADS, 0))
256        .map_err(SqliteSyncError::database)?;
257    let page_bytes: i64 = connection
258        .pragma_query_value(None, "page_size", |row| row.get(0))
259        .map_err(SqliteSyncError::database)?;
260    let page_bytes =
261        u64::try_from(page_bytes).map_err(|_| SqliteSyncError::CorruptRecord("page size"))?;
262    if page_bytes == 0 {
263        return Err(SqliteSyncError::CorruptRecord("page size"));
264    }
265    let max_pages = i64::try_from(config.max_database_bytes / page_bytes)
266        .map_err(|_| SqliteSyncError::InvalidConfiguration("database page bound overflow"))?;
267    connection
268        .pragma_update(None, "max_page_count", max_pages)
269        .map_err(SqliteSyncError::database)
270}
271
272fn open_connection(config: &SqliteSyncConfig) -> SqliteSyncResult<Connection> {
273    let flags = OpenFlags::SQLITE_OPEN_READ_WRITE
274        | OpenFlags::SQLITE_OPEN_CREATE
275        | OpenFlags::SQLITE_OPEN_FULL_MUTEX;
276    let connection =
277        Connection::open_with_flags(&config.path, flags).map_err(SqliteSyncError::database)?;
278    configure(&connection, config)?;
279    Ok(connection)
280}
281
282fn integrity_check(connection: &Connection, config: &SqliteSyncConfig) -> SqliteSyncResult<()> {
283    let result: String = connection
284        .query_row("PRAGMA quick_check(1)", [], |row| row.get(0))
285        .map_err(SqliteSyncError::database)?;
286    if result != "ok" {
287        return Err(SqliteSyncError::IntegrityFailed);
288    }
289    validate_internal_records(connection, config)
290}
291
292pub(crate) fn normalize_path(path: &Path) -> SqliteSyncResult<std::path::PathBuf> {
293    let file_name = path.file_name().ok_or(SqliteSyncError::UnsafePath)?;
294    match fs::symlink_metadata(path) {
295        Ok(metadata) if metadata.file_type().is_symlink() || !metadata.is_file() => {
296            return Err(SqliteSyncError::UnsafePath);
297        }
298        Ok(_) => {}
299        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
300        Err(_) => return Err(SqliteSyncError::UnsafePath),
301    }
302    let parent = path.parent().unwrap_or_else(|| Path::new("."));
303    fs::create_dir_all(parent).map_err(|_| SqliteSyncError::UnsafePath)?;
304    let canonical_parent = fs::canonicalize(parent).map_err(|_| SqliteSyncError::UnsafePath)?;
305    if !canonical_parent.is_dir() {
306        return Err(SqliteSyncError::UnsafePath);
307    }
308    Ok(canonical_parent.join(file_name))
309}
appcore_sync_sqlite/store.rs

appcore_sync_sqlite/
store.rs
