matrix_sdk/paginators/
thread.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//! Paginator facilities for a thread.
16//!
17//! See also the documentation for the [`ThreadedEventsLoader`] struct.
18
19use std::{
20    fmt::Formatter,
21    future::Future,
22    sync::{Arc, Mutex},
23};
24
25use matrix_sdk_base::{SendOutsideWasm, SyncOutsideWasm, deserialized_responses::TimelineEvent};
26use ruma::{EventId, OwnedEventId, UInt, api::Direction};
27
28use crate::{
29    Error, Room,
30    paginators::{PaginationResult, PaginationToken, PaginationTokens, PaginatorError},
31    room::{IncludeRelations, Relations, RelationsOptions},
32};
33
34/// A paginable thread interface, useful for testing purposes.
35pub trait PaginableThread: SendOutsideWasm + SyncOutsideWasm {
36    /// Runs a /relations query for the given thread, with the given options.
37    fn relations(
38        &self,
39        thread_root: OwnedEventId,
40        opts: RelationsOptions,
41    ) -> impl Future<Output = Result<Relations, Error>> + SendOutsideWasm;
42
43    /// Load an event, given its event ID.
44    fn load_event(
45        &self,
46        event_id: &OwnedEventId,
47    ) -> impl Future<Output = Result<TimelineEvent, Error>> + SendOutsideWasm;
48}
49
50impl PaginableThread for Room {
51    async fn relations(
52        &self,
53        thread_root: OwnedEventId,
54        opts: RelationsOptions,
55    ) -> Result<Relations, Error> {
56        self.relations(thread_root, opts).await
57    }
58
59    async fn load_event(&self, event_id: &OwnedEventId) -> Result<TimelineEvent, Error> {
60        self.event(event_id, None).await
61    }
62}
63
64/// A paginator for a thread of events.
65pub struct ThreadedEventsLoader<P: PaginableThread> {
66    /// Room provider for the paginated thread.
67    room: P,
68
69    /// The thread root event ID (the event that started the thread).
70    root_event_id: OwnedEventId,
71
72    /// The current pagination tokens, which are used to keep track of the
73    /// pagination state.
74    tokens: Mutex<PaginationTokens>,
75}
76
77impl<P: PaginableThread> ThreadedEventsLoader<P> {
78    /// Create a new [`ThreadedEventsLoader`], given a room implementation.
79    pub fn new(room: P, root_event_id: OwnedEventId, tokens: PaginationTokens) -> Self {
80        Self { room, root_event_id, tokens: Mutex::new(tokens) }
81    }
82
83    /// Run a single pagination backwards, returning the next set of events and
84    /// information whether we've reached the start of the thread.
85    ///
86    /// Note: when the thread start is reached, the root event _will_ be
87    /// included in the result.
88    pub async fn paginate_backwards(
89        &self,
90        num_events: UInt,
91    ) -> Result<PaginationResult, PaginatorError> {
92        let token = {
93            let token = &self.tokens.lock().unwrap().previous;
94
95            match token {
96                PaginationToken::None => None,
97                PaginationToken::HasMore(token) => Some(token.clone()),
98                PaginationToken::HitEnd => {
99                    return Ok(PaginationResult { events: Vec::new(), hit_end_of_timeline: true });
100                }
101            }
102        };
103
104        let options = RelationsOptions {
105            from: token,
106            dir: Direction::Backward,
107            limit: Some(num_events),
108            include_relations: IncludeRelations::AllRelations,
109            recurse: true,
110        };
111
112        let mut result = self
113            .room
114            .relations(self.root_event_id.to_owned(), options)
115            .await
116            .map_err(|error| PaginatorError::SdkError(Arc::new(error)))?;
117
118        let hit_end_of_timeline = result.next_batch_token.is_none();
119
120        // Update the stored tokens
121        {
122            let mut tokens = self.tokens.lock().unwrap();
123
124            tokens.previous = match result.next_batch_token {
125                Some(val) => PaginationToken::HasMore(val),
126                None => PaginationToken::HitEnd,
127            };
128        }
129
130        // Finally insert the thread root if at the end of the timeline going
131        // backwards
132        if hit_end_of_timeline {
133            let root_event = self
134                .room
135                .load_event(&self.root_event_id)
136                .await
137                .map_err(|err| PaginatorError::SdkError(Arc::new(err)))?;
138
139            result.chunk.push(root_event);
140        }
141
142        Ok(PaginationResult { events: result.chunk, hit_end_of_timeline })
143    }
144
145    /// Run a single pagination forwards, returning the next set of events and
146    /// information whether we've reached the end of the thread.
147    pub async fn paginate_forwards(
148        &self,
149        num_events: UInt,
150    ) -> Result<PaginationResult, PaginatorError> {
151        let token = {
152            let token = &self.tokens.lock().unwrap().next;
153
154            match token {
155                PaginationToken::None => None,
156                PaginationToken::HasMore(token) => Some(token.clone()),
157                PaginationToken::HitEnd => {
158                    return Ok(PaginationResult { events: Vec::new(), hit_end_of_timeline: true });
159                }
160            }
161        };
162
163        let options = RelationsOptions {
164            from: token,
165            dir: Direction::Forward,
166            limit: Some(num_events),
167            include_relations: IncludeRelations::AllRelations,
168            recurse: true,
169        };
170
171        let result = self
172            .room
173            .relations(self.root_event_id.to_owned(), options)
174            .await
175            .map_err(|error| PaginatorError::SdkError(Arc::new(error)))?;
176
177        let hit_end_of_timeline = result.next_batch_token.is_none();
178
179        // Update the stored tokens
180        {
181            let mut tokens = self.tokens.lock().unwrap();
182
183            tokens.next = match result.next_batch_token {
184                Some(val) => PaginationToken::HasMore(val),
185                None => PaginationToken::HitEnd,
186            };
187        }
188
189        Ok(PaginationResult { events: result.chunk, hit_end_of_timeline })
190    }
191
192    /// Returns the root [`EventId`] for the thread.
193    pub fn thread_root_event_id(&self) -> &EventId {
194        &self.root_event_id
195    }
196}
197
198#[cfg(not(tarpaulin_include))]
199impl<P: PaginableThread> std::fmt::Debug for ThreadedEventsLoader<P> {
200    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
201        f.debug_struct("ThreadedEventsLoader").finish()
202    }
203}
matrix_sdk/paginators/thread.rs

matrix_sdk/paginators/
thread.rs
