matrix_sdk/event_cache/room/
threads.rs

1// Copyright 2025 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
15//! Threads-related data structures.
16
17use std::collections::BTreeSet;
18
19use eyeball_im::VectorDiff;
20use matrix_sdk_base::{
21    event_cache::{Event, Gap},
22    linked_chunk::{ChunkContent, OwnedLinkedChunkId, Position},
23};
24use ruma::{EventId, OwnedEventId, OwnedRoomId};
25use tokio::sync::broadcast::{Receiver, Sender};
26use tracing::{error, trace};
27
28use crate::event_cache::{
29    BackPaginationOutcome, EventsOrigin, RoomEventCacheLinkedChunkUpdate,
30    deduplicator::DeduplicationOutcome,
31    room::{LoadMoreEventsBackwardsOutcome, events::EventLinkedChunk},
32};
33
34/// An update coming from a thread event cache.
35#[derive(Clone, Debug)]
36pub struct ThreadEventCacheUpdate {
37    /// New vector diff for the thread timeline.
38    pub diffs: Vec<VectorDiff<Event>>,
39    /// The origin that triggered this update.
40    pub origin: EventsOrigin,
41}
42
43/// All the information related to a single thread.
44pub(crate) struct ThreadEventCache {
45    /// The room owning this thread.
46    room_id: OwnedRoomId,
47
48    /// The ID of the thread root event, which is the first event in the thread
49    /// (and eventually the first in the linked chunk).
50    thread_root: OwnedEventId,
51
52    /// The linked chunk for this thread.
53    chunk: EventLinkedChunk,
54
55    /// A sender for live events updates in this thread.
56    sender: Sender<ThreadEventCacheUpdate>,
57
58    /// A sender for the globally observable linked chunk updates that happened
59    /// during a sync or a back-pagination.
60    ///
61    /// See also [`super::super::EventCacheInner::linked_chunk_update_sender`].
62    linked_chunk_update_sender: Sender<RoomEventCacheLinkedChunkUpdate>,
63}
64
65impl ThreadEventCache {
66    /// Create a new empty thread event cache.
67    pub fn new(
68        room_id: OwnedRoomId,
69        thread_root: OwnedEventId,
70        linked_chunk_update_sender: Sender<RoomEventCacheLinkedChunkUpdate>,
71    ) -> Self {
72        Self {
73            chunk: EventLinkedChunk::new(),
74            sender: Sender::new(32),
75            room_id,
76            thread_root,
77            linked_chunk_update_sender,
78        }
79    }
80
81    /// Subscribe to live events from this thread.
82    pub fn subscribe(&self) -> (Vec<Event>, Receiver<ThreadEventCacheUpdate>) {
83        let events = self.chunk.events().map(|(_position, item)| item.clone()).collect();
84
85        let recv = self.sender.subscribe();
86
87        (events, recv)
88    }
89
90    /// Clear a thread, after a gappy sync for instance.
91    pub fn clear(&mut self) {
92        self.chunk.reset();
93
94        let diffs = self.chunk.updates_as_vector_diffs();
95        if !diffs.is_empty() {
96            let _ = self.sender.send(ThreadEventCacheUpdate { diffs, origin: EventsOrigin::Cache });
97        }
98    }
99
100    // TODO(bnjbvr): share more code with `RoomEventCacheState` to avoid the
101    // duplication here too.
102    fn propagate_changes(&mut self) {
103        // This is a lie, at the moment! We're not persisting threads yet, so we're just
104        // forwarding all updates to the linked chunk update sender.
105        let updates = self.chunk.store_updates().take();
106
107        let _ = self.linked_chunk_update_sender.send(RoomEventCacheLinkedChunkUpdate {
108            updates,
109            linked_chunk_id: OwnedLinkedChunkId::Thread(
110                self.room_id.clone(),
111                self.thread_root.clone(),
112            ),
113        });
114    }
115
116    /// Push some live events to this thread, and propagate the updates to
117    /// the listeners.
118    pub fn add_live_events(&mut self, events: Vec<Event>) {
119        if events.is_empty() {
120            return;
121        }
122
123        let deduplication = self.filter_duplicate_events(events);
124
125        if deduplication.non_empty_all_duplicates {
126            // If all events are duplicates, we don't need to do anything; ignore
127            // the new events.
128            return;
129        }
130
131        // Remove the duplicated events from the thread chunk.
132        self.remove_in_memory_duplicated_events(deduplication.in_memory_duplicated_event_ids);
133        assert!(
134            deduplication.in_store_duplicated_event_ids.is_empty(),
135            "persistent storage for threads is not implemented yet"
136        );
137
138        let events = deduplication.all_events;
139
140        self.chunk.push_live_events(None, &events);
141
142        self.propagate_changes();
143
144        let diffs = self.chunk.updates_as_vector_diffs();
145        if !diffs.is_empty() {
146            let _ = self.sender.send(ThreadEventCacheUpdate { diffs, origin: EventsOrigin::Sync });
147        }
148    }
149
150    /// Remove an event from an thread event linked chunk, if it exists.
151    ///
152    /// If the event has been found and removed, then an update will be
153    /// propagated to observers.
154    pub(crate) fn remove_if_present(&mut self, event_id: &EventId) {
155        let Some(pos) = self.chunk.events().find_map(|(pos, event)| {
156            (event.event_id().as_deref() == Some(event_id)).then_some(pos)
157        }) else {
158            // Event not found in the linked chunk, nothing to do.
159            return;
160        };
161
162        if let Err(err) = self.chunk.remove_events_by_position(vec![pos]) {
163            error!(%err, "a thread linked chunk position was valid a few lines above, but invalid when deleting");
164            return;
165        }
166
167        // We've touched the linked chunk; propagate changes to storage and observers.
168        self.propagate_changes();
169
170        let diffs = self.chunk.updates_as_vector_diffs();
171        if !diffs.is_empty() {
172            let _ = self.sender.send(ThreadEventCacheUpdate { diffs, origin: EventsOrigin::Sync });
173        }
174    }
175
176    /// Simplified version of
177    /// [`RoomEventCacheState::load_more_events_backwards`], which
178    /// returns the outcome of the pagination without actually loading from
179    /// disk.
180    pub fn load_more_events_backwards(&self) -> LoadMoreEventsBackwardsOutcome {
181        // If any in-memory chunk is a gap, don't load more events, and let the caller
182        // resolve the gap.
183        if let Some(prev_token) = self.chunk.rgap().map(|gap| gap.prev_token) {
184            trace!(%prev_token, "thread chunk has at least a gap");
185            return LoadMoreEventsBackwardsOutcome::Gap { prev_token: Some(prev_token) };
186        }
187
188        // If we don't have a gap, then the first event should be the the thread's root;
189        // otherwise, we'll restart a pagination from the end.
190        if let Some((_pos, event)) = self.chunk.events().next() {
191            let first_event_id =
192                event.event_id().expect("a linked chunk only stores events with IDs");
193
194            if first_event_id == self.thread_root {
195                trace!("thread chunk is fully loaded and non-empty: reached_start=true");
196                return LoadMoreEventsBackwardsOutcome::StartOfTimeline;
197            }
198        }
199
200        // Otherwise, we don't have a gap nor events. We don't have anything. Poor us.
201        // Well, is ok: start a pagination from the end.
202        LoadMoreEventsBackwardsOutcome::Gap { prev_token: None }
203    }
204
205    /// Find duplicates in a thread, until there's persistent storage for
206    /// those.
207    ///
208    /// TODO: when persistent storage is implemented for thread, only use
209    /// the regular `filter_duplicate_events` method.
210    fn filter_duplicate_events(&self, mut new_events: Vec<Event>) -> DeduplicationOutcome {
211        let mut new_event_ids = BTreeSet::new();
212
213        new_events.retain(|event| {
214            // Only keep events with IDs, and those for which `insert` returns `true`
215            // (meaning they were not in the set).
216            event.event_id().is_some_and(|event_id| new_event_ids.insert(event_id))
217        });
218
219        let in_memory_duplicated_event_ids: Vec<_> = self
220            .chunk
221            .events()
222            .filter_map(|(position, event)| {
223                let event_id = event.event_id()?;
224                new_event_ids.contains(&event_id).then_some((event_id, position))
225            })
226            .collect();
227
228        // Right now, there's no persistent storage for threads.
229        let in_store_duplicated_event_ids = Vec::new();
230
231        let at_least_one_event = !new_events.is_empty();
232        let all_duplicates = (in_memory_duplicated_event_ids.len()
233            + in_store_duplicated_event_ids.len())
234            == new_events.len();
235        let non_empty_all_duplicates = at_least_one_event && all_duplicates;
236
237        DeduplicationOutcome {
238            all_events: new_events,
239            in_memory_duplicated_event_ids,
240            in_store_duplicated_event_ids,
241            non_empty_all_duplicates,
242        }
243    }
244
245    /// Remove in-memory duplicated events from the thread chunk, that have
246    /// been found with [`Self::filter_duplicate_events`].
247    ///
248    /// Precondition: the `in_memory_duplicated_event_ids` must be the
249    /// result of the above function, otherwise this can panic.
250    fn remove_in_memory_duplicated_events(
251        &mut self,
252        in_memory_duplicated_event_ids: Vec<(OwnedEventId, Position)>,
253    ) {
254        // Remove the duplicated events from the thread chunk.
255        self.chunk
256            .remove_events_by_position(
257                in_memory_duplicated_event_ids
258                    .iter()
259                    .map(|(_event_id, position)| *position)
260                    .collect(),
261            )
262            .expect("we collected the position of the events to remove just before");
263    }
264
265    /// Finish a network pagination started with the gap retrieved from
266    /// [`Self::load_more_events_backwards`].
267    ///
268    /// Returns `None` if the gap couldn't be found anymore (meaning the
269    /// thread has been reset while the pagination was ongoing).
270    pub fn finish_network_pagination(
271        &mut self,
272        prev_token: Option<String>,
273        new_token: Option<String>,
274        events: Vec<Event>,
275    ) -> Option<BackPaginationOutcome> {
276        // TODO(bnjbvr): consider deduplicating this code (~same for room) at some
277        // point.
278        let prev_gap_id = if let Some(token) = prev_token {
279            // If the gap id is missing, it means that the gap disappeared during
280            // pagination; in this case, early return to the caller.
281            let gap_id = self.chunk.chunk_identifier(|chunk| {
282                    matches!(chunk.content(), ChunkContent::Gap(Gap { prev_token }) if *prev_token == token)
283                })?;
284
285            Some(gap_id)
286        } else {
287            None
288        };
289
290        // This is a backwards pagination, so the events were returned in the reverse
291        // topological order.
292        let topo_ordered_events = events.iter().cloned().rev().collect::<Vec<_>>();
293        let new_gap = new_token.map(|token| Gap { prev_token: token });
294
295        let deduplication = self.filter_duplicate_events(topo_ordered_events);
296
297        let (events, new_gap) = if deduplication.non_empty_all_duplicates {
298            // If all events are duplicates, we don't need to do anything; ignore
299            // the new events and the new gap.
300            (Vec::new(), None)
301        } else {
302            assert!(
303                deduplication.in_store_duplicated_event_ids.is_empty(),
304                "persistent storage for threads is not implemented yet"
305            );
306            self.remove_in_memory_duplicated_events(deduplication.in_memory_duplicated_event_ids);
307
308            // Keep events and the gap.
309            (deduplication.all_events, new_gap)
310        };
311
312        // Add the paginated events to the thread chunk.
313        let reached_start = self.chunk.finish_back_pagination(prev_gap_id, new_gap, &events);
314
315        self.propagate_changes();
316
317        // Notify observers about the updates.
318        let updates = self.chunk.updates_as_vector_diffs();
319        if !updates.is_empty() {
320            // Send the updates to the listeners.
321            let _ = self
322                .sender
323                .send(ThreadEventCacheUpdate { diffs: updates, origin: EventsOrigin::Pagination });
324        }
325
326        Some(BackPaginationOutcome { reached_start, events })
327    }
328
329    /// Returns the latest event ID in this thread, if any.
330    pub fn latest_event_id(&self) -> Option<OwnedEventId> {
331        self.chunk.revents().next().and_then(|(_position, event)| event.event_id())
332    }
333}
matrix_sdk/event_cache/room/threads.rs

matrix_sdk/event_cache/room/
threads.rs
