use std::{collections::hash_map::HashMap, path::PathBuf, sync::Arc};
use futures_util::future::join_all;
use matrix_sdk_base::{
check_validity_of_replacement_events, deserialized_responses::TimelineEvent,
};
use matrix_sdk_search::{
error::IndexError,
index::{IndexableEvent, RoomIndex, RoomIndexOperation, builder::RoomIndexBuilder},
};
use ruma::{
EventId, MilliSecondsSinceUnixEpoch, OwnedEventId, OwnedRoomId, RoomId,
events::{
AnySyncMessageLikeEvent, AnySyncTimelineEvent,
poll::{
start::SyncPollStartEvent,
unstable_start::{SyncUnstablePollStartEvent, UnstablePollStartEventContent},
},
room::{
message::{MessageType, OriginalSyncRoomMessageEvent, Relation, SyncRoomMessageEvent},
redaction::SyncRoomRedactionEvent,
},
sticker::SyncStickerEvent,
},
room_version_rules::RedactionRules,
};
use tokio::sync::{Mutex, MutexGuard};
use tracing::{debug, warn};
use crate::event_cache::RoomEventCache;
type Password = String;
#[derive(Clone, Debug)]
pub enum SearchIndexStoreKind {
UnencryptedDirectory(PathBuf),
EncryptedDirectory(PathBuf, Password),
InMemory,
}
#[derive(Clone, Debug)]
pub struct SearchIndex {
room_indexes: Arc<Mutex<HashMap<OwnedRoomId, RoomIndex>>>,
search_index_store_kind: SearchIndexStoreKind,
}
impl SearchIndex {
pub fn new(
room_indexes: Arc<Mutex<HashMap<OwnedRoomId, RoomIndex>>>,
search_index_store_kind: SearchIndexStoreKind,
) -> Self {
Self { room_indexes, search_index_store_kind }
}
pub async fn lock(&self) -> SearchIndexGuard<'_> {
SearchIndexGuard {
index_map: self.room_indexes.lock().await,
search_index_store_kind: &self.search_index_store_kind,
}
}
}
#[derive(Debug)]
pub struct SearchIndexGuard<'a> {
index_map: MutexGuard<'a, HashMap<OwnedRoomId, RoomIndex>>,
search_index_store_kind: &'a SearchIndexStoreKind,
}
impl SearchIndexGuard<'_> {
fn create_index(&self, room_id: &RoomId) -> Result<RoomIndex, IndexError> {
let index = match self.search_index_store_kind {
SearchIndexStoreKind::UnencryptedDirectory(path) => {
RoomIndexBuilder::new_on_disk(path.to_path_buf(), room_id).unencrypted().build()?
}
SearchIndexStoreKind::EncryptedDirectory(path, password) => {
RoomIndexBuilder::new_on_disk(path.to_path_buf(), room_id)
.encrypted(password)
.build()?
}
SearchIndexStoreKind::InMemory => RoomIndexBuilder::new_in_memory(room_id).build(),
};
Ok(index)
}
pub(crate) fn execute(
&mut self,
operation: RoomIndexOperation,
room_id: &RoomId,
) -> Result<(), IndexError> {
if !self.index_map.contains_key(room_id) {
let index = self.create_index(room_id)?;
self.index_map.insert(room_id.to_owned(), index);
}
let index = self.index_map.get_mut(room_id).expect("index should exist");
index.execute(operation)
}
pub(crate) fn bulk_execute(
&mut self,
operations: Vec<RoomIndexOperation>,
room_id: &RoomId,
) -> Result<(), IndexError> {
if !self.index_map.contains_key(room_id) {
let index = self.create_index(room_id)?;
self.index_map.insert(room_id.to_owned(), index);
}
let index = self.index_map.get_mut(room_id).expect("index should exist");
index.bulk_execute(operations)
}
pub(crate) fn search(
&mut self,
query: &str,
max_number_of_results: usize,
pagination_offset: Option<usize>,
room_id: &RoomId,
) -> Result<Vec<(f32, OwnedEventId)>, IndexError> {
if !self.index_map.contains_key(room_id) {
let index = self.create_index(room_id)?;
self.index_map.insert(room_id.to_owned(), index);
}
let index = self.index_map.get_mut(room_id).expect("index should exist");
index.search(query, max_number_of_results, pagination_offset)
}
pub async fn handle_timeline_event(
&mut self,
event: TimelineEvent,
room_cache: &RoomEventCache,
room_id: &RoomId,
redaction_rules: &RedactionRules,
) -> Result<(), IndexError> {
if let Some(index_operation) =
parse_timeline_event(room_cache, event, redaction_rules).await
{
self.execute(index_operation, room_id)
} else {
Ok(())
}
}
pub async fn bulk_handle_timeline_event<T>(
&mut self,
events: T,
room_cache: &RoomEventCache,
room_id: &RoomId,
redaction_rules: &RedactionRules,
) -> Result<(), IndexError>
where
T: Iterator<Item = TimelineEvent>,
{
let futures = events.map(|ev| parse_timeline_event(room_cache, ev, redaction_rules));
let operations: Vec<_> = join_all(futures).await.into_iter().flatten().collect();
self.bulk_execute(operations, room_id)
}
}
async fn get_most_recent_edit(
cache: &RoomEventCache,
original: &EventId,
) -> Option<OriginalSyncRoomMessageEvent> {
use ruma::events::{AnySyncTimelineEvent, relation::RelationType};
let Ok(Some((original_ev, related))) =
cache.find_event_with_relations(original, Some(vec![RelationType::Replacement])).await
else {
debug!("Couldn't find relations for {}", original);
return None;
};
let latest = related
.iter()
.rev()
.find(|edit| {
check_validity_of_replacement_events(
original_ev.raw(),
original_ev.encryption_info().map(|info| &**info),
edit.raw(),
edit.encryption_info().map(|info| &**info),
)
.is_ok()
})
.unwrap_or(&original_ev);
match latest.raw().deserialize() {
Ok(AnySyncTimelineEvent::MessageLike(AnySyncMessageLikeEvent::RoomMessage(latest))) => {
latest.as_original().cloned()
}
_ => None,
}
}
fn media_body(filename: &str, caption: Option<&str>) -> String {
match caption {
Some(caption) => format!("{filename} {caption}"),
None => filename.to_owned(),
}
}
fn room_message_body(msgtype: &MessageType) -> Option<String> {
match msgtype {
MessageType::Text(content) => Some(content.body.clone()),
MessageType::Emote(content) => Some(content.body.clone()),
MessageType::Notice(content) => Some(content.body.clone()),
MessageType::ServerNotice(content) => Some(content.body.clone()),
MessageType::Location(content) => Some(content.body.clone()),
MessageType::Image(content) => Some(media_body(content.filename(), content.caption())),
MessageType::Video(content) => Some(media_body(content.filename(), content.caption())),
MessageType::Audio(content) => Some(media_body(content.filename(), content.caption())),
MessageType::File(content) => Some(media_body(content.filename(), content.caption())),
_ => None,
}
}
fn indexable_from_room_message(
event: &OriginalSyncRoomMessageEvent,
timestamp: Option<MilliSecondsSinceUnixEpoch>,
) -> Option<IndexableEvent> {
let body = room_message_body(&event.content.msgtype)?;
let original_event_id = match &event.content.relates_to {
Some(Relation::Replacement(replacement)) => replacement.event_id.clone(),
_ => event.event_id.clone(),
};
Some(IndexableEvent::new(
event.event_id.clone(),
original_event_id,
event.sender.clone(),
timestamp,
body,
))
}
async fn handle_possible_edit(
event: &OriginalSyncRoomMessageEvent,
timestamp: Option<MilliSecondsSinceUnixEpoch>,
cache: &RoomEventCache,
) -> Option<RoomIndexOperation> {
if let Some(Relation::Replacement(replacement_data)) = &event.content.relates_to {
if let Some(recent) = get_most_recent_edit(cache, &replacement_data.event_id).await {
return Some(
indexable_from_room_message(&recent, timestamp).map_or(
RoomIndexOperation::Noop,
|indexable| {
RoomIndexOperation::Edit(replacement_data.event_id.clone(), indexable)
},
),
);
} else {
return Some(RoomIndexOperation::Noop);
}
}
None
}
async fn handle_room_message(
event: SyncRoomMessageEvent,
timestamp: Option<MilliSecondsSinceUnixEpoch>,
cache: &RoomEventCache,
) -> Option<RoomIndexOperation> {
if let Some(event) = event.as_original() {
return handle_possible_edit(event, timestamp, cache).await.or(get_most_recent_edit(
cache,
&event.event_id,
)
.await
.and_then(|recent| {
indexable_from_room_message(&recent, timestamp).map(RoomIndexOperation::Add)
}));
}
None
}
async fn handle_room_redaction(
event: SyncRoomRedactionEvent,
timestamp: Option<MilliSecondsSinceUnixEpoch>,
cache: &RoomEventCache,
rules: &RedactionRules,
) -> Option<RoomIndexOperation> {
let redacted_event_id = event.redacts(rules)?;
if let Ok(Some(redacted_event)) = cache.find_event(redacted_event_id).await
&& let Ok(AnySyncTimelineEvent::MessageLike(AnySyncMessageLikeEvent::RoomMessage(
redacted_event,
))) = redacted_event.raw().deserialize()
&& let Some(redacted_event) = redacted_event.as_original()
&& let Some(operation) = handle_possible_edit(redacted_event, timestamp, cache).await
{
return Some(operation);
}
Some(RoomIndexOperation::Remove(redacted_event_id.to_owned()))
}
fn handle_sticker(
event: SyncStickerEvent,
timestamp: Option<MilliSecondsSinceUnixEpoch>,
) -> Option<RoomIndexOperation> {
let event = event.as_original()?;
Some(RoomIndexOperation::Add(IndexableEvent::new(
event.event_id.clone(),
event.event_id.clone(),
event.sender.clone(),
timestamp,
event.content.body.clone(),
)))
}
fn handle_unstable_poll_start(
event: SyncUnstablePollStartEvent,
timestamp: Option<MilliSecondsSinceUnixEpoch>,
) -> Option<RoomIndexOperation> {
let event = event.as_original()?;
let UnstablePollStartEventContent::New(content) = &event.content else {
return None;
};
let block = &content.poll_start;
let mut body = block.question.text.clone();
for answer in block.answers.iter() {
body.push(' ');
body.push_str(&answer.text);
}
Some(RoomIndexOperation::Add(IndexableEvent::new(
event.event_id.clone(),
event.event_id.clone(),
event.sender.clone(),
timestamp,
body,
)))
}
fn handle_poll_start(
event: SyncPollStartEvent,
timestamp: Option<MilliSecondsSinceUnixEpoch>,
) -> Option<RoomIndexOperation> {
let event = event.as_original()?;
if let Some(Relation::Replacement(_)) = &event.content.relates_to {
return None;
}
let block = &event.content.poll;
let mut body = block.question.text.find_plain()?.to_owned();
for answer in block.answers.iter() {
if let Some(text) = answer.text.find_plain() {
body.push(' ');
body.push_str(text);
}
}
Some(RoomIndexOperation::Add(IndexableEvent::new(
event.event_id.clone(),
event.event_id.clone(),
event.sender.clone(),
timestamp,
body,
)))
}
async fn parse_timeline_event(
cache: &RoomEventCache,
event: TimelineEvent,
redaction_rules: &RedactionRules,
) -> Option<RoomIndexOperation> {
use ruma::events::AnySyncTimelineEvent;
if event.kind.is_utd() {
return None;
}
let timestamp = event.timestamp();
match event.raw().deserialize() {
Ok(event) => match event {
AnySyncTimelineEvent::MessageLike(event) => match event {
AnySyncMessageLikeEvent::RoomMessage(event) => {
handle_room_message(event, timestamp, cache).await
}
AnySyncMessageLikeEvent::RoomRedaction(event) => {
handle_room_redaction(event, timestamp, cache, redaction_rules).await
}
AnySyncMessageLikeEvent::Sticker(event) => handle_sticker(event, timestamp),
AnySyncMessageLikeEvent::PollStart(event) => handle_poll_start(event, timestamp),
AnySyncMessageLikeEvent::UnstablePollStart(event) => {
handle_unstable_poll_start(event, timestamp)
}
_ => None,
},
AnySyncTimelineEvent::State(_) => None,
},
Err(e) => {
warn!("failed to parse event: {e:?}");
None
}
}
}
#[cfg(test)]
mod tests {
use matrix_sdk_test::{JoinedRoomBuilder, async_test, event_factory::EventFactory};
use ruma::{
event_id, events::room::message::RoomMessageEventContentWithoutRelation, room_id, user_id,
};
use crate::test_utils::mocks::MatrixMockServer;
#[cfg(feature = "experimental-search")]
#[async_test]
async fn test_sync_message_is_indexed() {
let mock_server = MatrixMockServer::new().await;
let client = mock_server.client_builder().build().await;
client.event_cache().subscribe().unwrap();
let room_id = room_id!("!room_id:localhost");
let event_id = event_id!("$event_id:localost");
let user_id = user_id!("@user_id:localost");
let event_factory = EventFactory::new();
let room = mock_server
.sync_room(
&client,
JoinedRoomBuilder::new(room_id).add_timeline_bulk(vec![
event_factory
.text_msg("this is a sentence")
.event_id(event_id)
.sender(user_id)
.into_raw_sync(),
]),
)
.await;
let response = room.search("this", 5, None).await.expect("search should have 1 result");
assert_eq!(response.len(), 1, "unexpected numbers of responses: {response:?}");
assert_eq!(response[0].1, event_id, "event id doesn't match: {response:?}");
}
#[cfg(feature = "experimental-search")]
#[async_test]
async fn test_sync_media_message_is_indexed() {
use ruma::owned_mxc_uri;
let mock_server = MatrixMockServer::new().await;
let client = mock_server.client_builder().build().await;
client.event_cache().subscribe().unwrap();
let room_id = room_id!("!room_id:localhost");
let image_id = event_id!("$image_id:localhost");
let file_id = event_id!("$file_id:localhost");
let user_id = user_id!("@user_id:localhost");
let f = EventFactory::new();
let room = mock_server
.sync_room(
&client,
JoinedRoomBuilder::new(room_id).add_timeline_bulk(vec![
f.image("holiday_beach.jpg".to_owned(), owned_mxc_uri!("mxc://localhost/1"))
.caption(Some("sunset over the ocean".to_owned()), None)
.event_id(image_id)
.sender(user_id)
.into_raw_sync(),
f.image("quarterly_report.pdf".to_owned(), owned_mxc_uri!("mxc://localhost/2"))
.event_id(file_id)
.sender(user_id)
.into_raw_sync(),
]),
)
.await;
let response = room.search("sunset", 5, None).await.unwrap();
assert_eq!(response.len(), 1, "unexpected results for caption search: {response:?}");
assert_eq!(response[0].1, image_id, "event id doesn't match: {response:?}");
let response = room.search("holiday_beach", 5, None).await.unwrap();
assert_eq!(response.len(), 1, "unexpected results for filename search: {response:?}");
assert_eq!(response[0].1, image_id, "event id doesn't match: {response:?}");
let response = room.search("quarterly_report", 5, None).await.unwrap();
assert_eq!(response.len(), 1, "unexpected results for filename search: {response:?}");
assert_eq!(response[0].1, file_id, "event id doesn't match: {response:?}");
}
#[cfg(feature = "experimental-search")]
#[async_test]
async fn test_sync_sticker_and_poll_are_indexed() {
use ruma::{events::room::ImageInfo, owned_mxc_uri};
let mock_server = MatrixMockServer::new().await;
let client = mock_server.client_builder().build().await;
client.event_cache().subscribe().unwrap();
let room_id = room_id!("!room_id:localhost");
let sticker_id = event_id!("$sticker_id:localhost");
let poll_id = event_id!("$poll_id:localhost");
let user_id = user_id!("@user_id:localhost");
let f = EventFactory::new().room(room_id).sender(user_id);
let room = mock_server
.sync_room(
&client,
JoinedRoomBuilder::new(room_id).add_timeline_bulk(vec![
f.sticker(
"a waving cat",
ImageInfo::new(),
owned_mxc_uri!("mxc://localhost/1"),
)
.event_id(sticker_id)
.into_raw_sync(),
f.poll_start("fallback", "favourite cheese?", vec!["comté", "gruyère"])
.event_id(poll_id)
.into_raw_sync(),
]),
)
.await;
let response = room.search("waving", 5, None).await.unwrap();
assert_eq!(response.len(), 1, "unexpected results for sticker search: {response:?}");
assert_eq!(response[0].1, sticker_id, "event id doesn't match: {response:?}");
let response = room.search("cheese", 5, None).await.unwrap();
assert_eq!(response.len(), 1, "unexpected results for poll question search: {response:?}");
assert_eq!(response[0].1, poll_id, "event id doesn't match: {response:?}");
let response = room.search("gruyère", 5, None).await.unwrap();
assert_eq!(response.len(), 1, "unexpected results for poll answer search: {response:?}");
assert_eq!(response[0].1, poll_id, "event id doesn't match: {response:?}");
}
#[cfg(feature = "experimental-search")]
#[async_test]
async fn test_sync_stable_poll_is_indexed() {
use ruma::events::{
message::TextContentBlock,
poll::start::{PollAnswer, PollAnswers, PollContentBlock, PollStartEventContent},
};
let mock_server = MatrixMockServer::new().await;
let client = mock_server.client_builder().build().await;
client.event_cache().subscribe().unwrap();
let room_id = room_id!("!room_id:localhost");
let poll_id = event_id!("$stable_poll_id:localhost");
let user_id = user_id!("@user_id:localhost");
let answers: PollAnswers = vec![
PollAnswer::new("0".to_owned(), TextContentBlock::plain("comté")),
PollAnswer::new("1".to_owned(), TextContentBlock::plain("gruyère")),
]
.try_into()
.unwrap();
let poll = PollContentBlock::new(TextContentBlock::plain("favourite cheese?"), answers);
let content = PollStartEventContent::new(TextContentBlock::plain("fallback"), poll);
let f = EventFactory::new().room(room_id).sender(user_id);
let room = mock_server
.sync_room(
&client,
JoinedRoomBuilder::new(room_id)
.add_timeline_bulk(vec![f.event(content).event_id(poll_id).into_raw_sync()]),
)
.await;
let response = room.search("cheese", 5, None).await.unwrap();
assert_eq!(response.len(), 1, "unexpected results for poll question search: {response:?}");
assert_eq!(response[0].1, poll_id, "event id doesn't match: {response:?}");
let response = room.search("gruyère", 5, None).await.unwrap();
assert_eq!(response.len(), 1, "unexpected results for poll answer search: {response:?}");
assert_eq!(response[0].1, poll_id, "event id doesn't match: {response:?}");
}
#[cfg(feature = "experimental-search")]
#[async_test]
async fn test_search_index_edit_ordering() {
let room_id = room_id!("!room_id:localhost");
let dummy_id = event_id!("$dummy");
let edit1_id = event_id!("$edit1");
let edit2_id = event_id!("$edit2");
let edit3_id = event_id!("$edit3");
let original_id = event_id!("$original");
let server = MatrixMockServer::new().await;
let client = server.client_builder().build().await;
let event_cache = client.event_cache();
event_cache.subscribe().unwrap();
let room = server.sync_joined_room(&client, room_id).await;
let f = EventFactory::new().room(room_id).sender(user_id!("@user_id:localhost"));
let dummy = f.text_msg("dummy").event_id(dummy_id);
let original = f.text_msg("This is a message").event_id(original_id);
let edit1 = f
.text_msg("* A new message")
.edit(original_id, RoomMessageEventContentWithoutRelation::text_plain("A new message"))
.event_id(edit1_id);
let edit2 = f
.text_msg("* An even newer message")
.edit(
original_id,
RoomMessageEventContentWithoutRelation::text_plain("An even newer message"),
)
.event_id(edit2_id);
let edit3 = f
.text_msg("* The newest message")
.edit(
original_id,
RoomMessageEventContentWithoutRelation::text_plain("The newest message"),
)
.event_id(edit3_id);
server
.sync_room(
&client,
JoinedRoomBuilder::new(room_id)
.add_timeline_event(dummy)
.add_timeline_event(edit1)
.add_timeline_event(edit2),
)
.await;
let results = room.search("message", 3, None).await.unwrap();
assert_eq!(results.len(), 0, "Search should return 0 results, got {results:?}");
server
.sync_room(&client, JoinedRoomBuilder::new(room_id).add_timeline_event(original))
.await;
let results = room.search("message", 3, None).await.unwrap();
assert_eq!(results.len(), 1, "Search should return 1 result, got {results:?}");
assert_eq!(
results[0].1, edit2_id,
"Search should return latest edit, got {:?}",
results[0].1
);
server.sync_room(&client, JoinedRoomBuilder::new(room_id).add_timeline_event(edit3)).await;
let results = room.search("message", 3, None).await.unwrap();
assert_eq!(results.len(), 1, "Search should return 1 result, got {results:?}");
assert_eq!(
results[0].1, edit3_id,
"Search should return latest edit, got {:?}",
results[0].1
);
}
#[cfg(feature = "experimental-search")]
#[async_test]
async fn test_search_index_ignores_cross_sender_edit() {
let room_id = room_id!("!room_id:localhost");
let original_id = event_id!("$original");
let edit_id = event_id!("$edit");
let server = MatrixMockServer::new().await;
let client = server.client_builder().build().await;
let event_cache = client.event_cache();
event_cache.subscribe().unwrap();
let room = server.sync_joined_room(&client, room_id).await;
let f = EventFactory::new().room(room_id);
let original =
f.text_msg("original alpha").sender(user_id!("@alice:localhost")).event_id(original_id);
let malicious_edit = f
.text_msg("* malicious beta")
.edit(original_id, RoomMessageEventContentWithoutRelation::text_plain("malicious beta"))
.sender(user_id!("@bob:localhost"))
.event_id(edit_id);
server
.sync_room(&client, JoinedRoomBuilder::new(room_id).add_timeline_event(original))
.await;
let results = room.search("alpha", 3, None).await.unwrap();
assert_eq!(results.len(), 1, "Original should be indexed, got {results:?}");
assert_eq!(results[0].1, original_id, "unexpected event id: {results:?}");
server
.sync_room(&client, JoinedRoomBuilder::new(room_id).add_timeline_event(malicious_edit))
.await;
let results = room.search("beta", 3, None).await.unwrap();
assert_eq!(results.len(), 0, "Cross-sender edit should be ignored, got {results:?}");
let results = room.search("alpha", 3, None).await.unwrap();
assert_eq!(results.len(), 1, "Original should stay indexed, got {results:?}");
assert_eq!(results[0].1, original_id, "unexpected event id: {results:?}");
}
}