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, Position},
23};
24use ruma::OwnedEventId;
25use tokio::sync::broadcast::{Receiver, Sender};
26use tracing::trace;
27
28use crate::event_cache::{
29    deduplicator::DeduplicationOutcome,
30    room::{events::EventLinkedChunk, LoadMoreEventsBackwardsOutcome},
31    BackPaginationOutcome, EventsOrigin,
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 ID of the thread root event, which is the first event in the thread
46    /// (and eventually the first in the linked chunk).
47    thread_root: OwnedEventId,
48
49    /// The linked chunk for this thread.
50    chunk: EventLinkedChunk,
51
52    /// A sender for live events updates in this thread.
53    sender: Sender<ThreadEventCacheUpdate>,
54}
55
56impl ThreadEventCache {
57    /// Create a new empty thread event cache.
58    pub fn new(thread_root: OwnedEventId) -> Self {
59        Self { chunk: EventLinkedChunk::new(), sender: Sender::new(32), thread_root }
60    }
61
62    /// Subscribe to live events from this thread.
63    pub fn subscribe(&self) -> (Vec<Event>, Receiver<ThreadEventCacheUpdate>) {
64        let events = self.chunk.events().map(|(_position, item)| item.clone()).collect();
65
66        let recv = self.sender.subscribe();
67
68        (events, recv)
69    }
70
71    /// Clear a thread, after a gappy sync for instance.
72    pub fn clear(&mut self) {
73        self.chunk.reset();
74
75        let diffs = self.chunk.updates_as_vector_diffs();
76        if !diffs.is_empty() {
77            let _ = self.sender.send(ThreadEventCacheUpdate { diffs, origin: EventsOrigin::Cache });
78        }
79    }
80
81    /// Push some live events to this thread, and propagate the updates to
82    /// the listeners.
83    pub fn add_live_events(&mut self, events: Vec<Event>) {
84        if events.is_empty() {
85            return;
86        }
87
88        let deduplication = self.filter_duplicate_events(events);
89
90        if deduplication.non_empty_all_duplicates {
91            // If all events are duplicates, we don't need to do anything; ignore
92            // the new events.
93            return;
94        }
95
96        // Remove the duplicated events from the thread chunk.
97        self.remove_in_memory_duplicated_events(deduplication.in_memory_duplicated_event_ids);
98        assert!(
99            deduplication.in_store_duplicated_event_ids.is_empty(),
100            "persistent storage for threads is not implemented yet"
101        );
102
103        let events = deduplication.all_events;
104
105        self.chunk.push_live_events(None, &events);
106
107        let diffs = self.chunk.updates_as_vector_diffs();
108        if !diffs.is_empty() {
109            let _ = self.sender.send(ThreadEventCacheUpdate { diffs, origin: EventsOrigin::Sync });
110        }
111    }
112
113    /// Simplified version of
114    /// [`RoomEventCacheState::load_more_events_backwards`], which
115    /// returns the outcome of the pagination without actually loading from
116    /// disk.
117    pub fn load_more_events_backwards(&self) -> LoadMoreEventsBackwardsOutcome {
118        // If any in-memory chunk is a gap, don't load more events, and let the caller
119        // resolve the gap.
120        if let Some(prev_token) = self.chunk.rgap().map(|gap| gap.prev_token) {
121            trace!(%prev_token, "thread chunk has at least a gap");
122            return LoadMoreEventsBackwardsOutcome::Gap { prev_token: Some(prev_token) };
123        }
124
125        // If we don't have a gap, then the first event should be the the thread's root;
126        // otherwise, we'll restart a pagination from the end.
127        if let Some((_pos, event)) = self.chunk.events().next() {
128            let first_event_id =
129                event.event_id().expect("a linked chunk only stores events with IDs");
130
131            if first_event_id == self.thread_root {
132                trace!("thread chunk is fully loaded and non-empty: reached_start=true");
133                return LoadMoreEventsBackwardsOutcome::StartOfTimeline;
134            }
135        }
136
137        // Otherwise, we don't have a gap nor events. We don't have anything. Poor us.
138        // Well, is ok: start a pagination from the end.
139        LoadMoreEventsBackwardsOutcome::Gap { prev_token: None }
140    }
141
142    /// Find duplicates in a thread, until there's persistent storage for
143    /// those.
144    ///
145    /// TODO: when persistent storage is implemented for thread, only use
146    /// the regular `filter_duplicate_events` method.
147    fn filter_duplicate_events(&self, mut new_events: Vec<Event>) -> DeduplicationOutcome {
148        let mut new_event_ids = BTreeSet::new();
149
150        new_events.retain(|event| {
151            // Only keep events with IDs, and those for which `insert` returns `true`
152            // (meaning they were not in the set).
153            event.event_id().is_some_and(|event_id| new_event_ids.insert(event_id))
154        });
155
156        let in_memory_duplicated_event_ids: Vec<_> = self
157            .chunk
158            .events()
159            .filter_map(|(position, event)| {
160                let event_id = event.event_id()?;
161                new_event_ids.contains(&event_id).then_some((event_id, position))
162            })
163            .collect();
164
165        // Right now, there's no persistent storage for threads.
166        let in_store_duplicated_event_ids = Vec::new();
167
168        let at_least_one_event = !new_events.is_empty();
169        let all_duplicates = (in_memory_duplicated_event_ids.len()
170            + in_store_duplicated_event_ids.len())
171            == new_events.len();
172        let non_empty_all_duplicates = at_least_one_event && all_duplicates;
173
174        DeduplicationOutcome {
175            all_events: new_events,
176            in_memory_duplicated_event_ids,
177            in_store_duplicated_event_ids,
178            non_empty_all_duplicates,
179        }
180    }
181
182    /// Remove in-memory duplicated events from the thread chunk, that have
183    /// been found with [`Self::filter_duplicate_events`].
184    ///
185    /// Precondition: the `in_memory_duplicated_event_ids` must be the
186    /// result of the above function, otherwise this can panic.
187    fn remove_in_memory_duplicated_events(
188        &mut self,
189        in_memory_duplicated_event_ids: Vec<(OwnedEventId, Position)>,
190    ) {
191        // Remove the duplicated events from the thread chunk.
192        self.chunk
193            .remove_events_by_position(
194                in_memory_duplicated_event_ids
195                    .iter()
196                    .map(|(_event_id, position)| *position)
197                    .collect(),
198            )
199            .expect("we collected the position of the events to remove just before");
200    }
201
202    /// Finish a network pagination started with the gap retrieved from
203    /// [`Self::load_more_events_backwards`].
204    ///
205    /// Returns `None` if the gap couldn't be found anymore (meaning the
206    /// thread has been reset while the pagination was ongoing).
207    pub fn finish_network_pagination(
208        &mut self,
209        prev_token: Option<String>,
210        new_token: Option<String>,
211        events: Vec<Event>,
212    ) -> Option<BackPaginationOutcome> {
213        // TODO(bnjbvr): consider deduplicating this code (~same for room) at some
214        // point.
215        let prev_gap_id = if let Some(token) = prev_token {
216            // If the gap id is missing, it means that the gap disappeared during
217            // pagination; in this case, early return to the caller.
218            let gap_id = self.chunk.chunk_identifier(|chunk| {
219                    matches!(chunk.content(), ChunkContent::Gap(Gap { ref prev_token }) if *prev_token == token)
220                })?;
221
222            Some(gap_id)
223        } else {
224            None
225        };
226
227        // This is a backwards pagination, so the events were returned in the reverse
228        // topological order.
229        let topo_ordered_events = events.iter().cloned().rev().collect::<Vec<_>>();
230        let new_gap = new_token.map(|token| Gap { prev_token: token });
231
232        let deduplication = self.filter_duplicate_events(topo_ordered_events);
233
234        let (events, new_gap) = if deduplication.non_empty_all_duplicates {
235            // If all events are duplicates, we don't need to do anything; ignore
236            // the new events and the new gap.
237            (Vec::new(), None)
238        } else {
239            assert!(
240                deduplication.in_store_duplicated_event_ids.is_empty(),
241                "persistent storage for threads is not implemented yet"
242            );
243            self.remove_in_memory_duplicated_events(deduplication.in_memory_duplicated_event_ids);
244
245            // Keep events and the gap.
246            (deduplication.all_events, new_gap)
247        };
248
249        // Add the paginated events to the thread chunk.
250        let reached_start = self.chunk.finish_back_pagination(prev_gap_id, new_gap, &events);
251
252        // Notify observers about the updates.
253        let updates = self.chunk.updates_as_vector_diffs();
254        if !updates.is_empty() {
255            // Send the updates to the listeners.
256            let _ = self
257                .sender
258                .send(ThreadEventCacheUpdate { diffs: updates, origin: EventsOrigin::Pagination });
259        }
260
261        Some(BackPaginationOutcome { reached_start, events })
262    }
263
264    /// Returns the latest event ID in this thread, if any.
265    pub fn latest_event_id(&self) -> Option<OwnedEventId> {
266        self.chunk.revents().next().and_then(|(_position, event)| event.event_id())
267    }
268}
matrix_sdk/event_cache/room/threads.rs

matrix_sdk/event_cache/room/
threads.rs
