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