feat: Add a stream to listen for historic room key bundles

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

@@ -8,7 +8,10 @@ use tokio::sync::{broadcast, Mutex};
 use tokio_stream::wrappers::{errors::BroadcastStreamRecvError, BroadcastStream};
 use tracing::{debug, trace, warn};
 
-use super::{caches::SessionStore, DeviceChanges, IdentityChanges, LockableCryptoStore};
+use super::{
+    caches::SessionStore, types::RoomKeyBundleInfo, DeviceChanges, IdentityChanges,
+    LockableCryptoStore,
+};
 use crate::{
     olm::InboundGroupSession,
     store,
@@ -46,6 +49,10 @@ pub(crate) struct CryptoStoreWrapper {
     /// identities which got updated or newly created.
     identities_broadcaster:
         broadcast::Sender<(Option<OwnUserIdentityData>, IdentityChanges, DeviceChanges)>,
+
+    /// The sender side of a broadcast channel which sends out information about
+    /// historic room key bundles we have received.
+    historic_room_key_bundles_broadcaster: broadcast::Sender<RoomKeyBundleInfo>,
 }
 
 impl CryptoStoreWrapper {
@@ -56,6 +63,7 @@ impl CryptoStoreWrapper {
         // 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);
+        let historic_room_key_bundles_broadcaster = broadcast::Sender::new(10);
 
         Self {
             user_id: user_id.to_owned(),
@@ -66,6 +74,7 @@ impl CryptoStoreWrapper {
             room_keys_withheld_received_sender,
             secrets_broadcaster,
             identities_broadcaster,
+            historic_room_key_bundles_broadcaster,
         }
     }
 
@@ -107,6 +116,8 @@ impl CryptoStoreWrapper {
         let secrets = changes.secrets.to_owned();
         let devices = changes.devices.to_owned();
         let identities = changes.identities.to_owned();
+        let room_key_bundle_updates: Vec<_> =
+            changes.received_room_key_bundles.iter().map(RoomKeyBundleInfo::from).collect();
 
         if devices
             .changed
@@ -159,6 +170,10 @@ impl CryptoStoreWrapper {
             let _ = self.secrets_broadcaster.send(secret);
         }
 
+        for bundle_info in room_key_bundle_updates {
+            let _ = self.historic_room_key_bundles_broadcaster.send(bundle_info);
+        }
+
         if !devices.is_empty() || !identities.is_empty() {
             // Mapping the devices and user identities from the read-only variant to one's
             // that contain side-effects requires our own identity. This is
@@ -330,6 +345,13 @@ impl CryptoStoreWrapper {
         Self::filter_errors_out_of_stream(stream, "secrets_stream")
     }
 
+    /// Receive notifications of historic room key bundles being received and
+    /// stored in the store as a [`Stream`].
+    pub fn historic_room_key_stream(&self) -> impl Stream<Item = RoomKeyBundleInfo> {
+        let stream = BroadcastStream::new(self.historic_room_key_bundles_broadcaster.subscribe());
+        Self::filter_errors_out_of_stream(stream, "bundle_stream")
+    }
+
     /// Returns a stream of newly created or updated cryptographic identities.
     ///
     /// This is just a helper method which allows us to build higher level

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

@@ -60,6 +60,7 @@ use thiserror::Error;
 use tokio::sync::{Mutex, Notify, OwnedRwLockWriteGuard, RwLock};
 use tokio_stream::wrappers::errors::BroadcastStreamRecvError;
 use tracing::{error, info, instrument, trace, warn};
+use types::RoomKeyBundleInfo;
 use vodozemac::{megolm::SessionOrdering, Curve25519PublicKey};
 
 use self::types::{
@@ -1313,6 +1314,52 @@ impl Store {
         self.inner.store.secrets_stream()
     }
 
+    /// Receive notifications of historic room key bundles as a [`Stream`].
+    ///
+    /// Historic room key bundles are defined in [MSC4268](https://github.com/matrix-org/matrix-spec-proposals/pull/4268).
+    ///
+    /// Each time a historic room key bundle was received, an update will be
+    /// sent to the stream. This stream can be used to accept historic room key
+    /// bundles that arrive out of order, i.e. the bundle arrives after the
+    /// user has already accepted a room invitation.
+    ///
+    /// # Examples
+    ///
+    /// ```no_run
+    /// # use matrix_sdk_crypto::{
+    /// #    OlmMachine,
+    /// #    store::types::StoredRoomKeyBundleData,
+    /// #    types::room_history::RoomKeyBundle
+    /// # };
+    /// # use ruma::{device_id, user_id};
+    /// # use futures_util::{pin_mut, StreamExt};
+    /// # let alice = user_id!("@alice:example.org").to_owned();
+    /// # async {
+    /// # let machine = OlmMachine::new(&alice, device_id!("DEVICEID")).await;
+    /// let bundle_stream = machine.store().historic_room_key_stream();
+    /// pin_mut!(bundle_stream);
+    ///
+    /// while let Some(bundle_info) = bundle_stream.next().await {
+    ///     // Try to find the bundle content in the store and if it's valid accept it.
+    ///     if let Some(bundle_content) = machine.store().get_received_room_key_bundle_data(&bundle_info.room_id, &bundle_info.sender).await? {
+    ///         let StoredRoomKeyBundleData { sender_user, sender_data, bundle_data } = bundle_content;
+    ///         // Download the bundle now and import it.
+    ///         let bundle: RoomKeyBundle = todo!("Download the bundle");
+    ///         machine.store().receive_room_key_bundle(
+    ///             &bundle_info.room_id,
+    ///             &sender_user,
+    ///             &sender_data,
+    ///             bundle,
+    ///             |_, _| {},
+    ///         ).await?;
+    ///     }
+    /// }
+    /// # anyhow::Ok(()) };
+    /// ```
+    pub fn historic_room_key_stream(&self) -> impl Stream<Item = RoomKeyBundleInfo> {
+        self.inner.store.historic_room_key_stream()
+    }
+
     /// Import the given room keys into the store.
     ///
     /// # Arguments

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

@@ -478,3 +478,26 @@ pub struct RoomKeyWithheldInfo {
     /// withheld.
     pub withheld_event: RoomKeyWithheldEvent,
 }
+
+/// Information about a received historic room key bundle.
+///
+/// This struct contains information needed to uniquely identify a room key
+/// bundle. Only a single bundle per sender for a given room is persisted at a
+/// time.
+///
+/// It is used to notify listeners about received room key bundles.
+#[derive(Debug, Clone)]
+pub struct RoomKeyBundleInfo {
+    /// The user ID of the person that sent us the historic room key bundle.
+    pub sender: OwnedUserId,
+
+    /// The ID of the room the bundle should be used in.
+    pub room_id: OwnedRoomId,
+}
+
+impl From<&StoredRoomKeyBundleData> for RoomKeyBundleInfo {
+    fn from(value: &StoredRoomKeyBundleData) -> Self {
+        let StoredRoomKeyBundleData { sender_user, sender_data: _, bundle_data } = value;
+        Self { sender: sender_user.clone(), room_id: bundle_data.room_id.clone() }
+    }
+}

crates/matrix-sdk/src/encryption/mod.rs

@@ -33,7 +33,7 @@ use futures_util::{
     stream::{self, StreamExt},
 };
 use matrix_sdk_base::crypto::{
-    store::types::RoomKeyInfo,
+    store::types::{RoomKeyBundleInfo, RoomKeyInfo},
     types::requests::{
         OutgoingRequest, OutgoingVerificationRequest, RoomMessageRequest, ToDeviceRequest,
     },
@@ -1478,6 +1478,43 @@ impl Encryption {
         Some(olm.store().room_keys_received_stream())
     }
 
+    /// Receive notifications of historic room key bundles as a [`Stream`].
+    ///
+    /// Historic room key bundles are defined in [MSC4268](https://github.com/matrix-org/matrix-spec-proposals/pull/4268).
+    ///
+    /// Each time a historic room key bundle was received, an update will be
+    /// sent to the stream. This stream is useful for informative purposes
+    /// exclusively, historic room key bundles are handled by the SDK
+    /// automatically.
+    ///
+    /// # Examples
+    ///
+    /// ```no_run
+    /// # use matrix_sdk::Client;
+    /// # use url::Url;
+    /// # async {
+    /// # let homeserver = Url::parse("http://example.com")?;
+    /// # let client = Client::new(homeserver).await?;
+    /// use futures_util::StreamExt;
+    ///
+    /// let Some(mut bundle_stream) =
+    ///     client.encryption().historic_room_key_stream().await
+    /// else {
+    ///     return Ok(());
+    /// };
+    ///
+    /// while let Some(bundle_info) = bundle_stream.next().await {
+    ///     println!("Received a historic room key bundle {bundle_info:?}");
+    /// }
+    /// # anyhow::Ok(()) };
+    /// ```
+    pub async fn historic_room_key_stream(&self) -> Option<impl Stream<Item = RoomKeyBundleInfo>> {
+        let olm = self.client.olm_machine().await;
+        let olm = olm.as_ref()?;
+
+        Some(olm.store().historic_room_key_stream())
+    }
+
     /// Get the secret storage manager of the client.
     pub fn secret_storage(&self) -> SecretStorage {
         SecretStorage { client: self.client.to_owned() }

crates/matrix-sdk/src/room/shared_room_history.rs

@@ -129,6 +129,8 @@ pub(super) async fn maybe_accept_key_bundle(room: &Room, inviter: &UserId) -> Re
     else {
         // No bundle received (yet).
         // TODO: deal with the bundle arriving later (https://github.com/matrix-org/matrix-rust-sdk/issues/4926)
+        // We need to check for all them bundles in the store when we create the client
+        // object and we need to process them when they arrive.
         return Ok(());
     };
 

testing/matrix-sdk-integration-testing/src/tests/e2ee/shared_history.rs

@@ -1,6 +1,9 @@
+use std::ops::Deref;
+
 use anyhow::Result;
 use assert_matches2::assert_let;
 use assign::assign;
+use futures::{pin_mut, FutureExt, StreamExt};
 use matrix_sdk::{
     assert_decrypted_message_eq,
     encryption::EncryptionSettings,
@@ -68,6 +71,12 @@ async fn test_history_share_on_invite() -> Result<()> {
         .expect("We should be able to send a message to the room")
         .event_id;
 
+    let bundle_stream = bob
+        .encryption()
+        .historic_room_key_stream()
+        .await
+        .expect("We should be able to get the bundle stream");
+
     // Alice invites Bob to the room
     alice_room.invite_user_by_id(bob.user_id().unwrap()).await?;
 
@@ -88,6 +97,17 @@ async fn test_history_share_on_invite() -> Result<()> {
 
     let bob_room = bob.get_room(alice_room.room_id()).expect("Bob should have received the invite");
 
+    pin_mut!(bundle_stream);
+
+    let info = bundle_stream
+        .next()
+        .now_or_never()
+        .flatten()
+        .expect("We should be notified about the received bundle");
+
+    assert_eq!(Some(info.sender.deref()), alice.user_id());
+    assert_eq!(info.room_id, alice_room.room_id());
+
     bob_room
         .join()
         .instrument(bob_span.clone())