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

matrix_sdk_search/index/
mod.rs
