crypto: Wire up save_inbound_group_sessions to room_keys_received stream

https://github.com/matrix-org/matrix-rust-sdk/pull/3448 added a new method
`save_inbound_group_sessions` to `CrytoStore`, but forgot to wire it up to the
`room_keys_received` stream.

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

@@ -10,6 +10,7 @@ use tracing::warn;
 
 use super::{DeviceChanges, IdentityChanges, LockableCryptoStore};
 use crate::{
+    olm::InboundGroupSession,
     store,
     store::{Changes, DynCryptoStore, IntoCryptoStore, RoomKeyInfo},
     GossippedSecret, ReadOnlyOwnUserIdentity,
@@ -95,6 +96,31 @@ impl CryptoStoreWrapper {
         Ok(())
     }
 
+    /// Save a list of inbound group sessions to the store.
+    ///
+    /// # Arguments
+    ///
+    /// * `sessions` - The sessions to be saved.
+    /// * `backed_up_to_version` - If the keys should be marked as having been
+    ///   backed up, the version of the backup.
+    ///
+    /// Note: some implementations ignore `backup_version` and assume the
+    /// current backup version, which is normally the same.
+    pub async fn save_inbound_group_sessions(
+        &self,
+        sessions: Vec<InboundGroupSession>,
+        backed_up_to_version: Option<&str>,
+    ) -> store::Result<()> {
+        let room_key_updates: Vec<_> = sessions.iter().map(RoomKeyInfo::from).collect();
+        self.store.save_inbound_group_sessions(sessions, backed_up_to_version).await?;
+
+        if !room_key_updates.is_empty() {
+            // Ignore the result. It can only fail if there are no listeners.
+            let _ = self.room_keys_received_sender.send(room_key_updates);
+        }
+        Ok(())
+    }
+
     /// Receive notifications of room keys being received as a [`Stream`].
     ///
     /// Each time a room key is updated in any way, an update will be sent to

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

@@ -1869,6 +1869,29 @@ mod tests {
 
     use crate::{machine::tests::get_machine_pair, types::EventEncryptionAlgorithm};
 
+    #[async_test]
+    async fn import_room_keys_notifies_stream() {
+        use futures_util::FutureExt;
+
+        let (alice, bob, _) =
+            get_machine_pair(user_id!("@a:s.co"), user_id!("@b:s.co"), false).await;
+
+        let room1_id = room_id!("!room1:localhost");
+        alice.create_outbound_group_session_with_defaults_test_helper(room1_id).await.unwrap();
+        let exported_sessions = alice.store().export_room_keys(|_| true).await.unwrap();
+
+        let mut room_keys_received_stream = Box::pin(bob.store().room_keys_received_stream());
+        bob.store().import_room_keys(exported_sessions, None, |_, _| {}).await.unwrap();
+
+        let room_keys = room_keys_received_stream
+            .next()
+            .now_or_never()
+            .flatten()
+            .expect("We should have received an update of room key infos");
+        assert_eq!(room_keys.len(), 1);
+        assert_eq!(room_keys[0].room_id, "!room1:localhost");
+    }
+
     #[async_test]
     async fn export_room_keys_provides_selected_keys() {
         // Given an OlmMachine with room keys in it