feature: Crypto store wrapper migration framework

crates/matrix-sdk-crypto/src/dehydrated_devices.rs

@@ -96,7 +96,7 @@ impl DehydratedDevices {
         let user_identity = self.inner.store().private_identity();
 
         let account = Account::new_dehydrated(user_id);
-        let store = Arc::new(CryptoStoreWrapper::new(user_id, MemoryStore::new()));
+        let store = Arc::new(CryptoStoreWrapper::new(user_id, MemoryStore::new()).await.unwrap());
 
         let verification_machine = VerificationMachine::new(
             account.static_data().clone(),

crates/matrix-sdk-crypto/src/gossiping/machine.rs

@@ -1177,7 +1177,7 @@ mod tests {
         let device_id = DeviceId::new();
 
         let account = Account::with_device_id(&user_id, &device_id);
-        let store = Arc::new(CryptoStoreWrapper::new(&user_id, MemoryStore::new()));
+        let store = Arc::new(CryptoStoreWrapper::new(&user_id, MemoryStore::new()).await.unwrap());
         let identity = Arc::new(Mutex::new(PrivateCrossSigningIdentity::empty(alice_id())));
         let verification =
             VerificationMachine::new(account.static_data.clone(), identity.clone(), store.clone());
@@ -1197,7 +1197,7 @@ mod tests {
         let another_device =
             DeviceData::from_account(&Account::with_device_id(&user_id, alice2_device_id()));
 
-        let store = Arc::new(CryptoStoreWrapper::new(&user_id, MemoryStore::new()));
+        let store = Arc::new(CryptoStoreWrapper::new(&user_id, MemoryStore::new()).await.unwrap());
         let identity = Arc::new(Mutex::new(PrivateCrossSigningIdentity::empty(alice_id())));
         let verification =
             VerificationMachine::new(account.static_data.clone(), identity.clone(), store.clone());

crates/matrix-sdk-crypto/src/identities/manager.rs

@@ -1067,7 +1067,7 @@ pub(crate) mod testing {
         let user_id = user_id.to_owned();
         let account = Account::with_device_id(&user_id, device_id);
         let static_account = account.static_data().clone();
-        let store = Arc::new(CryptoStoreWrapper::new(&user_id, MemoryStore::new()));
+        let store = Arc::new(CryptoStoreWrapper::new(&user_id, MemoryStore::new()).await.unwrap());
         let verification =
             VerificationMachine::new(static_account.clone(), identity.clone(), store.clone());
         let store = Store::new(static_account, identity, store, verification);

crates/matrix-sdk-crypto/src/identities/user.rs

@@ -1084,8 +1084,8 @@ pub(crate) mod tests {
         assert_eq!("1", with_serializer.version.unwrap());
     }
 
-    #[test]
-    fn own_identity_check_signatures() {
+    #[async_test]
+    async fn own_identity_check_signatures() {
         let response = own_key_query();
         let identity = get_own_identity();
         let (first, second) = device(&response);
@@ -1098,7 +1098,7 @@ pub(crate) mod tests {
         let verification_machine = VerificationMachine::new(
             Account::with_device_id(second.user_id(), second.device_id()).static_data,
             private_identity,
-            Arc::new(CryptoStoreWrapper::new(second.user_id(), MemoryStore::new())),
+            Arc::new(CryptoStoreWrapper::new(second.user_id(), MemoryStore::new()).await.unwrap()),
         );
 
         let first = Device {
@@ -1139,7 +1139,7 @@ pub(crate) mod tests {
         let verification_machine = VerificationMachine::new(
             Account::with_device_id(device.user_id(), device.device_id()).static_data,
             id.clone(),
-            Arc::new(CryptoStoreWrapper::new(device.user_id(), MemoryStore::new())),
+            Arc::new(CryptoStoreWrapper::new(device.user_id(), MemoryStore::new()).await.unwrap()),
         );
 
         let public_identity = identity.to_public_identity().await.unwrap();

crates/matrix-sdk-crypto/src/machine.rs

@@ -173,7 +173,8 @@ impl OlmMachine {
         let account = Account::rehydrate(pickle_key, self.user_id(), device_id, device_data)?;
         let static_account = account.static_data().clone();
 
-        let store = Arc::new(CryptoStoreWrapper::new(self.user_id(), MemoryStore::new()));
+        let store =
+            Arc::new(CryptoStoreWrapper::new(self.user_id(), MemoryStore::new()).await.unwrap());
         let device = DeviceData::from_account(&account);
         store.save_pending_changes(PendingChanges { account: Some(account) }).await?;
         store
@@ -356,7 +357,7 @@ impl OlmMachine {
         });
 
         let identity = Arc::new(Mutex::new(identity));
-        let store = Arc::new(CryptoStoreWrapper::new(user_id, store));
+        let store = Arc::new(CryptoStoreWrapper::new(user_id, store).await?);
         Ok(OlmMachine::new_helper(device_id, store, static_account, identity, maybe_backup_key))
     }
 

crates/matrix-sdk-crypto/src/olm/group_sessions/sender_data_finder.rs

@@ -860,10 +860,10 @@ mod tests {
             let sender_device = if options.device_is_signed {
                 create_signed_device(&sender.account, &*sender.private_identity.lock().await).await
             } else {
-                create_unsigned_device(&sender.account)
+                create_unsigned_device(&sender.account).await
             };
 
-            let store = create_store(&me);
+            let store = create_store(&me).await;
 
             save_to_store(&store, &me, &sender, &sender_device, &options).await;
 
@@ -912,8 +912,9 @@ mod tests {
         }
     }
 
-    fn create_store(me: &TestUser) -> Store {
-        let store_wrapper = Arc::new(CryptoStoreWrapper::new(&me.user_id, MemoryStore::new()));
+    async fn create_store(me: &TestUser) -> Store {
+        let store_wrapper =
+            Arc::new(CryptoStoreWrapper::new(&me.user_id, MemoryStore::new()).await.unwrap());
 
         let verification_machine = VerificationMachine::new(
             me.account.deref().clone(),
@@ -1063,22 +1064,24 @@ mod tests {
         self_signing.sign_device(&mut device_keys).unwrap();
         read_only_device.update_device(&device_keys).unwrap();
 
-        wrap_device(account, read_only_device)
+        wrap_device(account, read_only_device).await
     }
 
-    fn create_unsigned_device(account: &Account) -> Device {
-        wrap_device(account, DeviceData::from_account(account))
+    async fn create_unsigned_device(account: &Account) -> Device {
+        wrap_device(account, DeviceData::from_account(account)).await
     }
 
-    fn wrap_device(account: &Account, read_only_device: DeviceData) -> Device {
+    async fn wrap_device(account: &Account, read_only_device: DeviceData) -> Device {
         Device {
             inner: read_only_device,
             verification_machine: VerificationMachine::new(
                 account.deref().clone(),
                 Arc::new(Mutex::new(PrivateCrossSigningIdentity::new(
                     account.user_id().to_owned(),
                 ))),
-                Arc::new(CryptoStoreWrapper::new(account.user_id(), MemoryStore::new())),
+                Arc::new(
+                    CryptoStoreWrapper::new(account.user_id(), MemoryStore::new()).await.unwrap(),
+                ),
             ),
             own_identity: None,
             device_owner_identity: None,

crates/matrix-sdk-crypto/src/session_manager/sessions.rs

@@ -680,7 +680,7 @@ mod tests {
         let device_id = device_id();
 
         let account = Account::with_device_id(user_id, device_id);
-        let store = Arc::new(CryptoStoreWrapper::new(user_id, MemoryStore::new()));
+        let store = Arc::new(CryptoStoreWrapper::new(user_id, MemoryStore::new()).await.unwrap());
         let identity = Arc::new(Mutex::new(PrivateCrossSigningIdentity::empty(user_id)));
         let verification = VerificationMachine::new(
             account.static_data().clone(),

crates/matrix-sdk-crypto/src/store/crypto_store_wrapper.rs

@@ -13,6 +13,8 @@ use crate::{
     olm::InboundGroupSession,
     store,
     store::{Changes, DynCryptoStore, IntoCryptoStore, RoomKeyInfo, RoomKeyWithheldInfo},
+    CryptoStoreError,
+    CryptoStoreError::CryptoStoreWrapperMigrationError,
     GossippedSecret, OwnUserIdentityData,
 };
 
@@ -43,23 +45,81 @@ pub(crate) struct CryptoStoreWrapper {
         broadcast::Sender<(Option<OwnUserIdentityData>, IdentityChanges, DeviceChanges)>,
 }
 
+const STORE_WRAPPER_VERSION_KEY: &str = "CryptoStoreWrapper_VERSION";
+const STORE_WRAPPER_VERSION: u32 = 1;
+
+async fn migrate(wrapper: &CryptoStoreWrapper, old_version: u32) -> Result<(), CryptoStoreError> {
+    if old_version < 1 {
+        // this
+        wrapper
+            .store
+            .set_custom_value(STORE_WRAPPER_VERSION_KEY, 1_u32.to_le_bytes().into())
+            .await?;
+    }
+
+    // if old_version < 2 { // Not `else if` because we want to run each migration
+    // in turn     // Future migration
+    //     ... Something like marking all users as dirty?
+    //     // Update the current version to complete this migration
+    //     wrapper.store.set_custom_value(STORE_WRAPPER_VERSION_KEY,
+    // 2_u32.to_le_bytes().into()).await?; }
+
+    Ok(())
+}
+
+async fn read_store_version(store: &DynCryptoStore) -> Result<Option<u32>, CryptoStoreError> {
+    let value = store.get_custom_value(STORE_WRAPPER_VERSION_KEY).await?;
+    Ok(match value {
+        Some(u8_vec) => Some(u32::from_le_bytes(u8_vec.try_into().map_err(|_| {
+            CryptoStoreWrapperMigrationError("Failed to read store version".to_owned())
+        })?)),
+        _ => None,
+    })
+}
 impl CryptoStoreWrapper {
-    pub(crate) fn new(user_id: &UserId, store: impl IntoCryptoStore) -> Self {
+    pub(crate) async fn new(
+        user_id: &UserId,
+        store: impl IntoCryptoStore,
+    ) -> Result<Self, CryptoStoreError> {
         let room_keys_received_sender = broadcast::Sender::new(10);
         let room_keys_withheld_received_sender = broadcast::Sender::new(10);
         let secrets_broadcaster = broadcast::Sender::new(10);
         // The identities broadcaster is responsible for user identities as well as
         // devices, that's why we increase the capacity here.
         let identities_broadcaster = broadcast::Sender::new(20);
 
-        Self {
+        let store_wrapper = Self {
             user_id: user_id.to_owned(),
             store: store.into_crypto_store(),
             room_keys_received_sender,
             room_keys_withheld_received_sender,
             secrets_broadcaster,
             identities_broadcaster,
+        };
+
+        // Simple wrapper level migration handling
+        let old_version = read_store_version(store_wrapper.store.as_ref()).await?.unwrap_or(0);
+        let new_version = STORE_WRAPPER_VERSION;
+        if new_version < old_version {
+            // Backward migration
+            return Err(CryptoStoreWrapperMigrationError(
+                "The database format changed in an incompatible way".into(),
+            ));
         }
+
+        migrate(&store_wrapper, old_version).await.map_err(|e| {
+            CryptoStoreWrapperMigrationError(format!("An Error occurred during migration {}", e))
+        })?;
+
+        let upgraded_version = read_store_version(store_wrapper.store.as_ref()).await?.unwrap_or(0);
+
+        if upgraded_version != new_version {
+            return Err(CryptoStoreWrapperMigrationError(
+                "Migration did not upgrade up to the expected store version".into(),
+            ));
+        }
+
+        Ok(store_wrapper)
     }
 
     /// Save the set of changes to the store.
@@ -232,3 +292,52 @@ impl Deref for CryptoStoreWrapper {
         self.store.deref()
     }
 }
+
+#[cfg(test)]
+pub(crate) mod tests {
+    use assert_matches::assert_matches;
+    use matrix_sdk_test::async_test;
+    use ruma::user_id;
+
+    use crate::{
+        store::{
+            crypto_store_wrapper::{
+                read_store_version, STORE_WRAPPER_VERSION, STORE_WRAPPER_VERSION_KEY,
+            },
+            CryptoStore, CryptoStoreWrapper, MemoryStore,
+        },
+        CryptoStoreError,
+    };
+
+    #[async_test]
+    async fn test_migration() {
+        let store = MemoryStore::new();
+
+        let version = store.get_custom_value(STORE_WRAPPER_VERSION_KEY).await.unwrap();
+
+        assert!(version.is_none());
+
+        let alice_id = user_id!("@alice:localhost");
+        let wrapper = CryptoStoreWrapper::new(alice_id, store).await.unwrap();
+
+        let version = read_store_version(wrapper.store.as_ref()).await.unwrap().unwrap();
+
+        assert_eq!(STORE_WRAPPER_VERSION, version);
+    }
+
+    #[async_test]
+    async fn test_backward_migration_should_error() {
+        let store = MemoryStore::new();
+
+        store
+            .set_custom_value(STORE_WRAPPER_VERSION_KEY, 4_u32.to_le_bytes().into())
+            .await
+            .unwrap();
+
+        let alice_id = user_id!("@alice:localhost");
+        let wrapper = CryptoStoreWrapper::new(alice_id, store).await;
+        assert!(wrapper.is_err());
+        let error = wrapper.unwrap_err();
+        assert_matches!(error, CryptoStoreError::CryptoStoreWrapperMigrationError(_));
+    }
+}

crates/matrix-sdk-crypto/src/store/error.rs

@@ -82,6 +82,10 @@ pub enum CryptoStoreError {
     /// An error due to an invalid generation in a cross-process locking scheme.
     #[error("invalid lock generation: {0}")]
     InvalidLockGeneration(String),
+
+    /// A problem when migrating the crypto store wrapper
+    #[error("Crypto store wrapper error: {0}")]
+    CryptoStoreWrapperMigrationError(String),
 }
 
 impl CryptoStoreError {

crates/matrix-sdk-crypto/src/verification/machine.rs

@@ -578,7 +578,7 @@ mod tests {
         let _ = VerificationMachine::new(
             alice.static_data,
             identity,
-            Arc::new(CryptoStoreWrapper::new(alice_id(), MemoryStore::new())),
+            Arc::new(CryptoStoreWrapper::new(alice_id(), MemoryStore::new()).await.unwrap()),
         );
     }
 

crates/matrix-sdk-crypto/src/verification/mod.rs

@@ -868,14 +868,14 @@ pub(crate) mod tests {
         bob_store.save_devices(vec![alice_device]);
 
         let alice_store = VerificationStore {
-            inner: Arc::new(CryptoStoreWrapper::new(alice.user_id(), alice_store)),
+            inner: Arc::new(CryptoStoreWrapper::new(alice.user_id(), alice_store).await.unwrap()),
             account: alice.static_data.clone(),
             private_identity: alice_private_identity.into(),
         };
 
         let bob_store = VerificationStore {
             account: bob.static_data.clone(),
-            inner: Arc::new(CryptoStoreWrapper::new(bob.user_id(), bob_store)),
+            inner: Arc::new(CryptoStoreWrapper::new(bob.user_id(), bob_store).await.unwrap()),
             private_identity: bob_private_identity.into(),
         };
 

crates/matrix-sdk-crypto/src/verification/sas/mod.rs

@@ -902,8 +902,8 @@ mod tests {
         device_id!("BOBDEVCIE")
     }
 
-    fn machine_pair_test_helper() -> (VerificationStore, DeviceData, VerificationStore, DeviceData)
-    {
+    async fn machine_pair_test_helper(
+    ) -> (VerificationStore, DeviceData, VerificationStore, DeviceData) {
         let alice = Account::with_device_id(alice_id(), alice_device_id());
         let alice_device = DeviceData::from_account(&alice);
 
@@ -912,7 +912,9 @@ mod tests {
 
         let alice_store = VerificationStore {
             account: alice.static_data.clone(),
-            inner: Arc::new(CryptoStoreWrapper::new(alice.user_id(), MemoryStore::new())),
+            inner: Arc::new(
+                CryptoStoreWrapper::new(alice.user_id(), MemoryStore::new()).await.unwrap(),
+            ),
             private_identity: Mutex::new(PrivateCrossSigningIdentity::empty(alice_id())).into(),
         };
 
@@ -921,7 +923,7 @@ mod tests {
 
         let bob_store = VerificationStore {
             account: bob.static_data.clone(),
-            inner: Arc::new(CryptoStoreWrapper::new(bob.user_id(), bob_store)),
+            inner: Arc::new(CryptoStoreWrapper::new(bob.user_id(), bob_store).await.unwrap()),
             private_identity: Mutex::new(PrivateCrossSigningIdentity::empty(bob_id())).into(),
         };
 
@@ -930,7 +932,7 @@ mod tests {
 
     #[async_test]
     async fn sas_wrapper_full() {
-        let (alice_store, alice_device, bob_store, bob_device) = machine_pair_test_helper();
+        let (alice_store, alice_device, bob_store, bob_device) = machine_pair_test_helper().await;
 
         let identities = alice_store.get_identities(bob_device).await.unwrap();
 
@@ -1004,7 +1006,7 @@ mod tests {
 
     #[async_test]
     async fn sas_with_restricted_methods() {
-        let (alice_store, alice_device, bob_store, bob_device) = machine_pair_test_helper();
+        let (alice_store, alice_device, bob_store, bob_device) = machine_pair_test_helper().await;
         let identities = alice_store.get_identities(bob_device).await.unwrap();
 
         let short_auth_strings = vec![ShortAuthenticationString::Decimal];