Implement the redecryption logic in the event cache

crates/matrix-sdk-ui/src/timeline/tests/encryption.rs

@@ -424,6 +424,17 @@ async fn test_retry_edit_decryption() {
     assert!(msg.is_edited());
     assert_eq!(msg.body(), "This is Error");
 
+    let item =
+        assert_next_matches_with_timeout!(stream, VectorDiff::Set { index: 0, value } => value);
+
+    // TODO: We receive this update twice, since the event cache decrypts things as
+    // well as the timeline.
+    assert_matches!(item.encryption_info(), Some(_));
+    assert_matches!(item.latest_edit_json(), Some(_));
+    assert_let!(Some(msg) = item.content().as_message());
+    assert!(msg.is_edited());
+    assert_eq!(msg.body(), "This is Error");
+
     // (There are no more items)
     assert_pending!(stream);
 }

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

@@ -74,9 +74,13 @@ use crate::{
 
 mod deduplicator;
 mod pagination;
+#[cfg(feature = "e2e-encryption")]
+mod redecryptor;
 mod room;
 
 pub use pagination::{RoomPagination, RoomPaginationStatus};
+#[cfg(feature = "e2e-encryption")]
+pub use redecryptor::{DecryptionRetryRequest, RedecryptorReport};
 pub use room::{RoomEventCache, RoomEventCacheSubscriber, ThreadEventCacheUpdate};
 
 /// An error observed in the [`EventCache`].
@@ -148,6 +152,10 @@ pub struct EventCacheDropHandles {
 
     /// The task used to automatically shrink the linked chunks.
     auto_shrink_linked_chunk_task: JoinHandle<()>,
+
+    /// The task used to automatically redecrypt UTDs.
+    #[cfg(feature = "e2e-encryption")]
+    _redecryptor: redecryptor::Redecryptor,
 }
 
 impl fmt::Debug for EventCacheDropHandles {
@@ -200,6 +208,9 @@ impl EventCache {
             linked_chunk_update_sender.clone(),
         )));
 
+        #[cfg(feature = "e2e-encryption")]
+        let redecryption_channels = redecryptor::RedecryptorChannels::new();
+
         Self {
             inner: Arc::new(EventCacheInner {
                 client,
@@ -213,6 +224,8 @@ impl EventCache {
                 _thread_subscriber_task: thread_subscriber_task,
                 #[cfg(feature = "experimental-search")]
                 _search_indexing_task: search_indexing_task,
+                #[cfg(feature = "e2e-encryption")]
+                redecryption_channels,
                 thread_subscriber_receiver,
             }),
         }
@@ -257,10 +270,26 @@ impl EventCache {
                 auto_shrink_receiver,
             ));
 
+            #[cfg(feature = "e2e-encryption")]
+            let redecryptor = {
+                let receiver = self
+                    .inner
+                    .redecryption_channels
+                    .decryption_request_receiver
+                    .lock()
+                    .take()
+                    .expect("We should have initialized the channel an subscribing should happen only once");
+
+                redecryptor::Redecryptor::new(Arc::downgrade(&self.inner), receiver)
+            };
+
+
             Arc::new(EventCacheDropHandles {
                 listen_updates_task,
                 ignore_user_list_update_task,
                 auto_shrink_linked_chunk_task,
+                #[cfg(feature = "e2e-encryption")]
+                _redecryptor: redecryptor,
             })
         });
 
@@ -840,6 +869,9 @@ struct EventCacheInner {
     /// This is helpful for tests to coordinate that a new thread subscription
     /// has been sent or not.
     thread_subscriber_receiver: Receiver<()>,
+
+    #[cfg(feature = "e2e-encryption")]
+    redecryption_channels: redecryptor::RedecryptorChannels,
 }
 
 type AutoShrinkChannelPayload = OwnedRoomId;