matrix_sdk_search/index/
mod.rs

1// Copyright 2024 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/// A module for building a [`RoomIndex`]
16pub mod builder;
17
18use std::{
19    collections::{HashMap, HashSet},
20    fmt,
21};
22
23use once_cell::sync::OnceCell;
24use ruma::{
25    EventId, MilliSecondsSinceUnixEpoch, OwnedEventId, OwnedRoomId, OwnedUserId, RoomId, UInt,
26};
27use tantivy::{
28    DateTime, Index, IndexReader, Order, ReloadPolicy, TantivyDocument,
29    collector::{Count, MultiCollector, TopDocs},
30    directory::error::OpenDirectoryError,
31    merge_policy::LogMergePolicy,
32    query::QueryParser,
33    schema::Value,
34};
35use tracing::{debug, error, warn};
36
37use crate::{
38    OpStamp, TANTIVY_INDEX_MEMORY_BUDGET,
39    error::IndexError,
40    schema::{MatrixSearchIndexSchema, RoomMessageSchema},
41    writer::SearchIndexWriter,
42};
43
44/// The result type for a search operation.
45#[derive(Debug)]
46pub struct SearchResult {
47    /// The total count of search results that were found.
48    ///
49    /// Might be bigger than the length of [`SearchResult::events`] as the
50    /// search is paginated.
51    pub total_count: usize,
52
53    /// A list of search results.
54    ///
55    /// Each entry contains the timestamp and the ID of the event that matched
56    /// the search query, newest first. The list may be incomplete because the
57    /// search results are paginated; use [`SearchResult::total_count`] to
58    /// determine the total number of matching results.
59    pub events: Vec<(MilliSecondsSinceUnixEpoch, OwnedEventId)>,
60}
61
62/// The subset of an event's data required to index it and later retrieve it.
63///
64/// Produced by the matrix-sdk layer, which knows how to extract searchable text
65/// from each event type. This crate stays agnostic to Matrix event content.
66#[derive(Clone)]
67pub struct IndexableEvent {
68    /// The event's own id (primary key).
69    pub(crate) event_id: OwnedEventId,
70    /// The id used as the deletion key: the original event id for edits,
71    /// otherwise the event's own id.
72    pub(crate) original_event_id: OwnedEventId,
73    /// The sender of the event.
74    pub(crate) sender: OwnedUserId,
75    /// The origin server timestamp of the event.
76    ///
77    /// Please use the `matrix_sdk_common::TimelineEvent::timestamp` as much as
78    /// possible as it protects against malformed `origin_server_ts`. At worst,
79    /// use the `matrix_sdk_common::serde_helpers::extract_timestamp` function.
80    pub(crate) timestamp: Option<MilliSecondsSinceUnixEpoch>,
81    /// The text to index for this event.
82    pub(crate) body: String,
83}
84
85impl fmt::Debug for IndexableEvent {
86    /// Don't log bodies
87    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
88        f.debug_struct("IndexableEvent")
89            .field("event_id", &self.event_id)
90            .field("original_event_id", &self.original_event_id)
91            .field("sender", &self.sender)
92            .field("timestamp", &self.timestamp)
93            .finish_non_exhaustive()
94    }
95}
96
97/// Maximum value for the timestamp to not overflow when converted to
98/// nanoseconds by Tantivy. See [`IndexableEvent::new`] to learn more.
99const MAX_MILLISECONDS: u64 = (i64::MAX / 1_000_000).cast_unsigned();
100
101impl IndexableEvent {
102    /// Create a new [`IndexableEvent`].
103    pub fn new(
104        event_id: OwnedEventId,
105        original_event_id: OwnedEventId,
106        sender: OwnedUserId,
107        mut timestamp: Option<MilliSecondsSinceUnixEpoch>,
108        body: String,
109    ) -> Self {
110        // Tantivy will transform the number of milliseconds to nanoseconds by
111        // multiplying by 1_000_000 ([1]). If the number of milliseconds is too
112        // big, the multiplication will overflow.
113        //
114        // To avoid this panic, we cap the number of milliseconds to a maximum
115        // value.
116        //
117        // [1]: https://github.com/quickwit-oss/tantivy/blob/31ca1a8ba290b425f871d2e2384592045ec01b8d/common/src/datetime.rs#L62-L67
118        if let Some(timestamp) = &mut timestamp {
119            *timestamp = MilliSecondsSinceUnixEpoch(
120                timestamp.get().min(UInt::new_saturating(MAX_MILLISECONDS)),
121            );
122        }
123
124        Self { event_id, original_event_id, sender, timestamp, body }
125    }
126}
127
128/// A struct to represent the operations on a [`RoomIndex`]
129#[derive(Debug, Clone)]
130pub enum RoomIndexOperation {
131    /// Add this event to the index.
132    Add(IndexableEvent),
133    /// Remove all documents in the index where
134    /// `MatrixSearchIndexSchema::deletion_key()` matches this event id.
135    Remove(OwnedEventId),
136    /// Replace all documents in the index where
137    /// `MatrixSearchIndexSchema::deletion_key()` matches this event id with the
138    /// new event.
139    Edit(OwnedEventId, IndexableEvent),
140    /// Do nothing.
141    Noop,
142}
143
144/// A struct that holds all data pertaining to a particular room's message
145/// index.
146pub struct RoomIndex {
147    index: Index,
148    schema: RoomMessageSchema,
149    query_parser: QueryParser,
150    room_id: OwnedRoomId,
151    /// Events added but not yet committed, mapping each document's primary key
152    /// (event id) to its deletion key (original event id). The deletion key is
153    /// needed so that a [`RoomIndex::remove`] in the same uncommitted batch,
154    /// which deletes by that key, can reconcile these entries too.
155    uncommitted_adds: HashMap<OwnedEventId, OwnedEventId>,
156    uncommitted_removes: HashSet<OwnedEventId>,
157    /// Cached [`IndexReader`].
158    ///
159    /// It is costly to create one, so let's keep it in memory when needed; see
160    /// [`RoomIndex::reader`] to learn more.
161    reader: OnceCell<IndexReader>,
162}
163
164impl fmt::Debug for RoomIndex {
165    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
166        f.debug_struct("RoomIndex")
167            .field("schema", &self.schema)
168            .field("room_id", &self.room_id)
169            .finish()
170    }
171}
172
173impl RoomIndex {
174    pub(crate) fn new_with(index: Index, schema: RoomMessageSchema, room_id: &RoomId) -> RoomIndex {
175        let query_parser = QueryParser::for_index(&index, schema.default_search_fields());
176        Self {
177            index,
178            schema,
179            query_parser,
180            room_id: room_id.to_owned(),
181            uncommitted_adds: HashMap::new(),
182            uncommitted_removes: HashSet::new(),
183            reader: OnceCell::new(),
184        }
185    }
186
187    /// Get a [`SearchIndexWriter`] for this index.
188    ///
189    /// Every commit creates a segment, and opening an index opens every file of
190    /// every segment, which dominates the cost of opening many room indexes.
191    /// Merge as soon as there are two segments of the same layer, so a room
192    /// index usually stays at a single segment.
193    fn writer(&self) -> Result<SearchIndexWriter, IndexError> {
194        let writer = self.index.writer(TANTIVY_INDEX_MEMORY_BUDGET)?;
195
196        let mut merge_policy = LogMergePolicy::default();
197        merge_policy.set_min_num_segments(2);
198        writer.set_merge_policy(Box::new(merge_policy));
199
200        Ok(SearchIndexWriter::new(writer, self.schema.clone()))
201    }
202
203    /// Get or create the cached [`IndexReader`] for this index.
204    ///
205    /// The reload policy is [`ReloadPolicy::Manual`]: we are the only writer of
206    /// this index, and every commit goes through
207    /// [`RoomIndex::commit_and_reload`], which reloads explicitly. Tantivy's
208    /// default policy would instead spawn one meta file watcher thread per
209    /// index, i.e. one per room, and panics if that thread cannot be spawned.
210    fn reader(&self) -> Result<&IndexReader, IndexError> {
211        self.reader.get_or_try_init(|| {
212            Ok(self.index.reader_builder().reload_policy(ReloadPolicy::Manual).try_into()?)
213        })
214    }
215
216    /// Do the costly part of opening this index ahead of the first search, i.e.
217    /// opening all its segments.
218    pub fn warm_up(&self) -> Result<(), IndexError> {
219        self.reader().map(|_| ())
220    }
221
222    /// Commit added events to [`RoomIndex`]. The changes are not reflected in
223    /// the search results until the serchers are reloaded.
224    ///
225    /// Use [`RoomIndex::commit_and_reload`] for this purpose.
226    fn commit(&mut self, writer: &mut SearchIndexWriter) -> Result<OpStamp, IndexError> {
227        let last_commit_opstamp = writer.commit()?; // TODO: This is blocking. Handle it.
228        self.uncommitted_adds.clear();
229        self.uncommitted_removes.clear();
230        Ok(last_commit_opstamp)
231    }
232
233    /// Commit added events to [`RoomIndex`] and update searchers so that they
234    /// reflect the state of the last `.commit()`.
235    ///
236    /// Every commit should be rapidly reflected on your `IndexReader` and you
237    /// should not need to call `reload()` at all.
238    ///
239    /// This automatic reload can take 10s of milliseconds to kick in however,
240    /// and in unit tests it can be nice to deterministically force the reload
241    /// of searchers.
242    fn commit_and_reload(&mut self, writer: &mut SearchIndexWriter) -> Result<OpStamp, IndexError> {
243        debug!(
244            "RoomIndex: committing and reloading: uncommitted: {:?}, {:?}",
245            self.uncommitted_adds, self.uncommitted_removes
246        );
247        let last_commit_opstamp = self.commit(writer)?;
248        self.reader()?.reload()?;
249        Ok(last_commit_opstamp)
250    }
251
252    /// Search the [`RoomIndex`] for some query. Returns a list of results with
253    /// a maximum given length. If `pagination_offset` is set then the results
254    /// will start there, i.e.
255    ///
256    /// if `max_number_of_results = 3` and `pagination_offset = 10` (and there
257    /// are a surplus of results) then this will return results `11, 12, 13`
258    pub fn search(
259        &self,
260        query: &str,
261        max_number_of_results: usize,
262        pagination_offset: Option<usize>,
263    ) -> Result<SearchResult, IndexError> {
264        let query = self.query_parser.parse_query(query)?;
265        let searcher = self.reader()?.searcher();
266
267        let offset = pagination_offset.unwrap_or(0);
268
269        let top_docs = TopDocs::with_limit(max_number_of_results)
270            .and_offset(offset)
271            .order_by_fast_field::<DateTime>("date", Order::Desc);
272
273        let mut multi = MultiCollector::new();
274        let count_handle = multi.add_collector(Count);
275        let top_docs_handle = multi.add_collector(top_docs);
276
277        let mut results = searcher.search(&query, &multi)?;
278        let mut ret: Vec<(MilliSecondsSinceUnixEpoch, OwnedEventId)> = Vec::new();
279        let pk = self.schema.primary_key();
280
281        let result_count = count_handle.extract(&mut results);
282
283        for (date, doc_address) in top_docs_handle.extract(&mut results) {
284            // Timestamps are indexed from a `MilliSecondsSinceUnixEpoch`, so
285            // they are never negative.
286            let timestamp = MilliSecondsSinceUnixEpoch(UInt::new_saturating(
287                date.map_or(0, DateTime::into_timestamp_millis).cast_unsigned(),
288            ));
289            let retrieved_doc: TantivyDocument = searcher.doc(doc_address)?;
290            match retrieved_doc.get_first(pk).and_then(|maybe_value| maybe_value.as_str()) {
291                Some(value) => match OwnedEventId::try_from(value) {
292                    Ok(event_id) => ret.push((timestamp, event_id)),
293                    Err(err) => error!("error while parsing event_id from search result: {err:?}"),
294                },
295                _ => error!("unexpected value type while searching documents"),
296            }
297        }
298
299        Ok(SearchResult { total_count: result_count, events: ret })
300    }
301
302    fn events_to_be_removed(&self, event_id: &EventId) -> Result<Vec<OwnedEventId>, IndexError> {
303        Ok(self
304            .search(
305                format!(
306                    "{}:\"{event_id}\"",
307                    self.schema.get_field_name(self.schema.deletion_key())
308                )
309                .as_str(),
310                10000,
311                None,
312            )?
313            .events
314            .into_iter()
315            .map(|(_, id)| id)
316            .collect())
317    }
318
319    fn add(
320        &mut self,
321        writer: &mut SearchIndexWriter,
322        event: IndexableEvent,
323    ) -> Result<(), IndexError> {
324        if !self.contains(&event.event_id) {
325            writer.add(self.schema.make_doc(event.clone())?)?;
326        }
327        self.uncommitted_removes.remove(&event.event_id);
328        self.uncommitted_adds.insert(event.event_id, event.original_event_id);
329        Ok(())
330    }
331
332    fn remove(
333        &mut self,
334        writer: &mut SearchIndexWriter,
335        event_id: OwnedEventId,
336    ) -> Result<(), IndexError> {
337        let events = self.events_to_be_removed(&event_id)?;
338
339        writer.remove(&event_id);
340
341        // Committed documents matching the deletion key.
342        for event in events.into_iter() {
343            self.uncommitted_adds.remove(&event);
344            self.uncommitted_removes.insert(event);
345        }
346
347        // Uncommitted documents added in this same batch also get deleted by
348        // the term above, so reconcile them too. Otherwise `contains` would
349        // still report them as present and a subsequent re-add (e.g. from an
350        // edit) would be wrongly skipped, leaving the document deleted.
351        let uncommitted: Vec<_> = self
352            .uncommitted_adds
353            .iter()
354            .filter(|(_, deletion_key)| **deletion_key == event_id)
355            .map(|(primary_key, _)| primary_key.clone())
356            .collect();
357        for event in uncommitted {
358            self.uncommitted_adds.remove(&event);
359            self.uncommitted_removes.insert(event);
360        }
361
362        Ok(())
363    }
364
365    fn execute_impl(
366        &mut self,
367        writer: &mut SearchIndexWriter,
368        operation: &RoomIndexOperation,
369    ) -> Result<(), IndexError> {
370        debug!("INDEX: executing {operation:?}");
371        match operation.clone() {
372            RoomIndexOperation::Add(event) => {
373                self.add(writer, event)?;
374            }
375            RoomIndexOperation::Remove(event_id) => {
376                self.remove(writer, event_id)?;
377            }
378            RoomIndexOperation::Edit(remove_event_id, event) => {
379                self.remove(writer, remove_event_id)?;
380                self.add(writer, event)?;
381            }
382            RoomIndexOperation::Noop => {}
383        }
384        Ok(())
385    }
386
387    /// Execute [`RoomIndexOperation`] with retry
388    fn execute_with_retry(
389        &mut self,
390        writer: &mut SearchIndexWriter,
391        operation: &RoomIndexOperation,
392        retries: usize,
393    ) -> Result<(), IndexError> {
394        let mut num_tries = 0;
395
396        while let Err(err) = self.execute_impl(writer, operation) {
397            if num_tries == retries {
398                return Err(err);
399            }
400            match err {
401                // Retry
402                IndexError::TantivyError(_)
403                | IndexError::IndexSchemaError(_)
404                | IndexError::IndexWriteError(_)
405                | IndexError::IO(_) => {
406                    num_tries += 1;
407                }
408                IndexError::OpenDirectoryError(ref e) => match e {
409                    // Retry
410                    OpenDirectoryError::IoError { io_error: _, directory_path: _ } => {
411                        num_tries += 1;
412                    }
413                    // Bubble
414                    OpenDirectoryError::DoesNotExist(_)
415                    | OpenDirectoryError::FailedToCreateTempDir(_)
416                    | OpenDirectoryError::NotADirectory(_) => return Err(err),
417                },
418                // Bubble
419                IndexError::QueryParserError(_) => return Err(err),
420                // Ignore
421                IndexError::CannotIndexRedactedMessage
422                | IndexError::EmptyMessage
423                | IndexError::MessageTypeNotSupported => break,
424            }
425            debug!("Failed to execute operation in room index (try {num_tries}): {err}");
426        }
427        Ok(())
428    }
429
430    /// Execute [`RoomIndexOperation`]
431    ///
432    /// If an error occurs, retry 5 times if possible.
433    ///
434    /// This which will add/remove/edit an event in the index based on the
435    /// operation.
436    ///
437    /// Prefer [`RoomIndex::bulk_execute`] for multiple operations.
438    pub fn execute(&mut self, operation: RoomIndexOperation) -> Result<(), IndexError> {
439        let mut writer = self.writer()?;
440        self.execute_with_retry(&mut writer, &operation, 5)?;
441        self.commit_and_reload(&mut writer)?;
442        Ok(())
443    }
444
445    /// Bulk execute [`RoomIndexOperation`]s
446    ///
447    /// If an error occurs in the batch it retries 5 times if possible.
448    ///
449    /// This which will add/remove/edit an events in the index based on the
450    /// operations.
451    ///
452    /// Waits for background merge threads to prevent racing with subsequent
453    /// calls.
454    pub fn bulk_execute(&mut self, operations: Vec<RoomIndexOperation>) -> Result<(), IndexError> {
455        let mut writer = self.writer()?;
456        let mut operations = operations.into_iter();
457        let mut next_operation = operations.next();
458
459        while let Some(ref operation) = next_operation {
460            self.execute_with_retry(&mut writer, operation, 5)?;
461            next_operation = operations.next();
462        }
463
464        self.commit_and_reload(&mut writer)?;
465        writer.wait_merging_threads()?;
466
467        Ok(())
468    }
469
470    fn contains(&self, event_id: &EventId) -> bool {
471        let search_result = self.search(
472            format!("{}:\"{event_id}\"", self.schema.get_field_name(self.schema.primary_key()))
473                .as_str(),
474            1,
475            None,
476        );
477        match search_result {
478            Ok(results) => {
479                !self.uncommitted_removes.contains(event_id)
480                    && (!results.events.is_empty() || self.uncommitted_adds.contains_key(event_id))
481            }
482            Err(err) => {
483                warn!("Failed to check if event has been indexed, assuming it has: {err}");
484                true
485            }
486        }
487    }
488}
489
490#[cfg(test)]
491mod tests {
492    use std::{collections::HashSet, error::Error};
493
494    use matrix_sdk_test::event_factory::EventFactory;
495    use ruma::{
496        EventId, event_id,
497        events::{
498            AnySyncMessageLikeEvent,
499            room::message::{
500                MessageType, OriginalSyncRoomMessageEvent, Relation,
501                RoomMessageEventContentWithoutRelation,
502            },
503        },
504        room_id, user_id,
505    };
506
507    use super::{
508        IndexableEvent, MAX_MILLISECONDS, MilliSecondsSinceUnixEpoch, RoomIndex,
509        RoomIndexOperation, UInt, builder::RoomIndexBuilder,
510    };
511    use crate::error::IndexError;
512
513    /// Build an [`IndexableEvent`] from a text room message (tests only handle
514    /// text).
515    fn to_indexable(event: &OriginalSyncRoomMessageEvent) -> IndexableEvent {
516        let MessageType::Text(content) = &event.content.msgtype else {
517            panic!("test helper only supports text messages")
518        };
519        let original_event_id = match &event.content.relates_to {
520            Some(Relation::Replacement(replacement)) => replacement.event_id.clone(),
521            _ => event.event_id.clone(),
522        };
523
524        IndexableEvent::new(
525            event.event_id.clone(),
526            original_event_id,
527            event.sender.clone(),
528            Some(event.origin_server_ts),
529            content.body.clone(),
530        )
531    }
532
533    /// Helper function to add a regular message to the index
534    ///
535    /// # Panic
536    ///
537    /// Panics when event is not an [`OriginalSyncRoomMessageEvent`] with no
538    /// relations.
539    fn index_message(
540        index: &mut RoomIndex,
541        event: AnySyncMessageLikeEvent,
542    ) -> Result<(), IndexError> {
543        if let AnySyncMessageLikeEvent::RoomMessage(ev) = event
544            && let Some(ev) = ev.as_original()
545            && ev.content.relates_to.is_none()
546        {
547            return index.execute(RoomIndexOperation::Add(to_indexable(ev)));
548        }
549        panic!("Event was not a relationless OriginalSyncRoomMessageEvent.")
550    }
551
552    /// Helper function to remove events to the index
553    fn index_remove(index: &mut RoomIndex, event_id: &EventId) -> Result<(), IndexError> {
554        index.execute(RoomIndexOperation::Remove(event_id.to_owned()))
555    }
556
557    /// Helper function to edit events in index
558    ///
559    /// Edit event with `event_id` into new [`OriginalSyncRoomMessageEvent`]
560    fn index_edit(
561        index: &mut RoomIndex,
562        event_id: &EventId,
563        new: OriginalSyncRoomMessageEvent,
564    ) -> Result<(), IndexError> {
565        index.execute(RoomIndexOperation::Edit(event_id.to_owned(), to_indexable(&new)))
566    }
567
568    #[test]
569    fn test_add_event() {
570        let room_id = room_id!("!room_id:localhost");
571        let mut index = RoomIndexBuilder::new_in_memory(room_id).build();
572
573        let event = EventFactory::new()
574            .text_msg("event message")
575            .event_id(event_id!("$event_id:localhost"))
576            .room(room_id)
577            .sender(user_id!("@user_id:localhost"))
578            .into_any_sync_message_like_event();
579
580        index_message(&mut index, event).expect("failed to add event");
581    }
582
583    #[test]
584    fn test_encrypted_index_is_recreated_with_another_key() {
585        let tmpdir = tempfile::tempdir().unwrap();
586        let room_id = room_id!("!r");
587        let add = RoomIndexOperation::Add(IndexableEvent::new(
588            event_id!("$ev").to_owned(),
589            event_id!("$ev").to_owned(),
590            user_id!("@mnt_io:matrix.org").to_owned(),
591            None,
592            "body".to_owned(),
593        ));
594
595        let mut index = RoomIndexBuilder::new_on_disk(tmpdir.path().to_path_buf(), room_id)
596            .encrypted("key")
597            .build()
598            .unwrap();
599        index.execute(add).unwrap();
600        assert_eq!(index.search("body", 10, None).unwrap().events.len(), 1);
601        drop(index);
602
603        // The index can't be decrypted with another key, so it's
604        // recreated, empty.
605        let index = RoomIndexBuilder::new_on_disk(tmpdir.path().to_path_buf(), room_id)
606            .encrypted("another key")
607            .build()
608            .unwrap();
609        assert!(index.search("body", 10, None).unwrap().events.is_empty());
610    }
611
612    #[test]
613    fn test_commits_are_merged_into_a_single_segment() {
614        let tmpdir = tempfile::tempdir().unwrap();
615        let mut index = RoomIndexBuilder::new_on_disk(tmpdir.path().to_path_buf(), room_id!("!r"))
616            .unencrypted()
617            .build()
618            .unwrap();
619
620        for nth in 0..10 {
621            let event_id = EventId::parse(format!("$ev{nth}")).unwrap();
622            index
623                .bulk_execute(vec![RoomIndexOperation::Add(IndexableEvent::new(
624                    event_id.clone(),
625                    event_id,
626                    user_id!("@mnt_io:matrix.org").to_owned(),
627                    None,
628                    "body".to_owned(),
629                ))])
630                .unwrap();
631        }
632
633        assert_eq!(index.index.searchable_segment_ids().unwrap().len(), 1);
634        assert_eq!(index.search("body", 100, None).unwrap().events.len(), 10);
635    }
636
637    #[test]
638    fn test_add_event_with_no_timestamp() {
639        let mut index = RoomIndexBuilder::new_in_memory(room_id!("!r")).build();
640
641        index
642            .execute(RoomIndexOperation::Add(IndexableEvent::new(
643                event_id!("$ev").to_owned(),
644                event_id!("$ev").to_owned(),
645                user_id!("@mnt_io:matrix.org").to_owned(),
646                None,
647                "body".to_owned(),
648            )))
649            .expect("failed to add event");
650    }
651
652    #[test]
653    fn test_add_event_with_raw_malformed_timestamp_must_not_panic() {
654        let mut index = RoomIndexBuilder::new_in_memory(room_id!("!r")).build();
655
656        index
657            .execute(RoomIndexOperation::Add(IndexableEvent::new(
658                event_id!("$ev").to_owned(),
659                event_id!("$ev").to_owned(),
660                user_id!("@mnt_io:matrix.org").to_owned(),
661                Some(MilliSecondsSinceUnixEpoch(UInt::new(151393755000000).unwrap())),
662                "body".to_owned(),
663            )))
664            .expect("failed to add event");
665    }
666
667    #[test]
668    fn test_add_event_with_malformed_timestamp_must_not_panic() {
669        let room_id = room_id!("!r");
670        let mut index = RoomIndexBuilder::new_in_memory(room_id).build();
671
672        let body = "body".to_owned();
673        let event = EventFactory::new()
674            .server_ts(151393755000000)
675            .text_msg(body.clone())
676            .event_id(event_id!("$ev"))
677            .room(room_id)
678            .sender(user_id!("@mnt_io:matrix.org"))
679            .into_event();
680
681        index
682            .execute(RoomIndexOperation::Add(IndexableEvent::new(
683                event.event_id().unwrap().to_owned(),
684                event.event_id().unwrap().to_owned(),
685                event.sender().unwrap(),
686                event.timestamp(),
687                body,
688            )))
689            .expect("failed to add event");
690    }
691
692    #[test]
693    fn test_indexable_event_timestamp_is_capped() {
694        let event_id = event_id!("$ev").to_owned();
695        let sender = user_id!("@mnt_io:matrix.org").to_owned();
696        let body = "body".to_owned();
697
698        // Not capped.
699        let event = IndexableEvent::new(
700            event_id.clone(),
701            event_id.clone(),
702            sender.clone(),
703            Some(MilliSecondsSinceUnixEpoch(UInt::new(MAX_MILLISECONDS).unwrap())),
704            body.clone(),
705        );
706        assert_eq!(event.timestamp.unwrap().get(), UInt::new(MAX_MILLISECONDS).unwrap());
707
708        // Capped!
709        let event = IndexableEvent::new(
710            event_id.clone(),
711            event_id,
712            sender,
713            Some(MilliSecondsSinceUnixEpoch(UInt::new(MAX_MILLISECONDS + 42).unwrap())),
714            body,
715        );
716        assert_eq!(event.timestamp.unwrap().get(), UInt::new(MAX_MILLISECONDS).unwrap());
717    }
718
719    #[test]
720    fn test_search_populated_index() -> Result<(), Box<dyn Error>> {
721        let room_id = room_id!("!room_id:localhost");
722        let mut index = RoomIndexBuilder::new_in_memory(room_id).build();
723
724        let event_id_1 = event_id!("$event_id_1:localhost");
725        let event_id_2 = event_id!("$event_id_2:localhost");
726        let event_id_3 = event_id!("$event_id_3:localhost");
727        let user_id = user_id!("@user_id:localhost");
728        let f = EventFactory::new().room(room_id).sender(user_id);
729
730        index_message(
731            &mut index,
732            f.text_msg("This is a sentence")
733                .event_id(event_id_1)
734                .into_any_sync_message_like_event(),
735        )?;
736
737        index_message(
738            &mut index,
739            f.text_msg("All new words").event_id(event_id_2).into_any_sync_message_like_event(),
740        )?;
741
742        index_message(
743            &mut index,
744            f.text_msg("A similar sentence")
745                .event_id(event_id_3)
746                .into_any_sync_message_like_event(),
747        )?;
748
749        let result = index.search("sentence", 10, None).expect("search failed with: {result:?}");
750        let result: HashSet<_> = result.events.iter().map(|(_, id)| id).collect();
751
752        let true_value = [event_id_1.to_owned(), event_id_3.to_owned()];
753        let true_value: HashSet<_> = true_value.iter().collect();
754
755        assert_eq!(result, true_value, "search result not correct: {result:?}");
756
757        Ok(())
758    }
759
760    #[test]
761    fn test_search_empty_index() -> Result<(), Box<dyn Error>> {
762        let room_id = room_id!("!room_id:localhost");
763        let index = RoomIndexBuilder::new_in_memory(room_id).build();
764
765        let result = index.search("sentence", 10, None).expect("search failed with: {result:?}");
766
767        assert!(result.events.is_empty(), "search result not empty: {result:?}");
768
769        Ok(())
770    }
771
772    #[test]
773    fn test_index_contains_false() {
774        let room_id = room_id!("!room_id:localhost");
775        let index = RoomIndexBuilder::new_in_memory(room_id).build();
776
777        let event_id = event_id!("$event_id:localhost");
778
779        assert!(!index.contains(event_id), "Index should not contain event");
780    }
781
782    #[test]
783    fn test_index_contains_true() -> Result<(), Box<dyn Error>> {
784        let room_id = room_id!("!room_id:localhost");
785        let mut index = RoomIndexBuilder::new_in_memory(room_id).build();
786
787        let event_id = event_id!("$event_id:localhost");
788        let event = EventFactory::new()
789            .text_msg("This is a sentence")
790            .event_id(event_id)
791            .room(room_id)
792            .sender(user_id!("@user_id:localhost"))
793            .into_any_sync_message_like_event();
794
795        index_message(&mut index, event)?;
796
797        assert!(index.contains(event_id), "Index should contain event");
798
799        Ok(())
800    }
801
802    #[test]
803    fn test_index_add_idempotency() -> Result<(), Box<dyn Error>> {
804        let room_id = room_id!("!room_id:localhost");
805        let mut index = RoomIndexBuilder::new_in_memory(room_id).build();
806
807        let event_id = event_id!("$event_id:localhost");
808        let event = EventFactory::new()
809            .text_msg("This is a sentence")
810            .event_id(event_id)
811            .room(room_id)
812            .sender(user_id!("@user_id:localhost"))
813            .into_any_sync_message_like_event();
814
815        index_message(&mut index, event.clone())?;
816
817        assert!(index.contains(event_id), "Index should contain event");
818
819        // indexing again should do nothing
820        index_message(&mut index, event)?;
821
822        assert!(index.contains(event_id), "Index should still contain event");
823
824        let result = index.search("sentence", 10, None).expect("search failed with: {result:?}");
825
826        assert_eq!(result.events.len(), 1, "Index should have ignored second indexing");
827
828        Ok(())
829    }
830
831    #[test]
832    fn test_remove_event() -> Result<(), Box<dyn Error>> {
833        let room_id = room_id!("!room_id:localhost");
834        let mut index = RoomIndexBuilder::new_in_memory(room_id).build();
835
836        let event_id = event_id!("$event_id:localhost");
837        let user_id = user_id!("@user_id:localhost");
838        let f = EventFactory::new().room(room_id).sender(user_id);
839
840        let event =
841            f.text_msg("This is a sentence").event_id(event_id).into_any_sync_message_like_event();
842
843        index_message(&mut index, event)?;
844
845        assert!(index.contains(event_id), "Index should contain event");
846
847        index_remove(&mut index, event_id)?;
848
849        assert!(!index.contains(event_id), "Index should not contain event");
850
851        Ok(())
852    }
853
854    #[test]
855    fn test_edit_removes_old_and_adds_new_event() -> Result<(), Box<dyn Error>> {
856        let room_id = room_id!("!room_id:localhost");
857        let mut index = RoomIndexBuilder::new_in_memory(room_id).build();
858
859        let old_event_id = event_id!("$old_event_id:localhost");
860        let user_id = user_id!("@user_id:localhost");
861        let f = EventFactory::new().room(room_id).sender(user_id);
862
863        let old_event = f
864            .text_msg("This is a sentence")
865            .event_id(old_event_id)
866            .into_any_sync_message_like_event();
867
868        index_message(&mut index, old_event)?;
869
870        assert!(index.contains(old_event_id), "Index should contain event");
871
872        let new_event_id = event_id!("$new_event_id:localhost");
873        let edit = f
874            .text_msg("This is a brand new sentence!")
875            .edit(
876                old_event_id,
877                RoomMessageEventContentWithoutRelation::text_plain("This is a brand new sentence!"),
878            )
879            .event_id(new_event_id)
880            .into_original_sync_room_message_event();
881
882        index_edit(&mut index, old_event_id, edit)?;
883
884        assert!(!index.contains(old_event_id), "Index should not contain old event");
885        assert!(index.contains(new_event_id), "Index should contain edited event");
886
887        Ok(())
888    }
889
890    #[test]
891    fn test_bulk_add_then_edit_same_event_keeps_it_indexed() -> Result<(), Box<dyn Error>> {
892        let room_id = room_id!("!room_id:localhost");
893        let mut index = RoomIndexBuilder::new_in_memory(room_id).build();
894
895        let original_id = event_id!("$original:localhost");
896        let edit_id = event_id!("$edit:localhost");
897        let user_id = user_id!("@user_id:localhost");
898        let f = EventFactory::new().room(room_id).sender(user_id);
899
900        let edit = f
901            .text_msg("* brand new sentence")
902            .edit(
903                original_id,
904                RoomMessageEventContentWithoutRelation::text_plain("brand new sentence"),
905            )
906            .event_id(edit_id)
907            .into_original_sync_room_message_event();
908        let edit = to_indexable(&edit);
909
910        // An original and its edit arriving in the same batch produce an `Add`
911        // and an `Edit` of the same document. The `Edit`'s removal must not
912        // drop the document added earlier in the same uncommitted batch.
913        index.bulk_execute(vec![
914            RoomIndexOperation::Add(edit.clone()),
915            RoomIndexOperation::Edit(original_id.to_owned(), edit),
916        ])?;
917
918        assert!(index.contains(edit_id), "Edited document should be indexed");
919
920        let result = index.search("sentence", 10, None)?;
921        assert_eq!(result.total_count, 1, "Search should return the total count of results it got");
922        assert_eq!(
923            result.events.len(),
924            1,
925            "Search should find the edited document, got {result:?}"
926        );
927        assert_eq!(result.events[0].1, edit_id, "unexpected event id: {result:?}");
928
929        Ok(())
930    }
931}
matrix_sdk_search/index/mod.rs

matrix_sdk_search/index/
mod.rs
