1use std::{
16 collections::BTreeMap,
17 ops::Deref,
18 sync::{Arc, Mutex},
19 time::Duration,
20};
21
22use futures_util::{StreamExt as _, pin_mut};
23use itertools::Itertools;
24use matrix_sdk::{
25 Client, ClientBuildError, SlidingSyncList, SlidingSyncMode,
26 room::{PushContext, Room},
27};
28use matrix_sdk_base::{RoomState, StoreError, deserialized_responses::TimelineEvent};
29use matrix_sdk_common::{cross_process_lock::CrossProcessLockConfig, timeout::timeout};
30use ruma::{
31 EventId, OwnedEventId, OwnedRoomId, RoomId, UserId,
32 api::client::sync::sync_events::v5 as http,
33 assign,
34 events::{
35 AnyMessageLikeEventContent, AnyStateEvent, AnyStateEventContentChange,
36 AnySyncMessageLikeEvent, AnySyncTimelineEvent, StateEventContentChange, StateEventType,
37 TimelineEventType,
38 room::{
39 encrypted::OriginalSyncRoomEncryptedEvent,
40 join_rules::JoinRule,
41 member::{MembershipState, StrippedRoomMemberEvent},
42 message::{Relation, SyncRoomMessageEvent},
43 },
44 },
45 html::RemoveReplyFallback,
46 push::Action,
47 serde::Raw,
48 time::Instant,
49 uint,
50};
51use thiserror::Error;
52use tokio::sync::Mutex as AsyncMutex;
53use tracing::{debug, info, instrument, trace, warn};
54
55use crate::{
56 DEFAULT_SANITIZER_MODE,
57 encryption_sync_service::{EncryptionSyncPermit, EncryptionSyncService},
58 sync_service::SyncService,
59};
60
61#[derive(Clone)]
63pub enum NotificationProcessSetup {
64 MultipleProcesses,
73
74 SingleProcess { sync_service: Arc<SyncService> },
82}
83
84#[derive(Clone, Copy, Debug, PartialEq, Eq)]
87pub struct NotificationClientTimeouts {
88 pub sync_poll_timeout: Duration,
92
93 pub sync_network_timeout: Duration,
97
98 pub decryption_deadline: Duration,
118
119 pub encryption_sync_poll_timeout: Duration,
127
128 pub encryption_sync_network_timeout: Duration,
132}
133
134impl Default for NotificationClientTimeouts {
135 fn default() -> Self {
138 let decryption_deadline = Duration::from_secs(6);
139
140 Self {
141 sync_poll_timeout: Duration::from_secs(1),
142 sync_network_timeout: Duration::from_secs(3),
143 decryption_deadline,
144 encryption_sync_poll_timeout: decryption_deadline
148 / NotificationClient::MIN_DECRYPTION_ITERATIONS as u32,
149 encryption_sync_network_timeout: Duration::from_secs(4),
150 }
151 }
152}
153
154pub struct NotificationClient {
160 client: Client,
162
163 parent_client: Client,
165
166 process_setup: NotificationProcessSetup,
168
169 notification_sync_mutex: AsyncMutex<()>,
177
178 encryption_sync_mutex: AsyncMutex<()>,
183
184 timeouts: NotificationClientTimeouts,
187}
188
189impl NotificationClient {
190 const CONNECTION_ID: &'static str = "notifications";
191 const LOCK_ID: &'static str = "notifications";
192
193 const MIN_DECRYPTION_ITERATIONS: usize = 2;
201
202 pub async fn new(
204 parent_client: Client,
205 process_setup: NotificationProcessSetup,
206 ) -> Result<Self, Error> {
207 let cross_process_store_config = match process_setup {
210 NotificationProcessSetup::MultipleProcesses => {
211 CrossProcessLockConfig::multi_process(Self::LOCK_ID)
212 }
213 NotificationProcessSetup::SingleProcess { .. } => CrossProcessLockConfig::SingleProcess,
214 };
215 let client = parent_client.notification_client(cross_process_store_config).await?;
216
217 Ok(NotificationClient {
218 client,
219 parent_client,
220 notification_sync_mutex: AsyncMutex::new(()),
221 encryption_sync_mutex: AsyncMutex::new(()),
222 process_setup,
223 timeouts: NotificationClientTimeouts::default(),
224 })
225 }
226
227 pub fn with_timeouts(mut self, timeouts: NotificationClientTimeouts) -> Self {
229 self.timeouts = timeouts;
230 self
231 }
232
233 pub fn timeouts(&self) -> &NotificationClientTimeouts {
235 &self.timeouts
236 }
237
238 pub fn get_room(&self, room_id: &RoomId) -> Option<Room> {
242 self.client.get_room(room_id)
243 }
244
245 #[instrument(skip(self))]
254 pub async fn get_notification(
255 &self,
256 room_id: &RoomId,
257 event_id: &EventId,
258 ) -> Result<NotificationStatus, Error> {
259 let status = self.get_notification_with_sliding_sync(room_id, event_id).await?;
260 match status {
261 NotificationStatus::Event(..)
262 | NotificationStatus::EventFilteredOut
263 | NotificationStatus::EventRedacted => Ok(status),
264 NotificationStatus::EventNotFound => {
265 self.get_notification_with_context(room_id, event_id).await
266 }
267 }
268 }
269
270 pub async fn get_notifications(
284 &self,
285 requests: &[NotificationItemsRequest],
286 ) -> Result<BatchNotificationFetchingResult, Error> {
287 let mut notifications = self.get_notifications_with_sliding_sync(requests).await?;
288
289 for request in requests {
290 for event_id in &request.event_ids {
291 match notifications.get_mut(event_id) {
292 Some(Ok(NotificationStatus::EventNotFound)) | None => {
295 notifications.insert(
296 event_id.to_owned(),
297 self.get_notification_with_context(&request.room_id, event_id).await,
298 );
299 }
300
301 _ => {}
302 }
303 }
304 }
305
306 Ok(notifications)
307 }
308
309 #[instrument(skip_all)]
320 async fn retry_decryption(
321 &self,
322 room: &Room,
323 raw_event: &Raw<AnySyncTimelineEvent>,
324 ) -> Result<Option<TimelineEvent>, Error> {
325 let event: AnySyncTimelineEvent =
326 raw_event.deserialize().map_err(|_| Error::InvalidRumaEvent)?;
327
328 if !is_event_encrypted(event.event_type()) {
329 return Ok(None);
330 }
331
332 let _guard = self.encryption_sync_mutex.lock().await;
334
335 let push_ctx = room.push_context().await?;
336
337 let sync_permit_guard = match &self.process_setup {
338 NotificationProcessSetup::MultipleProcesses => {
339 let sync_permit = Arc::new(AsyncMutex::new(EncryptionSyncPermit::new()));
344 sync_permit.lock_owned().await
345 }
346
347 NotificationProcessSetup::SingleProcess { sync_service } => {
348 if let Some(permit_guard) = sync_service.try_get_encryption_sync_permit() {
349 permit_guard
350 } else {
351 debug!("Encryption sync running in background, waiting for the room key");
358 return self.wait_for_room_key(room, raw_event, push_ctx.as_ref()).await;
359 }
360 }
361 };
362
363 let encryption_sync = match EncryptionSyncService::new(
371 self.client.clone(),
372 Some((
373 self.timeouts.encryption_sync_poll_timeout,
374 self.timeouts.encryption_sync_network_timeout,
375 )),
376 )
377 .await
378 {
379 Ok(encryption_sync) => encryption_sync,
380 Err(err) => {
381 warn!("Encryption sync build error: {err:#}");
382 return Ok(None);
383 }
384 };
385
386 let deadline = Instant::now() + self.timeouts.decryption_deadline;
387 let iterations = encryption_sync.run_iterations(sync_permit_guard);
388 pin_mut!(iterations);
389
390 let mut num_iterations = 0;
391
392 loop {
393 let sync_ended = match iterations.next().await {
394 Some(Ok(())) => {
395 num_iterations += 1;
396 false
397 }
398
399 Some(Err(err)) => {
400 warn!("Encryption sync error, attempting to decrypt one last time: {err:#}");
404 true
405 }
406
407 None => {
408 trace!("Encryption sync ended, attempting to decrypt one last time");
412 true
413 }
414 };
415
416 match try_decrypt(room, raw_event, push_ctx.as_ref()).await {
417 Ok(DecryptionAttempt::Decrypted(new_event)) => {
418 trace!("Encryption sync managed to decrypt the event.");
419 return Ok(Some(new_event));
420 }
421 Ok(DecryptionAttempt::MissingRoomKey) => {
422 if sync_ended {
423 debug!("Encryption sync ended and the room key is still missing.");
424 return Ok(None);
425 }
426 if num_iterations >= Self::MIN_DECRYPTION_ITERATIONS
427 && Instant::now() >= deadline
428 {
429 debug!("Deadline reached while waiting for the room key, giving up.");
430 return Ok(None);
431 }
432 trace!("Still missing the room key, running another encryption sync iteration");
433 }
434 Ok(DecryptionAttempt::Unrecoverable) => return Ok(None),
435 Err(err) => {
436 trace!("Encryption sync failed to decrypt the event: {err}");
437 return Ok(None);
438 }
439 }
440 }
441 }
442
443 async fn wait_for_room_key(
454 &self,
455 room: &Room,
456 raw_event: &Raw<AnySyncTimelineEvent>,
457 push_ctx: Option<&PushContext>,
458 ) -> Result<Option<TimelineEvent>, Error> {
459 let Some(room_keys) = self.parent_client.encryption().room_keys_received_stream().await
465 else {
466 return Ok(match try_decrypt(room, raw_event, push_ctx).await? {
469 DecryptionAttempt::Decrypted(event) => Some(event),
470 DecryptionAttempt::MissingRoomKey | DecryptionAttempt::Unrecoverable => None,
471 });
472 };
473 pin_mut!(room_keys);
474
475 let deadline = Instant::now() + self.timeouts.decryption_deadline;
476
477 loop {
478 match try_decrypt(room, raw_event, push_ctx).await? {
479 DecryptionAttempt::Decrypted(event) => {
480 trace!("Waiting succeeded and event could be decrypted!");
481 return Ok(Some(event));
482 }
483 DecryptionAttempt::Unrecoverable => return Ok(None),
484 DecryptionAttempt::MissingRoomKey => {}
485 }
486
487 loop {
489 let remaining = deadline.saturating_duration_since(Instant::now());
490 if remaining.is_zero() {
491 debug!("Timeout waiting for the encryption sync to receive the room key.");
492 return Ok(None);
493 }
494
495 match timeout(room_keys.next(), remaining).await {
496 Ok(Some(Ok(keys))) => {
497 if keys.iter().any(|key| &*key.room_id == room.room_id()) {
498 trace!("Received room keys for the room, retrying decryption");
499 break;
500 }
501 }
503 Ok(Some(Err(_))) => {
504 break;
507 }
508 Ok(None) => {
509 debug!("The room keys stream ended while waiting for the room key.");
510 return Ok(None);
511 }
512 Err(_) => {
513 debug!("Timeout waiting for the encryption sync to receive the room key.");
514 return Ok(None);
515 }
516 }
517 }
518 }
519 }
520
521 #[instrument(skip_all)]
540 async fn try_sliding_sync(
541 &self,
542 requests: &[NotificationItemsRequest],
543 ) -> Result<BTreeMap<OwnedEventId, (OwnedRoomId, Option<RawNotificationEvent>)>, Error> {
544 const MAX_SLIDING_SYNC_ATTEMPTS: u64 = 3;
545 let _guard = self.notification_sync_mutex.lock().await;
548
549 let raw_notifications = Arc::new(Mutex::new(BTreeMap::new()));
554 let handler_raw_notification = raw_notifications.clone();
555
556 let raw_invites = Arc::new(Mutex::new(BTreeMap::new()));
557 let handler_raw_invites = raw_invites.clone();
558
559 let user_id = self.client.user_id().unwrap().to_owned();
560 let room_ids = requests.iter().map(|req| req.room_id.clone()).collect::<Vec<_>>();
561
562 let requests = Arc::new(requests.iter().map(|req| (*req).clone()).collect::<Vec<_>>());
563
564 let timeline_event_handler = self.client.add_event_handler({
565 let requests = requests.clone();
566 move |raw: Raw<AnySyncTimelineEvent>| async move {
567 match &raw.get_field::<OwnedEventId>("event_id") {
568 Ok(Some(event_id)) => {
569 let Some(request) =
570 &requests.iter().find(|request| request.event_ids.contains(event_id))
571 else {
572 return;
573 };
574
575 let room_id = request.room_id.clone();
576
577 handler_raw_notification.lock().unwrap().insert(
581 event_id.to_owned(),
582 (room_id, Some(RawNotificationEvent::Timeline(raw))),
583 );
584 }
585 Ok(None) => {
586 warn!("a sync event had no event id");
587 }
588 Err(err) => {
589 warn!("failed to deserialize sync event id: {err}");
590 }
591 }
592 }
593 });
594
595 let handler_raw_notifications = raw_notifications.clone();
596 let stripped_member_handler = self.client.add_event_handler({
597 let requests = requests.clone();
598 let room_ids: Vec<_> = room_ids.clone();
599 move |raw: Raw<StrippedRoomMemberEvent>, room: Room| async move {
600 if !room_ids.contains(&room.room_id().to_owned()) {
601 return;
602 }
603
604 let deserialized = match raw.deserialize() {
605 Ok(d) => d,
606 Err(err) => {
607 warn!("failed to deserialize raw stripped room member event: {err}");
608 return;
609 }
610 };
611
612 trace!("received a stripped room member event");
613
614 match &raw.get_field::<OwnedEventId>("event_id") {
618 Ok(Some(event_id)) => {
619 let request =
620 &requests.iter().find(|request| request.event_ids.contains(event_id));
621 if request.is_none() {
622 return;
623 }
624 let room_id = request.unwrap().room_id.clone();
625
626 handler_raw_notifications.lock().unwrap().insert(
630 event_id.to_owned(),
631 (room_id, Some(RawNotificationEvent::Invite(raw))),
632 );
633 return;
634 }
635 Ok(None) => {
636 warn!("a room member event had no id");
637 }
638 Err(err) => {
639 warn!("failed to deserialize room member event id: {err}");
640 }
641 }
642
643 if deserialized.content.membership == MembershipState::Invite
646 && deserialized.state_key == user_id
647 {
648 trace!("found an invite event for the current user");
649 handler_raw_invites
653 .lock()
654 .unwrap()
655 .insert(deserialized.state_key, Some(RawNotificationEvent::Invite(raw)));
656 } else {
657 trace!("not an invite event, or not for the current user");
658 }
659 }
660 });
661
662 let required_state = vec![
664 (StateEventType::RoomEncryption, "".to_owned()),
665 (StateEventType::RoomMember, "$LAZY".to_owned()),
666 (StateEventType::RoomMember, "$ME".to_owned()),
667 (StateEventType::RoomCanonicalAlias, "".to_owned()),
668 (StateEventType::RoomName, "".to_owned()),
669 (StateEventType::RoomAvatar, "".to_owned()),
670 (StateEventType::RoomPowerLevels, "".to_owned()),
671 (StateEventType::RoomJoinRules, "".to_owned()),
672 (StateEventType::CallMember, "*".to_owned()),
673 (StateEventType::RoomCreate, "".to_owned()),
674 (StateEventType::MemberHints, "".to_owned()),
675 ];
676
677 let invites = SlidingSyncList::builder("invites")
678 .sync_mode(SlidingSyncMode::new_selective().add_range(0..=16))
679 .timeline_limit(8)
680 .required_state(required_state.clone())
681 .filters(Some(assign!(http::request::ListFilters::default(), {
682 is_invite: Some(true),
683 })));
684
685 let sync = self
686 .client
687 .sliding_sync(Self::CONNECTION_ID)?
688 .poll_timeout(self.timeouts.sync_poll_timeout)
689 .network_timeout(self.timeouts.sync_network_timeout)
690 .with_account_data_extension(
691 assign!(http::request::AccountData::default(), { enabled: Some(true) }),
692 )
693 .add_list(invites)
694 .build()
695 .await?;
696
697 sync.add_room_subscriptions(
698 &room_ids.iter().map(|id| id.deref()).collect::<Vec<&RoomId>>(),
699 Some(assign!(http::request::RoomSubscription::default(), {
700 required_state,
701 timeline_limit: uint!(16)
702 })),
703 true,
704 );
705
706 let mut remaining_attempts = MAX_SLIDING_SYNC_ATTEMPTS;
707
708 let stream = sync.sync();
709 pin_mut!(stream);
710
711 let expected_event_count = requests.iter().map(|req| req.event_ids.len()).sum::<usize>();
713
714 loop {
715 if stream.next().await.is_none() {
716 break;
718 }
719
720 let event_count = raw_notifications.lock().unwrap().len();
721 let invite_count = raw_invites.lock().unwrap().len();
722
723 let current_attempt = 1 + MAX_SLIDING_SYNC_ATTEMPTS - remaining_attempts;
724 trace!(
725 "Attempt #{current_attempt}: \
726 Found {event_count} notification(s), \
727 {invite_count} invite event(s), \
728 expected {expected_event_count} total",
729 );
730
731 if event_count + invite_count == expected_event_count {
736 break;
738 }
739
740 remaining_attempts -= 1;
741 warn!("There are some missing notifications, remaining attempts: {remaining_attempts}");
742 if remaining_attempts == 0 {
743 break;
745 }
746 }
747
748 self.client.remove_event_handler(stripped_member_handler);
749 self.client.remove_event_handler(timeline_event_handler);
750
751 let mut notifications = raw_notifications.clone().lock().unwrap().clone();
752 let mut missing_event_ids = Vec::new();
753
754 for request in requests.iter() {
756 for event_id in &request.event_ids {
757 if !notifications.contains_key(event_id) {
758 missing_event_ids.push((request.room_id.to_owned(), event_id.to_owned()));
759 }
760 }
761 }
762
763 for (room_id, missing_event_id) in missing_event_ids {
765 trace!("we didn't have a non-invite event, looking for invited room now");
766 if let Some(room) = self.client.get_room(&room_id) {
767 if room.state() == RoomState::Invited {
768 if let Some((_, stripped_event)) = raw_invites.lock().unwrap().pop_first() {
769 notifications
770 .insert(missing_event_id, (room_id.to_owned(), stripped_event));
771 }
772 } else {
773 debug!("the room isn't in the invited state");
774 }
775 } else {
776 warn!(%room_id, "unknown room, can't check for invite events");
777 }
778 }
779
780 let found = if notifications.len() == expected_event_count { "" } else { "not " };
781 trace!("all notification events have{found} been found");
782
783 Ok(notifications)
784 }
785
786 pub async fn get_notification_with_sliding_sync(
787 &self,
788 room_id: &RoomId,
789 event_id: &EventId,
790 ) -> Result<NotificationStatus, Error> {
791 info!("fetching notification event with a sliding sync");
792
793 let request = NotificationItemsRequest {
794 room_id: room_id.to_owned(),
795 event_ids: vec![event_id.to_owned()],
796 };
797
798 let mut get_notifications_result =
799 self.get_notifications_with_sliding_sync(&[request]).await?;
800
801 get_notifications_result.remove(event_id).unwrap_or(Ok(NotificationStatus::EventNotFound))
802 }
803
804 async fn compute_status(
809 &self,
810 room: &Room,
811 push_actions: Option<&[Action]>,
812 raw_event: RawNotificationEvent,
813 state_events: Vec<Raw<AnyStateEvent>>,
814 ) -> Result<NotificationStatus, Error> {
815 if let Some(actions) = push_actions
816 && !actions.iter().any(|a| a.should_notify())
817 {
818 return Ok(NotificationStatus::EventFilteredOut);
820 }
821
822 let notification_item =
823 NotificationItem::new(room, raw_event, push_actions, state_events).await?;
824
825 if self.client.is_user_ignored(notification_item.event.sender()).await {
826 Ok(NotificationStatus::EventFilteredOut)
827 } else {
828 Ok(NotificationStatus::Event(Box::new(notification_item)))
829 }
830 }
831
832 pub async fn get_notifications_with_sliding_sync(
837 &self,
838 requests: &[NotificationItemsRequest],
839 ) -> Result<BatchNotificationFetchingResult, Error> {
840 let raw_events = self.try_sliding_sync(requests).await?;
841
842 let mut batch_result = BatchNotificationFetchingResult::new();
843
844 for (event_id, (room_id, raw_event)) in raw_events.into_iter() {
845 let Some(room) = self.client.get_room(&room_id) else { return Err(Error::UnknownRoom) };
848
849 let Some(raw_event) = raw_event else {
850 batch_result.insert(event_id, Ok(NotificationStatus::EventNotFound));
852 continue;
853 };
854
855 let (raw_event, push_actions) = match &raw_event {
856 RawNotificationEvent::Timeline(timeline_event) => {
857 let event_for_redaction_check: AnySyncTimelineEvent =
859 match timeline_event.deserialize() {
860 Ok(event) => event,
861 Err(_) => {
862 batch_result.insert(event_id, Err(Error::InvalidRumaEvent));
863 continue;
864 }
865 };
866
867 if is_event_redacted(&event_for_redaction_check) {
868 batch_result.insert(event_id, Ok(NotificationStatus::EventRedacted));
869 continue;
870 }
871
872 match self.retry_decryption(&room, timeline_event).await {
875 Ok(Some(timeline_event)) => {
876 let push_actions = timeline_event.push_actions().map(ToOwned::to_owned);
877 (
878 RawNotificationEvent::Timeline(timeline_event.into_raw()),
879 push_actions,
880 )
881 }
882
883 Ok(None) => {
884 match room.event_push_actions(timeline_event).await {
888 Ok(push_actions) => (raw_event.clone(), push_actions),
889 Err(err) => {
890 batch_result.insert(event_id, Err(err.into()));
892 continue;
893 }
894 }
895 }
896
897 Err(err) => {
898 batch_result.insert(event_id, Err(err));
899 continue;
900 }
901 }
902 }
903
904 RawNotificationEvent::Invite(invite_event) => {
905 match room.event_push_actions(invite_event).await {
908 Ok(push_actions) => {
909 (RawNotificationEvent::Invite(invite_event.clone()), push_actions)
910 }
911 Err(err) => {
912 batch_result.insert(event_id, Err(err.into()));
913 continue;
914 }
915 }
916 }
917 };
918
919 let notification_status_result =
920 self.compute_status(&room, push_actions.as_deref(), raw_event, Vec::new()).await;
921
922 batch_result.insert(event_id, notification_status_result);
923 }
924
925 Ok(batch_result)
926 }
927
928 pub async fn get_notification_with_context(
940 &self,
941 room_id: &RoomId,
942 event_id: &EventId,
943 ) -> Result<NotificationStatus, Error> {
944 info!("fetching notification event with a /context query");
945
946 let Some(room) = self.parent_client.get_room(room_id) else {
948 return Err(Error::UnknownRoom);
949 };
950
951 let response = room.event_with_context(event_id, true, uint!(0), None).await?;
952
953 let mut timeline_event = response.event.ok_or(Error::ContextMissingEvent)?;
954 let state_events = response.state;
955
956 let event_for_redaction_check: AnySyncTimelineEvent =
958 timeline_event.raw().deserialize().map_err(|_| Error::InvalidRumaEvent)?;
959
960 if is_event_redacted(&event_for_redaction_check) {
961 return Ok(NotificationStatus::EventRedacted);
962 }
963
964 if let Some(decrypted_event) = self.retry_decryption(&room, timeline_event.raw()).await? {
965 timeline_event = decrypted_event;
966 }
967
968 let push_actions = timeline_event.push_actions().map(ToOwned::to_owned);
969
970 self.compute_status(
971 &room,
972 push_actions.as_deref(),
973 RawNotificationEvent::Timeline(timeline_event.into_raw()),
974 state_events,
975 )
976 .await
977 }
978}
979
980enum DecryptionAttempt {
982 Decrypted(TimelineEvent),
984
985 MissingRoomKey,
988
989 Unrecoverable,
992}
993
994async fn try_decrypt(
996 room: &Room,
997 raw_event: &Raw<AnySyncTimelineEvent>,
998 push_ctx: Option<&PushContext>,
999) -> Result<DecryptionAttempt, matrix_sdk::Error> {
1000 let new_event = room
1004 .decrypt_event(raw_event.cast_ref_unchecked::<OriginalSyncRoomEncryptedEvent>(), push_ctx)
1005 .await?;
1006
1007 if let matrix_sdk::deserialized_responses::TimelineEventKind::UnableToDecrypt {
1008 utd_info, ..
1009 } = &new_event.kind
1010 {
1011 return Ok(if utd_info.reason.is_missing_room_key() {
1012 DecryptionAttempt::MissingRoomKey
1013 } else {
1014 debug!(
1015 "Event could not be decrypted, but waiting longer is unlikely to help: {:?}",
1016 utd_info.reason
1017 );
1018 DecryptionAttempt::Unrecoverable
1019 });
1020 }
1021
1022 Ok(DecryptionAttempt::Decrypted(new_event))
1023}
1024
1025fn is_event_encrypted(event_type: TimelineEventType) -> bool {
1026 let is_still_encrypted = matches!(event_type, TimelineEventType::RoomEncrypted);
1027
1028 #[cfg(feature = "unstable-msc3956")]
1029 let is_still_encrypted =
1030 is_still_encrypted || matches!(event_type, ruma::events::TimelineEventType::Encrypted);
1031
1032 is_still_encrypted
1033}
1034
1035fn is_event_redacted(event: &AnySyncTimelineEvent) -> bool {
1036 match event {
1039 AnySyncTimelineEvent::MessageLike(msg) => msg.is_redacted(),
1040 _ => false,
1041 }
1042}
1043
1044#[derive(Debug)]
1045pub enum NotificationStatus {
1046 Event(Box<NotificationItem>),
1048 EventNotFound,
1050 EventFilteredOut,
1054 EventRedacted,
1056}
1057
1058#[derive(Debug, Clone)]
1059pub struct NotificationItemsRequest {
1060 pub room_id: OwnedRoomId,
1061 pub event_ids: Vec<OwnedEventId>,
1062}
1063
1064type BatchNotificationFetchingResult = BTreeMap<OwnedEventId, Result<NotificationStatus, Error>>;
1065
1066#[derive(Debug, Clone)]
1070pub enum RawNotificationEvent {
1071 Timeline(Raw<AnySyncTimelineEvent>),
1073 Invite(Raw<StrippedRoomMemberEvent>),
1076}
1077
1078#[derive(Debug)]
1081pub enum NotificationEvent {
1082 Timeline(Box<AnySyncTimelineEvent>),
1084 Invite(Box<StrippedRoomMemberEvent>),
1086}
1087
1088impl NotificationEvent {
1089 pub fn sender(&self) -> &UserId {
1090 match self {
1091 NotificationEvent::Timeline(ev) => ev.sender(),
1092 NotificationEvent::Invite(ev) => &ev.sender,
1093 }
1094 }
1095
1096 fn thread_id(&self) -> Option<OwnedEventId> {
1099 let NotificationEvent::Timeline(sync_timeline_event) = &self else {
1100 return None;
1101 };
1102 let AnySyncTimelineEvent::MessageLike(event) = sync_timeline_event.as_ref() else {
1103 return None;
1104 };
1105 let content = event.original_content()?;
1106 match content {
1107 AnyMessageLikeEventContent::RoomMessage(content) => match content.relates_to? {
1108 Relation::Thread(thread) => Some(thread.event_id),
1109 _ => None,
1110 },
1111 _ => None,
1112 }
1113 }
1114}
1115
1116#[derive(Debug)]
1118pub struct NotificationItem {
1119 pub event: NotificationEvent,
1121
1122 pub raw_event: RawNotificationEvent,
1124
1125 pub sender_display_name: Option<String>,
1127 pub sender_avatar_url: Option<String>,
1129 pub is_sender_name_ambiguous: bool,
1131
1132 pub room_computed_display_name: String,
1134 pub room_avatar_url: Option<String>,
1136 pub room_canonical_alias: Option<String>,
1138 pub room_topic: Option<String>,
1140 pub room_join_rule: Option<JoinRule>,
1144 pub is_room_encrypted: Option<bool>,
1146 pub is_direct_message_room: bool,
1148 pub joined_members_count: u64,
1150 pub service_members: Vec<String>,
1152 pub active_service_members_count: u64,
1153 pub is_space: bool,
1155
1156 pub is_noisy: Option<bool>,
1161 pub has_mention: Option<bool>,
1162 pub thread_id: Option<OwnedEventId>,
1163
1164 pub actions: Option<Vec<Action>>,
1166
1167 pub room_is_dm: bool,
1169}
1170
1171impl NotificationItem {
1172 async fn new(
1173 room: &Room,
1174 raw_event: RawNotificationEvent,
1175 push_actions: Option<&[Action]>,
1176 state_events: Vec<Raw<AnyStateEvent>>,
1177 ) -> Result<Self, Error> {
1178 let event = match &raw_event {
1179 RawNotificationEvent::Timeline(raw_event) => {
1180 let mut event = raw_event.deserialize().map_err(|_| Error::InvalidRumaEvent)?;
1181 if let AnySyncTimelineEvent::MessageLike(AnySyncMessageLikeEvent::RoomMessage(
1182 SyncRoomMessageEvent::Original(ev),
1183 )) = &mut event
1184 {
1185 ev.content.sanitize(DEFAULT_SANITIZER_MODE, RemoveReplyFallback::Yes);
1186 }
1187 NotificationEvent::Timeline(Box::new(event))
1188 }
1189 RawNotificationEvent::Invite(raw_event) => NotificationEvent::Invite(Box::new(
1190 raw_event.deserialize().map_err(|_| Error::InvalidRumaEvent)?,
1191 )),
1192 };
1193
1194 let sender = match room.state() {
1195 RoomState::Invited => room.invite_details().await?.inviter,
1196 _ => room.get_member_no_sync(event.sender()).await?,
1197 };
1198
1199 let (mut sender_display_name, mut sender_avatar_url, is_sender_name_ambiguous) =
1200 match &sender {
1201 Some(sender) => (
1202 sender.display_name().map(|s| s.to_owned()),
1203 sender.avatar_url().map(|s| s.to_string()),
1204 sender.name_ambiguous(),
1205 ),
1206 None => (None, None, false),
1207 };
1208
1209 if sender_display_name.is_none() || sender_avatar_url.is_none() {
1210 let sender_id = event.sender();
1211 for ev in state_events {
1212 let ev = match ev.deserialize() {
1213 Ok(ev) => ev,
1214 Err(err) => {
1215 warn!("Failed to deserialize a state event: {err}");
1216 continue;
1217 }
1218 };
1219 if ev.sender() != sender_id {
1220 continue;
1221 }
1222 if let AnyStateEventContentChange::RoomMember(StateEventContentChange::Original {
1223 content,
1224 ..
1225 }) = ev.content_change()
1226 {
1227 if sender_display_name.is_none() {
1228 sender_display_name = content.displayname;
1229 }
1230 if sender_avatar_url.is_none() {
1231 sender_avatar_url = content.avatar_url.map(|url| url.to_string());
1232 }
1233 }
1234 }
1235 }
1236
1237 let is_noisy = push_actions.map(|actions| actions.iter().any(|a| a.sound().is_some()));
1238 let has_mention = push_actions.map(|actions| actions.iter().any(|a| a.is_highlight()));
1239 let thread_id = event.thread_id().clone();
1240 let service_members = room
1241 .service_members()
1242 .unwrap_or_default()
1243 .iter()
1244 .map(ToString::to_string)
1245 .collect_vec();
1246
1247 let active_service_members_count =
1248 room.update_active_service_members().await?.unwrap_or_default().len() as u64;
1249
1250 let item = NotificationItem {
1251 event,
1252 raw_event,
1253 sender_display_name,
1254 sender_avatar_url,
1255 is_sender_name_ambiguous,
1256 room_computed_display_name: room.display_name().await?.to_string(),
1257 room_avatar_url: room.avatar_url().map(|s| s.to_string()),
1258 room_canonical_alias: room.canonical_alias().map(|c| c.to_string()),
1259 room_topic: room.topic(),
1260 room_join_rule: room.join_rule(),
1261 is_direct_message_room: room.is_direct().await?,
1262 is_room_encrypted: room
1263 .latest_encryption_state()
1264 .await
1265 .map(|state| state.is_encrypted())
1266 .ok(),
1267 joined_members_count: room.joined_members_count(),
1268 service_members,
1269 active_service_members_count,
1270 is_space: room.is_space(),
1271 is_noisy,
1272 has_mention,
1273 thread_id,
1274 actions: push_actions.map(|actions| actions.to_vec()),
1275 room_is_dm: room.compute_is_dm().await?,
1276 };
1277
1278 Ok(item)
1279 }
1280
1281 pub fn is_public(&self) -> Option<bool> {
1285 self.room_join_rule.as_ref().map(|rule| matches!(rule, JoinRule::Public))
1286 }
1287}
1288
1289#[derive(Debug, Error)]
1291pub enum Error {
1292 #[error(transparent)]
1293 BuildingLocalClient(ClientBuildError),
1294
1295 #[error("unknown room for a notification")]
1297 UnknownRoom,
1298
1299 #[error("invalid ruma event")]
1301 InvalidRumaEvent,
1302
1303 #[error("the sliding sync response doesn't include the target room")]
1306 SlidingSyncEmptyRoom,
1307
1308 #[error("the event was missing in the `/context` query")]
1309 ContextMissingEvent,
1310
1311 #[error(transparent)]
1313 SdkError(#[from] matrix_sdk::Error),
1314
1315 #[error(transparent)]
1317 StoreError(#[from] StoreError),
1318}
1319
1320#[cfg(test)]
1321mod tests {
1322 use std::collections::BTreeMap;
1323
1324 use matrix_sdk::test_utils::mocks::MatrixMockServer;
1325 use matrix_sdk_test::{ALICE, async_test, event_factory::EventFactory};
1326 use ruma::{
1327 api::client::sync::sync_events::v5,
1328 assign, event_id,
1329 events::room::{member::MembershipState, message::RedactedRoomMessageEventContent},
1330 owned_event_id, owned_room_id, room_id, user_id,
1331 };
1332 use strass::assert_let;
1333
1334 use crate::notification_client::{
1335 NotificationClient, NotificationItem, NotificationItemsRequest, NotificationProcessSetup,
1336 NotificationStatus, RawNotificationEvent,
1337 };
1338
1339 #[async_test]
1340 async fn test_notification_item_returns_thread_id() {
1341 let server = MatrixMockServer::new().await;
1342 let client = server.client_builder().build().await;
1343
1344 let room_id = room_id!("!a:b.c");
1345 let thread_root_event_id = event_id!("$root:b.c");
1346 let message = EventFactory::new()
1347 .room(room_id)
1348 .sender(user_id!("@sender:b.c"))
1349 .text_msg("Threaded")
1350 .in_thread(thread_root_event_id, event_id!("$prev:b.c"))
1351 .into_raw_sync();
1352 let room = server.sync_joined_room(&client, room_id).await;
1353
1354 let raw_notification_event = RawNotificationEvent::Timeline(message);
1355 let notification_item =
1356 NotificationItem::new(&room, raw_notification_event, None, Vec::new())
1357 .await
1358 .expect("Could not create notification item");
1359
1360 assert_let!(Some(thread_id) = notification_item.thread_id);
1361 assert_eq!(thread_id, thread_root_event_id);
1362 }
1363
1364 #[async_test]
1365 async fn test_try_sliding_sync_ignores_invites_for_non_subscribed_rooms() {
1366 let server = MatrixMockServer::new().await;
1367 let client = server.client_builder().build().await;
1368
1369 let user_id = client.user_id().unwrap();
1370 let room_id = room_id!("!a:b.c");
1371 let invite = EventFactory::new()
1372 .room(room_id)
1373 .member(user_id)
1374 .membership(MembershipState::Invite)
1375 .no_event_id()
1376 .into_raw_sync_state();
1377 let mut room = v5::response::Room::new();
1378 room.invite_state = Some(vec![invite.cast_unchecked()]);
1379 let rooms = BTreeMap::from_iter([(room_id.to_owned(), room)]);
1380 server
1381 .mock_sliding_sync()
1382 .ok(assign!(v5::Response::new("1".to_owned()), {
1383 rooms: rooms,
1384 }))
1385 .mount()
1386 .await;
1387
1388 let notification_client =
1389 NotificationClient::new(client.clone(), NotificationProcessSetup::MultipleProcesses)
1390 .await
1391 .expect("Could not create a notification client");
1392
1393 let event_id = owned_event_id!("$a:b.c");
1396 let result = notification_client
1397 .try_sliding_sync(&[NotificationItemsRequest {
1398 room_id: owned_room_id!("!other:b.c"),
1399 event_ids: vec![event_id.clone()],
1400 }])
1401 .await
1402 .expect("Could not run sliding sync");
1403
1404 assert!(result.is_empty());
1405
1406 let result = notification_client
1408 .try_sliding_sync(&[NotificationItemsRequest {
1409 room_id: room_id.to_owned(),
1410 event_ids: vec![event_id.clone()],
1411 }])
1412 .await
1413 .expect("Could not run sliding sync");
1414
1415 assert!(!result.is_empty());
1417
1418 let (in_room_id, event) = &result[&event_id];
1421 assert_eq!(room_id, in_room_id);
1422 assert_let!(Some(RawNotificationEvent::Invite(raw_invite)) = event);
1423
1424 let invite = raw_invite.deserialize().expect("Could not deserialize invite event");
1425 assert_eq!(invite.state_key, user_id.to_string());
1426 assert_eq!(invite.content.membership, MembershipState::Invite);
1427 }
1428
1429 #[async_test]
1430 async fn test_redacted_event_returns_event_redacted_status() {
1431 let server = MatrixMockServer::new().await;
1432 let client = server.client_builder().build().await;
1433
1434 let room_id = room_id!("!a:b.c");
1435
1436 let event_id = owned_event_id!("$redacted:b.c");
1438 let redacted_event = EventFactory::new()
1439 .room(room_id)
1440 .sender(user_id!("@sender:b.c"))
1441 .redacted(&ALICE, RedactedRoomMessageEventContent::new())
1442 .event_id(&event_id)
1443 .into_raw();
1444 let mut room = v5::response::Room::new();
1445 room.timeline = vec![redacted_event];
1446
1447 let mut rooms = BTreeMap::new();
1448 rooms.insert(room_id.to_owned(), room);
1449
1450 server
1451 .mock_sliding_sync()
1452 .ok(assign!(v5::Response::new("1".to_owned()), {
1453 rooms: rooms,
1454 }))
1455 .mount()
1456 .await;
1457
1458 let notification_client =
1459 NotificationClient::new(client.clone(), NotificationProcessSetup::MultipleProcesses)
1460 .await
1461 .expect("Could not create a notification client");
1462
1463 let result: NotificationStatus = notification_client
1464 .get_notification_with_sliding_sync(room_id, &event_id)
1465 .await
1466 .expect("Could not get notification");
1467
1468 match result {
1469 NotificationStatus::EventRedacted => {
1470 }
1472 other => panic!("Expected EventRedacted, got {:?}", other),
1473 }
1474 }
1475}