feat(sdk): Create ThreadEventCacheUpdateSender.

This patch creates the `ThreadEventCacheUpdateSender` à la
`RoomEventCacheUpdateSender` to abstract the sender.

This patch uses `ThreadEventCacheUpdateSender` in
`ThreadEventCacheInner` (new) along with `ThreadEventCacheState`. The
use in the `ThreadEventCacheInner` type is necessary to be able to send
updates without reaching the state (which is behind an async faillible
lock). This is necessary for pagination, see next commit.

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

@@ -16,6 +16,7 @@
 
 pub mod pagination;
 mod state;
+mod updates;
 
 use std::{fmt, sync::Arc};
 
@@ -25,7 +26,7 @@ pub(super) use state::LockedThreadEventCacheState;
 use tokio::sync::broadcast::{Receiver, Sender};
 use tracing::{error, trace};
 
-use self::pagination::ThreadPagination;
+use self::{pagination::ThreadPagination, updates::ThreadEventCacheUpdateSender};
 use super::{
     super::Result, EventsOrigin, TimelineVectorDiffs, room::RoomEventCacheLinkedChunkUpdate,
 };
@@ -46,6 +47,9 @@ struct ThreadEventCacheInner {
 
     /// State for this thread's event cache.
     state: LockedThreadEventCacheState,
+
+    /// Update sender for this room.
+    update_sender: ThreadEventCacheUpdateSender,
 }
 
 impl fmt::Debug for ThreadEventCache {
@@ -64,6 +68,8 @@ impl ThreadEventCache {
         store: EventCacheStoreLock,
         linked_chunk_update_sender: Sender<RoomEventCacheLinkedChunkUpdate>,
     ) -> Result<Self> {
+        let update_sender = ThreadEventCacheUpdateSender::new();
+
         Ok(Self {
             inner: Arc::new(ThreadEventCacheInner {
                 thread_id: thread_id.clone(),
@@ -73,9 +79,11 @@ impl ThreadEventCache {
                     thread_id,
                     own_user_id,
                     store,
+                    update_sender.clone(),
                     linked_chunk_update_sender,
                 )
                 .await?,
+                update_sender,
             }),
         })
     }
@@ -87,7 +95,7 @@ impl ThreadEventCache {
         let events =
             state.thread_linked_chunk().events().map(|(_position, item)| item.clone()).collect();
 
-        let recv = state.state.sender.subscribe();
+        let recv = self.inner.update_sender.new_thread_receiver();
 
         Ok((events, recv))
     }
@@ -100,12 +108,10 @@ impl ThreadEventCache {
 
     /// Clear a thread, after a gappy sync for instance.
     pub async fn clear(&mut self) -> Result<()> {
-        let mut state = self.inner.state.write().await?;
-
-        let updates_as_vector_diffs = state.reset().await?;
+        let updates_as_vector_diffs = self.inner.state.write().await?.reset().await?;
 
         if !updates_as_vector_diffs.is_empty() {
-            let _ = state.state.sender.send(TimelineVectorDiffs {
+            let _ = self.inner.update_sender.send(TimelineVectorDiffs {
                 diffs: updates_as_vector_diffs,
                 origin: EventsOrigin::Cache,
             });
@@ -127,7 +133,7 @@ impl ThreadEventCache {
         let timeline_event_diffs = state.handle_sync(events).await?;
 
         if !timeline_event_diffs.is_empty() {
-            let _ = state.state.sender.send(TimelineVectorDiffs {
+            let _ = self.inner.update_sender.send(TimelineVectorDiffs {
                 diffs: timeline_event_diffs,
                 origin: EventsOrigin::Sync,
             });
@@ -158,7 +164,7 @@ impl ThreadEventCache {
         let timeline_event_diffs = state.thread_linked_chunk_mut().updates_as_vector_diffs();
 
         if !timeline_event_diffs.is_empty() {
-            let _ = state.state.sender.send(TimelineVectorDiffs {
+            let _ = self.inner.update_sender.send(TimelineVectorDiffs {
                 diffs: timeline_event_diffs,
                 origin: EventsOrigin::Sync,
             });

crates/matrix-sdk/src/event_cache/caches/thread/pagination.rs

@@ -276,9 +276,9 @@ impl PaginatedCache for ThreadEventCacheWrapper {
 
         if !updates.is_empty() {
             // Send the updates to the listeners.
-            let _ = state
-                .state
-                .sender
+            let _ = self
+                .cache
+                .update_sender
                 .send(TimelineVectorDiffs { diffs: updates, origin: EventsOrigin::Pagination });
         }
 

crates/matrix-sdk/src/event_cache/caches/thread/state.rs

@@ -27,16 +27,19 @@ use ruma::{OwnedEventId, OwnedRoomId, OwnedUserId};
 use tokio::sync::broadcast::Sender;
 use tracing::{debug, error, instrument};
 
-use super::super::{
+use super::{
     super::{
-        EventCacheError, EventsOrigin, Result,
-        deduplicator::{DeduplicationOutcome, filter_duplicate_events},
-        persistence::{load_linked_chunk_metadata, send_updates_to_store},
+        super::{
+            EventCacheError, EventsOrigin, Result,
+            deduplicator::{DeduplicationOutcome, filter_duplicate_events},
+            persistence::{load_linked_chunk_metadata, send_updates_to_store},
+        },
+        TimelineVectorDiffs,
+        event_linked_chunk::{EventLinkedChunk, sort_positions_descending},
+        lock,
+        room::RoomEventCacheLinkedChunkUpdate,
     },
-    TimelineVectorDiffs,
-    event_linked_chunk::{EventLinkedChunk, sort_positions_descending},
-    lock,
-    room::RoomEventCacheLinkedChunkUpdate,
+    ThreadEventCacheUpdateSender,
 };
 
 pub struct ThreadEventCacheState {
@@ -56,8 +59,11 @@ pub struct ThreadEventCacheState {
     /// The linked chunk for this thread.
     thread_linked_chunk: EventLinkedChunk,
 
-    /// A sender for live events updates in this thread.
-    pub sender: Sender<TimelineVectorDiffs>,
+    /// A clone of [`super::ThreadEventCacheInner::update_sender`].
+    ///
+    /// This is used only by the [`ThreadEventCacheStateLock::read`] and
+    /// [`ThreadEventCacheStateLock::write`] when the state must be reset.
+    update_sender: ThreadEventCacheUpdateSender,
 
     /// A sender for the globally observable linked chunk updates that happened
     /// during a sync or a back-pagination.
@@ -100,6 +106,7 @@ impl LockedThreadEventCacheState {
         thread_id: OwnedEventId,
         own_user_id: OwnedUserId,
         store: EventCacheStoreLock,
+        update_sender: ThreadEventCacheUpdateSender,
         linked_chunk_update_sender: Sender<RoomEventCacheLinkedChunkUpdate>,
     ) -> Result<Self> {
         let store_guard = match store.lock().await? {
@@ -167,7 +174,7 @@ impl LockedThreadEventCacheState {
                 linked_chunk,
                 full_linked_chunk_metadata,
             ),
-            sender: Sender::new(32),
+            update_sender,
             linked_chunk_update_sender,
             waited_for_initial_prev_token: false,
         }))
@@ -194,8 +201,10 @@ impl<'a> lock::Reload for ThreadEventCacheStateLockWriteGuard<'a> {
         let diffs = self.state.thread_linked_chunk.updates_as_vector_diffs();
 
         if !diffs.is_empty() {
-            let _ =
-                self.state.sender.send(TimelineVectorDiffs { diffs, origin: EventsOrigin::Cache });
+            let _ = self
+                .state
+                .update_sender
+                .send(TimelineVectorDiffs { diffs, origin: EventsOrigin::Cache });
         }
 
         Ok(())

crates/matrix-sdk/src/event_cache/caches/thread/updates.rs

@@ -0,0 +1,40 @@
+// Copyright 2026 The Matrix.org Foundation C.I.C.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use tokio::sync::broadcast::{Receiver, Sender};
+
+use crate::event_cache::TimelineVectorDiffs;
+
+/// A small type to send updates in all channels.
+#[derive(Clone)]
+pub struct ThreadEventCacheUpdateSender {
+    thread_sender: Sender<TimelineVectorDiffs>,
+}
+
+impl ThreadEventCacheUpdateSender {
+    /// Create a new [`ThreadEventCacheUpdateSender`].
+    pub fn new() -> Self {
+        Self { thread_sender: Sender::new(32) }
+    }
+
+    /// Send a [`TimelineVectorDiffs`].
+    pub fn send(&self, thread_update: TimelineVectorDiffs) {
+        let _ = self.thread_sender.send(thread_update);
+    }
+
+    /// Create a new [`Receiver`] of [`TimelineVectorDiffs`].
+    pub(super) fn new_thread_receiver(&self) -> Receiver<TimelineVectorDiffs> {
+        self.thread_sender.subscribe()
+    }
+}