matrix_sdk/event_cache/caches/room/
updates.rs

1// Copyright 2026 The Matrix.org Foundation C.I.C.
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::collections::BTreeMap;
16
17use matrix_sdk_base::{
18    deserialized_responses::{AmbiguityChange, ThreadSummary},
19    event_cache::{Event, Gap},
20    linked_chunk::{self, OwnedLinkedChunkId},
21};
22use ruma::{
23    OwnedEventId, OwnedMxcUri, OwnedRoomId, OwnedUserId, events::receipt::ReceiptEventContent,
24};
25use tokio::sync::broadcast::{Receiver, Sender};
26
27use super::super::TimelineVectorDiffs;
28
29/// An update related to events happened in a room.
30#[derive(Debug, Clone)]
31pub enum RoomEventCacheUpdate {
32    /// The fully read marker has moved to a different event.
33    MoveReadMarkerTo {
34        /// Event at which the read marker is now pointing.
35        event_id: OwnedEventId,
36    },
37
38    /// The members have changed.
39    UpdateMembers {
40        /// Collection of ambiguity changes that room member events trigger.
41        ///
42        /// This is a map of event ID of the `m.room.member` event to the
43        /// details of the ambiguity change.
44        ambiguity_changes: BTreeMap<OwnedEventId, AmbiguityChange>,
45
46        /// Collection of avatar changes that room member events trigger.
47        avatar_changes: Option<BTreeMap<OwnedUserId, Option<OwnedMxcUri>>>,
48    },
49
50    /// The room has received updates for the timeline as _diffs_.
51    UpdateTimelineEvents(TimelineVectorDiffs),
52
53    /// An event, which is a thread root, has received an update for its thread
54    /// summary.
55    UpdateThreadSummary {
56        /// The event ID of the thread root.
57        ///
58        /// Also called _thread ID_.
59        thread_root: OwnedEventId,
60
61        /// The updated thread summary.
62        thread_summary: ThreadSummary,
63    },
64
65    /// The room has received a new read receipt event.
66    AddReadReceiptEvent {
67        /// The event containing the receipts.
68        event: ReceiptEventContent,
69    },
70}
71
72/// Represents a timeline update of a room. It hides the details of
73/// [`RoomEventCacheUpdate`] by being more generic.
74///
75/// This is used by [`EventCache::subscribe_to_room_generic_updates`][0]. Please
76/// read it to learn more about the motivation behind this type.
77///
78/// [0]: super::super::super::EventCache::subscribe_to_room_generic_updates
79#[derive(Clone, Debug)]
80pub struct RoomEventCacheGenericUpdate {
81    /// The room ID owning the timeline.
82    pub room_id: OwnedRoomId,
83}
84
85/// An update being triggered when events change in the persisted event cache
86/// for any room.
87#[derive(Clone, Debug)]
88pub struct RoomEventCacheLinkedChunkUpdate {
89    /// The linked chunk affected by the update.
90    pub linked_chunk_id: OwnedLinkedChunkId,
91
92    /// A vector of all the linked chunk updates that happened during this event
93    /// cache update.
94    pub updates: Vec<linked_chunk::Update<Event, Gap>>,
95}
96
97impl RoomEventCacheLinkedChunkUpdate {
98    /// Return all the new events propagated by this update, in topological
99    /// order.
100    pub fn events(self) -> impl DoubleEndedIterator<Item = Event> {
101        use itertools::Either;
102        self.updates.into_iter().flat_map(|update| match update {
103            linked_chunk::Update::PushItems { items, .. } => {
104                Either::Left(Either::Left(items.into_iter()))
105            }
106            linked_chunk::Update::ReplaceItem { item, .. } => {
107                Either::Left(Either::Right(std::iter::once(item)))
108            }
109            linked_chunk::Update::RemoveItem { .. }
110            | linked_chunk::Update::DetachLastItems { .. }
111            | linked_chunk::Update::StartReattachItems
112            | linked_chunk::Update::EndReattachItems
113            | linked_chunk::Update::NewItemsChunk { .. }
114            | linked_chunk::Update::NewGapChunk { .. }
115            | linked_chunk::Update::RemoveChunk(..)
116            | linked_chunk::Update::Clear => {
117                // All these updates don't contain any new event.
118                Either::Right(std::iter::empty())
119            }
120        })
121    }
122}
123
124/// Sends [`RoomEventCacheLinkedChunkUpdate`]s to their observers.
125///
126/// The search index gets its own unbounded channel: a broadcast receiver that
127/// lags behind drops updates, whose events would then never get indexed.
128#[derive(Clone, Debug)]
129pub struct LinkedChunkUpdateSender {
130    sender: Sender<RoomEventCacheLinkedChunkUpdate>,
131
132    #[cfg(feature = "experimental-search")]
133    search_index_sender: tokio::sync::mpsc::UnboundedSender<(OwnedRoomId, Vec<Event>)>,
134}
135
136impl LinkedChunkUpdateSender {
137    /// Create a new [`LinkedChunkUpdateSender`], sending room events to be
138    /// indexed to `search_index_sender`.
139    pub fn new(
140        #[cfg(feature = "experimental-search")]
141        search_index_sender: tokio::sync::mpsc::UnboundedSender<(OwnedRoomId, Vec<Event>)>,
142    ) -> Self {
143        Self {
144            sender: Sender::new(128),
145            #[cfg(feature = "experimental-search")]
146            search_index_sender,
147        }
148    }
149
150    /// Send an update to all the observers.
151    pub fn send(&self, update: RoomEventCacheLinkedChunkUpdate) {
152        #[cfg(feature = "experimental-search")]
153        if let OwnedLinkedChunkId::Room(room_id) = &update.linked_chunk_id {
154            self.send_to_search_index(room_id.clone(), update.clone().events().collect());
155        }
156
157        let _ = self.sender.send(update);
158    }
159
160    /// Send events of a room to the search index only.
161    #[cfg(feature = "experimental-search")]
162    pub fn send_to_search_index(&self, room_id: OwnedRoomId, events: Vec<Event>) {
163        if !events.is_empty() {
164            let _ = self.search_index_sender.send((room_id, events));
165        }
166    }
167
168    /// Subscribe to the updates.
169    pub fn subscribe(&self) -> Receiver<RoomEventCacheLinkedChunkUpdate> {
170        self.sender.subscribe()
171    }
172}
173
174/// A small type to send updates in all channels.
175#[derive(Clone)]
176pub struct RoomEventCacheUpdateSender {
177    room_sender: Sender<RoomEventCacheUpdate>,
178    generic_sender: Sender<RoomEventCacheGenericUpdate>,
179}
180
181impl RoomEventCacheUpdateSender {
182    /// Create a new [`RoomEventCacheUpdateSender`].
183    pub fn new(generic_sender: Sender<RoomEventCacheGenericUpdate>) -> Self {
184        Self { room_sender: Sender::new(32), generic_sender }
185    }
186
187    /// Send a [`RoomEventCacheUpdate`] and an optional
188    /// [`RoomEventCacheGenericUpdate`].
189    pub fn send(
190        &self,
191        room_update: RoomEventCacheUpdate,
192        generic_update: Option<RoomEventCacheGenericUpdate>,
193    ) {
194        let _ = self.room_sender.send(room_update);
195
196        if let Some(generic_update) = generic_update {
197            let _ = self.generic_sender.send(generic_update);
198        }
199    }
200
201    /// Get the generic update sender.
202    pub(in super::super) fn generic_update_sender(&self) -> &Sender<RoomEventCacheGenericUpdate> {
203        &self.generic_sender
204    }
205
206    /// Create a new [`Receiver`] of [`RoomEventCacheUpdate`].
207    pub(super) fn new_room_receiver(&self) -> Receiver<RoomEventCacheUpdate> {
208        self.room_sender.subscribe()
209    }
210}
matrix_sdk/event_cache/caches/room/updates.rs

matrix_sdk/event_cache/caches/room/
updates.rs
