core_crypto_keystore/connection/
mod.rs

1pub mod platform;
2
3use std::{
4    borrow::Borrow,
5    ops::{Deref, DerefMut},
6    sync::Arc,
7};
8
9use async_lock::{Mutex, MutexGuard, Semaphore};
10use async_trait::async_trait;
11
12pub use self::platform::*;
13use crate::{
14    CryptoKeystoreError, CryptoKeystoreResult, DatabaseKey,
15    entities::{MlsPendingMessage, PersistedMlsGroup},
16    traits::{
17        BorrowPrimaryKey, Entity, EntityDatabaseMutation, EntityDeleteBorrowed, EntityGetBorrowed, FetchFromDatabase,
18        KeyType, SearchableEntity,
19    },
20    transaction::KeystoreTransaction,
21};
22
23/// Limit on the length of a blob to be stored in the database.
24///
25/// This limit applies to both SQLCipher-backed stores and WASM.
26/// This limit is conservative on purpose when targeting WASM, as the lower bound that exists is Safari with a limit of
27/// 1GB per origin.
28///
29/// See: [SQLite limits](https://www.sqlite.org/limits.html)
30/// See: [IndexedDB limits](https://stackoverflow.com/a/63019999/1934177)
31pub const MAX_BLOB_LEN: usize = 1_000_000_000;
32
33#[cfg(not(target_os = "unknown"))]
34// ? Because of UniFFI async requirements, we need our keystore to be Send as well now
35pub trait DatabaseConnectionRequirements: Sized + Send {}
36#[cfg(target_os = "unknown")]
37// ? On the other hand, things cannot be Send on WASM because of platform restrictions (all things are copied across the
38// FFI)
39pub trait DatabaseConnectionRequirements: Sized {}
40
41#[cfg_attr(target_os = "unknown", async_trait::async_trait(?Send))]
42#[cfg_attr(not(target_os = "unknown"), async_trait::async_trait)]
43pub trait DatabaseConnection<'a>: DatabaseConnectionRequirements {
44    type Connection: 'a;
45
46    async fn open(location: &str, key: &DatabaseKey) -> CryptoKeystoreResult<Self>;
47
48    async fn open_in_memory(key: &DatabaseKey) -> CryptoKeystoreResult<Self>;
49
50    async fn update_key(&mut self, new_key: &DatabaseKey) -> CryptoKeystoreResult<()>;
51
52    /// Clear all data from the database and close it.
53    async fn wipe(self) -> CryptoKeystoreResult<()>;
54
55    fn check_buffer_size(size: usize) -> CryptoKeystoreResult<()> {
56        #[cfg(not(target_os = "unknown"))]
57        if size > i32::MAX as usize {
58            return Err(CryptoKeystoreError::BlobTooBig);
59        }
60
61        if size >= MAX_BLOB_LEN {
62            return Err(CryptoKeystoreError::BlobTooBig);
63        }
64
65        Ok(())
66    }
67
68    /// Returns the database location if persistent or None if in-memory
69    fn location(&self) -> Option<&str>;
70}
71
72#[derive(Debug, Clone)]
73pub struct Database {
74    pub(crate) conn: Arc<Mutex<Option<KeystoreDatabaseConnection>>>,
75    pub(crate) transaction: Arc<Mutex<Option<KeystoreTransaction>>>,
76    transaction_semaphore: Arc<Semaphore>,
77}
78
79const ALLOWED_CONCURRENT_TRANSACTIONS_COUNT: usize = 1;
80
81// SAFETY: this has mutexes and atomics protecting underlying data so this is safe to share between threads
82unsafe impl Send for Database {}
83// SAFETY: this has mutexes and atomics protecting underlying data so this is safe to share between threads
84unsafe impl Sync for Database {}
85
86/// Where to open a connection
87#[derive(Debug, Clone)]
88pub enum ConnectionType<'a> {
89    /// This connection is persistent at the provided path
90    Persistent(&'a str),
91    /// This connection is transient and lives in memory
92    InMemory,
93}
94
95/// Exclusive access to the database connection
96///
97/// Note that this is only ever constructed when we already hold exclusive access,
98/// and the connection has already been tested to ensure that it is non-empty.
99pub struct ConnectionGuard<'a> {
100    guard: MutexGuard<'a, Option<KeystoreDatabaseConnection>>,
101}
102
103impl<'a> TryFrom<MutexGuard<'a, Option<KeystoreDatabaseConnection>>> for ConnectionGuard<'a> {
104    type Error = CryptoKeystoreError;
105
106    fn try_from(guard: MutexGuard<'a, Option<KeystoreDatabaseConnection>>) -> Result<Self, Self::Error> {
107        guard
108            .is_some()
109            .then_some(Self { guard })
110            .ok_or(CryptoKeystoreError::Closed)
111    }
112}
113
114impl Deref for ConnectionGuard<'_> {
115    type Target = KeystoreDatabaseConnection;
116
117    fn deref(&self) -> &Self::Target {
118        self.guard
119            .as_ref()
120            .expect("we have exclusive access and already checked that the connection exists")
121    }
122}
123
124impl DerefMut for ConnectionGuard<'_> {
125    fn deref_mut(&mut self) -> &mut Self::Target {
126        self.guard
127            .as_mut()
128            .expect("we have exclusive access and already checked that the connection exists")
129    }
130}
131
132// Only the functions in this impl block directly mess with `self.conn`
133impl Database {
134    pub async fn open(location: ConnectionType<'_>, key: &DatabaseKey) -> CryptoKeystoreResult<Self> {
135        let conn = match location {
136            ConnectionType::Persistent(location) => KeystoreDatabaseConnection::open(location, key).await?,
137            ConnectionType::InMemory => KeystoreDatabaseConnection::open_in_memory(key).await?,
138        };
139        let conn = Mutex::new(Some(conn));
140        #[allow(clippy::arc_with_non_send_sync)] // see https://github.com/rustwasm/wasm-bindgen/pull/955
141        let conn = Arc::new(conn);
142        Ok(Self {
143            conn,
144            transaction: Default::default(),
145            transaction_semaphore: Arc::new(Semaphore::new(ALLOWED_CONCURRENT_TRANSACTIONS_COUNT)),
146        })
147    }
148
149    #[cfg(all(test, not(target_os = "unknown")))]
150    pub(crate) async fn open_at_schema_version(
151        name: &str,
152        key: &DatabaseKey,
153        version: MigrationTarget,
154    ) -> CryptoKeystoreResult<Self> {
155        let conn = KeystoreDatabaseConnection::init_with_key_at_schema_version(name, key, version)?;
156        let conn = Mutex::new(Some(conn));
157        let conn = Arc::new(conn);
158        Ok(Self {
159            conn,
160            transaction: Default::default(),
161            transaction_semaphore: Arc::new(Semaphore::new(ALLOWED_CONCURRENT_TRANSACTIONS_COUNT)),
162        })
163    }
164
165    #[cfg(all(test, target_os = "unknown"))]
166    pub(crate) async fn open_at_schema_version(
167        name: &str,
168        key: &DatabaseKey,
169        version: Option<u32>,
170    ) -> CryptoKeystoreResult<Self> {
171        use crate::connection::{
172            storage::{WasmEncryptedStorage, WasmStorageWrapper},
173            wasm::migrations::{TARGET_VERSION, open_at},
174        };
175
176        let version = version.unwrap_or(TARGET_VERSION);
177        let idb_database = open_at(name, key, version).await;
178        let wasm_connection = KeystoreDatabaseConnection::from_inner(WasmEncryptedStorage::new(
179            key,
180            WasmStorageWrapper::Persistent(idb_database),
181        ));
182        let conn = Arc::new(Mutex::new(Some(wasm_connection)));
183        Ok(Self {
184            conn,
185            transaction: Default::default(),
186            transaction_semaphore: Arc::new(Semaphore::new(ALLOWED_CONCURRENT_TRANSACTIONS_COUNT)),
187        })
188    }
189
190    pub async fn location(&self) -> CryptoKeystoreResult<Option<String>> {
191        return Ok(self.conn().await?.location().map(ToString::to_string));
192    }
193
194    /// Get direct exclusive access to the connection.
195    pub async fn conn(&self) -> CryptoKeystoreResult<ConnectionGuard<'_>> {
196        self.conn.lock().await.try_into()
197    }
198
199    /// Wait for any running transaction to finish, then take the connection out of this database,
200    /// preventing this database from being used again.
201    async fn take(&self) -> CryptoKeystoreResult<KeystoreDatabaseConnection> {
202        let _semaphore = self.transaction_semaphore.acquire_arc().await;
203
204        let mut guard = self.conn.lock().await;
205        guard.take().ok_or(CryptoKeystoreError::Closed)
206    }
207
208    // Close this database connection
209    pub async fn close(&self) -> CryptoKeystoreResult<()> {
210        #[cfg(not(target_os = "unknown"))]
211        self.take().await?;
212
213        #[cfg(target_os = "unknown")]
214        {
215            let conn = self.take().await?;
216            conn.close().await?;
217        }
218        Ok(())
219    }
220
221    /// Close this database and delete its contents.
222    pub async fn wipe(&self) -> CryptoKeystoreResult<()> {
223        self.take().await?.wipe().await
224    }
225
226    /// Export a copy of the database to the specified path.
227    /// This creates a fully vacuumed and optimized copy of the database.
228    /// The copy will be encrypted with the same key as the source database.
229    ///
230    /// # Platform Support
231    /// This method is only available on platforms using SQLCipher (not WASM).
232    ///
233    /// # Arguments
234    /// * `destination_path` - The file path where the database copy should be created
235    ///
236    /// # Errors
237    /// Returns an error if:
238    /// - The database is in-memory (cannot export in-memory databases)
239    /// - The destination path is invalid
240    /// - The export operation fails
241    #[cfg(not(target_os = "unknown"))]
242    pub async fn export_copy(&self, destination_path: &str) -> CryptoKeystoreResult<()> {
243        let conn = self.conn().await?;
244        conn.export_copy(destination_path).await
245    }
246
247    pub async fn migrate_db_key_type_to_bytes(
248        name: &str,
249        old_key: &str,
250        new_key: &DatabaseKey,
251    ) -> CryptoKeystoreResult<()> {
252        KeystoreDatabaseConnection::migrate_db_key_type_to_bytes(name, old_key, new_key).await
253    }
254
255    pub async fn update_key(&self, new_key: &DatabaseKey) -> CryptoKeystoreResult<()> {
256        self.conn().await?.update_key(new_key).await
257    }
258
259    /// Waits for the current transaction to be committed or rolled back, then starts a new one.
260    pub async fn new_transaction(&self) -> CryptoKeystoreResult<()> {
261        let semaphore = self.transaction_semaphore.acquire_arc().await;
262        let mut transaction_guard = self.transaction.lock().await;
263        *transaction_guard = Some(KeystoreTransaction::new(semaphore).await?);
264        Ok(())
265    }
266
267    pub async fn commit_transaction(&self) -> CryptoKeystoreResult<()> {
268        let mut transaction_guard = self.transaction.lock().await;
269        let Some(transaction) = transaction_guard.as_ref() else {
270            return Err(CryptoKeystoreError::MutatingOperationWithoutTransaction);
271        };
272        transaction.commit(self).await?;
273        *transaction_guard = None;
274        Ok(())
275    }
276
277    pub async fn rollback_transaction(&self) -> CryptoKeystoreResult<()> {
278        let mut transaction_guard = self.transaction.lock().await;
279        if transaction_guard.is_none() {
280            return Err(CryptoKeystoreError::MutatingOperationWithoutTransaction);
281        };
282        *transaction_guard = None;
283        Ok(())
284    }
285
286    pub async fn child_groups(&self, entity: PersistedMlsGroup) -> CryptoKeystoreResult<Vec<PersistedMlsGroup>> {
287        let mut conn = self.conn().await?;
288        let persisted_records = entity.child_groups(conn.deref_mut()).await?;
289
290        let transaction_guard = self.transaction.lock().await;
291        let Some(transaction) = transaction_guard.as_ref() else {
292            return Ok(persisted_records);
293        };
294        transaction.child_groups(entity, persisted_records).await
295    }
296
297    pub async fn save<'a, E>(&self, entity: E) -> CryptoKeystoreResult<E::AutoGeneratedFields>
298    where
299        E: Entity + EntityDatabaseMutation<'a> + Send + Sync,
300    {
301        let transaction_guard = self.transaction.lock().await;
302        let Some(transaction) = transaction_guard.as_ref() else {
303            return Err(CryptoKeystoreError::MutatingOperationWithoutTransaction);
304        };
305        transaction.save(entity).await
306    }
307
308    pub async fn remove<'a, E>(&self, id: &E::PrimaryKey) -> CryptoKeystoreResult<()>
309    where
310        E: Entity + EntityDatabaseMutation<'a>,
311    {
312        let transaction_guard = self.transaction.lock().await;
313        let Some(transaction) = transaction_guard.as_ref() else {
314            return Err(CryptoKeystoreError::MutatingOperationWithoutTransaction);
315        };
316        transaction.remove::<E>(id).await
317    }
318
319    pub async fn remove_borrowed<'a, E>(&self, id: &E::BorrowedPrimaryKey) -> CryptoKeystoreResult<()>
320    where
321        E: Entity + EntityDatabaseMutation<'a> + BorrowPrimaryKey + EntityDeleteBorrowed<'a>,
322    {
323        let transaction_guard = self.transaction.lock().await;
324        let Some(transaction) = transaction_guard.as_ref() else {
325            return Err(CryptoKeystoreError::MutatingOperationWithoutTransaction);
326        };
327        transaction.remove_borrowed::<E>(id).await
328    }
329
330    pub async fn find_pending_messages_by_conversation_id(
331        &self,
332        conversation_id: &[u8],
333    ) -> CryptoKeystoreResult<Vec<MlsPendingMessage>> {
334        let mut conn = self.conn().await?;
335        let persisted_records = MlsPendingMessage::find_all_matching(&mut conn, &conversation_id.into()).await?;
336
337        let transaction_guard = self.transaction.lock().await;
338        let Some(transaction) = transaction_guard.as_ref() else {
339            return Ok(persisted_records);
340        };
341        transaction
342            .find_pending_messages_by_conversation_id(conversation_id, persisted_records)
343            .await
344    }
345
346    pub async fn remove_pending_messages_by_conversation_id(
347        &self,
348        conversation_id: impl AsRef<[u8]> + Send,
349    ) -> CryptoKeystoreResult<()> {
350        let transaction_guard = self.transaction.lock().await;
351        let Some(transaction) = transaction_guard.as_ref() else {
352            return Err(CryptoKeystoreError::MutatingOperationWithoutTransaction);
353        };
354        transaction
355            .remove_pending_messages_by_conversation_id(conversation_id)
356            .await;
357        Ok(())
358    }
359}
360
361#[cfg_attr(target_os = "unknown", async_trait(?Send))]
362#[cfg_attr(not(target_os = "unknown"), async_trait)]
363impl FetchFromDatabase for Database {
364    async fn get<E>(&self, id: &E::PrimaryKey) -> CryptoKeystoreResult<Option<E>>
365    where
366        E: Entity<ConnectionType = KeystoreDatabaseConnection> + Clone + Send + Sync,
367    {
368        // If a transaction is in progress...
369        if let Some(transaction) = self.transaction.lock().await.as_ref()
370            //... and it has information about this entity, ...
371            && let Some(cached_record) = transaction.get(id).await
372        {
373            return Ok(cached_record.map(Arc::unwrap_or_clone));
374        }
375
376        // Otherwise get it from the database
377        let mut conn = self.conn().await?;
378        E::get(&mut conn, id).await
379    }
380
381    async fn get_borrowed<E>(&self, id: &<E as BorrowPrimaryKey>::BorrowedPrimaryKey) -> CryptoKeystoreResult<Option<E>>
382    where
383        E: EntityGetBorrowed<ConnectionType = KeystoreDatabaseConnection> + Clone + Send + Sync,
384        E::PrimaryKey: Borrow<E::BorrowedPrimaryKey>,
385        for<'a> &'a E::BorrowedPrimaryKey: KeyType,
386    {
387        // If a transaction is in progress...
388        if let Some(transaction) = self.transaction.lock().await.as_ref()
389            //... and it has information about this entity, ...
390            && let Some(cached_record) = transaction.get_borrowed(id).await
391        {
392            return Ok(cached_record.map(Arc::unwrap_or_clone));
393        }
394
395        // Otherwise get it from the database
396        let mut conn = self.conn().await?;
397        E::get_borrowed(&mut conn, id).await
398    }
399
400    async fn count<E>(&self) -> CryptoKeystoreResult<u32>
401    where
402        E: Entity<ConnectionType = KeystoreDatabaseConnection> + Clone + Send + Sync,
403    {
404        if self.transaction.lock().await.is_some() {
405            // Unfortunately, we have to do this because of possible record id overlap
406            // between cache and db.
407            let count = self.load_all::<E>().await?.len();
408            Ok(count as _)
409        } else {
410            let mut conn = self.conn().await?;
411            E::count(&mut conn).await
412        }
413    }
414
415    async fn load_all<E>(&self) -> CryptoKeystoreResult<Vec<E>>
416    where
417        E: Entity<ConnectionType = KeystoreDatabaseConnection> + Clone + Send + Sync,
418    {
419        let mut conn = self.conn().await?;
420        let persisted_records = E::load_all(&mut conn).await?;
421
422        let transaction_guard = self.transaction.lock().await;
423        let Some(transaction) = transaction_guard.as_ref() else {
424            return Ok(persisted_records);
425        };
426        transaction.find_all(persisted_records).await
427    }
428
429    async fn search<E, SearchKey>(&self, search_key: &SearchKey) -> CryptoKeystoreResult<Vec<E>>
430    where
431        E: Entity<ConnectionType = KeystoreDatabaseConnection> + SearchableEntity<SearchKey> + Clone + Send + Sync,
432        SearchKey: KeyType,
433    {
434        let mut conn = self.conn().await?;
435        let persisted_records = E::find_all_matching(&mut conn, search_key).await?;
436
437        let transaction_guard = self.transaction.lock().await;
438        let Some(transaction) = transaction_guard.as_ref() else {
439            return Ok(persisted_records);
440        };
441
442        transaction.search(persisted_records, search_key).await
443    }
444}
core_crypto_keystore/connection/mod.rs

core_crypto_keystore/connection/
mod.rs
