Skip to main content

matrix_sdk_ui/
notification_client.rs

1// Copyright 2023 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 that specific language governing permissions and
13// limitations under the License.
14
15use 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/// What kind of process setup do we have for this notification client?
63#[derive(Clone)]
64pub enum NotificationProcessSetup {
65    /// The notification client may run on a separate process than the rest of
66    /// the app.
67    ///
68    /// For instance, this is the case on iOS, where notifications are handled
69    /// in a separate process (the Notification Service Extension, aka NSE).
70    ///
71    /// In that case, a cross-process lock will be used to coordinate writes
72    /// into the stores handled by the SDK.
73    MultipleProcesses,
74
75    /// The notification client runs in the same process as the rest of the
76    /// `Client` performing syncs.
77    ///
78    /// For instance, this is the case on Android, where a notification will
79    /// wake up the main app process.
80    ///
81    /// In that case, a smart reference to the [`SyncService`] must be provided.
82    SingleProcess { sync_service: Arc<SyncService> },
83}
84
85/// Timeouts applied by a [`NotificationClient`] while fetching the content of
86/// notifications.
87#[derive(Clone, Copy, Debug, PartialEq, Eq)]
88pub struct NotificationClientTimeouts {
89    /// Long-poll timeout of the sliding sync request retrieving the notified
90    /// events, i.e. how long the homeserver waits for the events to be
91    /// available before answering.
92    pub sync_poll_timeout: Duration,
93
94    /// Extra time allowed for the network round trip of the sliding sync
95    /// request retrieving the notified events, on top of
96    /// [`Self::sync_poll_timeout`].
97    pub sync_network_timeout: Duration,
98
99    /// Maximum time spent waiting for a missing room key, when an event in a
100    /// notification can't be decrypted.
101    ///
102    /// This applies to both ways of obtaining the key, so that they give up
103    /// after the same amount of time:
104    ///
105    /// - When the notification client runs the encryption sync itself, this
106    ///   bounds the time spent running it, once a minimum number of iterations
107    ///   (currently two) have been run. Decryption is attempted after each
108    ///   iteration, so this only bounds the unsuccessful case: once the
109    ///   deadline has passed, no new iteration is started.
110    /// - In a [`NotificationProcessSetup::SingleProcess`] setup where the main
111    ///   encryption sync is already running, the notification client must not
112    ///   run a second one and waits for the running one to receive the key
113    ///   instead. The wait ends as soon as a key for the room is received, so
114    ///   this only bounds the case where the key doesn't arrive.
115    ///
116    /// In both cases, the event is returned undecrypted once the deadline has
117    /// passed.
118    pub decryption_deadline: Duration,
119
120    /// Long-poll timeout of each request of the encryption sync run to obtain a
121    /// missing room key, i.e. how long the homeserver waits for a to-device
122    /// message to arrive before answering.
123    ///
124    /// Only applies when the notification client runs the encryption sync
125    /// itself. Together with [`Self::decryption_deadline`], this determines how
126    /// many iterations are run when the homeserver has nothing to return.
127    pub encryption_sync_poll_timeout: Duration,
128
129    /// Extra time allowed for the network round trip of each request of the
130    /// encryption sync, on top of [`Self::encryption_sync_poll_timeout`]. This
131    /// is an upper bound on how long a request may take.
132    pub encryption_sync_network_timeout: Duration,
133}
134
135impl Default for NotificationClientTimeouts {
136    /// Conservative defaults, kept small since a push handler might be short on
137    /// time.
138    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            // Set so that the minimum number of encryption sync iterations,
146            // when the homeserver has nothing to return, use up the whole
147            // deadline and no further iteration is started.
148            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
155/// A client specialized for handling push notifications received over the
156/// network, for an app.
157///
158/// In particular, it takes care of running a full decryption sync, in case the
159/// event in the notification was impossible to decrypt beforehand.
160pub struct NotificationClient {
161    /// SDK client that uses an in-memory state store.
162    client: Client,
163
164    /// SDK client that uses the same state store as the caller's context.
165    parent_client: Client,
166
167    /// Is the notification client running on its own process or not?
168    process_setup: NotificationProcessSetup,
169
170    /// A mutex to serialize requests to the notifications sliding sync.
171    ///
172    /// If several notifications come in at the same time (e.g. network was
173    /// unreachable because of airplane mode or something similar), then we need
174    /// to make sure that repeated calls to `get_notification` won't cause
175    /// multiple requests with the same `conn_id` we're using for notifications.
176    /// This mutex solves this by sequentializing the requests.
177    notification_sync_mutex: AsyncMutex<()>,
178
179    /// A mutex to serialize requests to the encryption sliding sync that's used
180    /// in case we didn't have the keys to decipher an event.
181    ///
182    /// Same reasoning as [`Self::notification_sync_mutex`].
183    encryption_sync_mutex: AsyncMutex<()>,
184
185    /// Timeouts applied while fetching notifications. See
186    /// [`Self::with_timeouts`].
187    timeouts: NotificationClientTimeouts,
188}
189
190impl NotificationClient {
191    const CONNECTION_ID: &'static str = "notifications";
192    const LOCK_ID: &'static str = "notifications";
193
194    /// Minimum number of encryption sync iterations to run when an event in a
195    /// notification can't be decrypted, before
196    /// [`NotificationClientTimeouts::decryption_deadline`] is considered.
197    ///
198    /// The first iteration sends the e2ee requests and receives pending
199    /// to-device messages; the second lets the homeserver forward what those
200    /// requests triggered.
201    const MIN_DECRYPTION_ITERATIONS: usize = 2;
202
203    /// Create a new notification client.
204    pub async fn new(
205        parent_client: Client,
206        process_setup: NotificationProcessSetup,
207    ) -> Result<Self, Error> {
208        // Only create the lock id if cross process lock is needed (multiple
209        // processes)
210        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    /// Overrides the timeouts applied while fetching notifications.
229    pub fn with_timeouts(mut self, timeouts: NotificationClientTimeouts) -> Self {
230        self.timeouts = timeouts;
231        self
232    }
233
234    /// Returns the timeouts applied while fetching notifications.
235    pub fn timeouts(&self) -> &NotificationClientTimeouts {
236        &self.timeouts
237    }
238
239    /// Fetches a room by its ID using the in-memory state store backed client.
240    /// Useful to retrieve room information after running the limited
241    /// notification client sliding sync loop.
242    pub fn get_room(&self, room_id: &RoomId) -> Option<Room> {
243        self.client.get_room(room_id)
244    }
245
246    /// Fetches the content of a notification.
247    ///
248    /// This will first try to get the notification using a short-lived sliding
249    /// sync, and if the sliding-sync can't find the event, then it'll use a
250    /// `/context` query to find the event with associated member information.
251    ///
252    /// An error result means that we couldn't resolve the notification; in that
253    /// case, a dummy notification may be displayed instead.
254    #[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    /// Fetches the content of several notifications.
272    ///
273    /// This will first try to get the notifications using a short-lived sliding
274    /// sync, and if the sliding-sync can't find the events, then it'll use a
275    /// `/context` query to find the events with associated member information.
276    ///
277    /// An error result at the top level means that something failed when trying
278    /// to set up the notification fetching.
279    ///
280    /// For each notification item you can also receive an error, which means
281    /// something failed when trying to fetch that particular notification
282    /// (decryption, fetching push actions, etc.); in that case, a dummy
283    /// notification may be displayed instead.
284    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                    // If the notification for a given event wasn't found with
294                    // sliding sync, try with a /context for each event.
295                    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    /// Run an encryption sync loop, in case an event is still encrypted.
311    ///
312    /// Will return `Ok(Some)` if and only if:
313    ///
314    /// - the event was encrypted,
315    /// - we successfully ran an encryption sync or waited long enough for an
316    ///   existing encryption sync to decrypt the event.
317    ///
318    /// Otherwise, if the event was not encrypted, or couldn't be decrypted
319    /// (without causing a fatal error), will return `Ok(None)`.
320    #[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        // Serialize calls to this function.
334        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                // We're running on our own process, dedicated for
341                // notifications. In that case, create a dummy sync permit;
342                // we're guaranteed there's at most one since we've acquired the
343                // `encryption_sync_mutex' lock here.
344                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                    // There's already a sync service active, thus the
353                    // encryption sync is already running elsewhere, and we must
354                    // not run a second one. As a matter of fact, if the event
355                    // was encrypted, that means we were racing against the
356                    // encryption sync: wait for it to receive the room key,
357                    // then decrypt.
358                    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        // Run an `EncryptionSync` loop, trying to decrypt the event after each
365        // iteration. The first one fetches SS events and sends e2ee requests;
366        // the rest let the homeserver forward events those requests triggered.
367        //
368        // Stop once the event is decrypted, or once the minimum number of
369        // iterations has run and the deadline has passed.
370
371        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                    // The room key might have been persisted before this error
402                    // was raised. Don't exit directly so that redrycption is
403                    // attempted one last time.
404                    warn!("Encryption sync error, attempting to decrypt one last time: {err:#}");
405                    true
406                }
407
408                None => {
409                    // The sync terminated, or the cross-process lock is held by
410                    // the main app, which may well have fetched the room key
411                    // itself in the meantime: attempt to decrypt one last time.
412                    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    /// Wait for the main encryption sync to receive the room key needed to
445    /// decrypt `raw_event`, then decrypt it.
446    ///
447    /// This is used in a [`NotificationProcessSetup::SingleProcess`] setup when
448    /// the encryption sync is already running, since the notification client
449    /// must not run a second one.
450    ///
451    /// Returns `Ok(None)` if no key for the room has been received within
452    /// [`NotificationClientTimeouts::decryption_deadline`], or if the event
453    /// can't be decrypted for another reason.
454    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        // Subscribe before the first decryption attempt, so that a key received
461        // in between can't be missed. The notification client shares its
462        // `OlmMachine` with the parent client, which the running encryption
463        // sync belongs to, so keys it receives are both reported here and
464        // usable by `try_decrypt` right away.
465        let Some(room_keys) = self.parent_client.encryption().room_keys_received_stream().await
466        else {
467            // No `OlmMachine`, hence no keys to wait for: a single attempt is
468            // all we can do.
469            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            // Wait for keys of this room to be received, then try again.
489            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                        // Keys for other rooms can't help, keep waiting.
503                    }
504                    Ok(Some(Err(_))) => {
505                        // The stream lagged behind, so we may have missed keys
506                        // for the room: retry to be on the safe side.
507                        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    /// Try to run a sliding sync (without encryption) to retrieve the events
523    /// from the notification.
524    ///
525    /// An event can either be:
526    ///
527    /// - an invite event,
528    /// - or a non-invite event.
529    ///
530    /// In case it's a non-invite event, it's rather easy: we'll request
531    /// explicit state that'll be useful for building the `NotificationItem`,
532    /// and subscribe to the room which the notification relates to.
533    ///
534    /// In case it's an invite-event, it's trickier because the stripped event
535    /// may not contain the event id, so we can't just match on it. Rather, we
536    /// look at stripped room member events that may be fitting (i.e. match the
537    /// current user and are invites), and if the SDK concludes the room was in
538    /// the invited state, and we didn't find the event by id, _then_ we'll use
539    /// that stripped room member event.
540    #[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        // Serialize all the calls to this method by taking a lock at the
547        // beginning, that will be dropped later.
548        let _guard = self.notification_sync_mutex.lock().await;
549
550        // Set up a sliding sync that only subscribes to the room that had the
551        // notification, so we can figure out the full event and associated
552        // information.
553
554        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                        // found it! There shouldn't be a previous event before,
579                        // but if there is, that should be ok to just replace
580                        // it.
581                        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                // Try to match the event by event_id, as it's the most precise.
616                // In theory, we shouldn't receive it, so that's a first
617                // attempt.
618                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                        // found it! There shouldn't be a previous event before,
628                        // but if there is, that should be ok to just replace
629                        // it.
630                        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                // Try to match the event by membership and state_key for the
645                // current user.
646                if deserialized.content.membership == MembershipState::Invite
647                    && deserialized.state_key == user_id
648                {
649                    trace!("found an invite event for the current user");
650                    // This could be it! There might be several of these
651                    // following each other, so assume it's the latest one (in
652                    // sync ordering), and override a previous one if present.
653                    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        // Room power levels are necessary to build the push context.
664        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        // Sum the expected event count for each room
713        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                // Sliding sync aborted early.
718                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            // We can stop looking once we've received the expected number of
733            // events from the sync. Since we can receive only events or invites
734            // for rooms but not both, and we're not taking into account invites
735            // from not subscribed rooms, this check should be accurate.
736            if event_count + invite_count == expected_event_count {
737                // We got the events.
738                break;
739            }
740
741            remaining_attempts -= 1;
742            warn!("There are some missing notifications, remaining attempts: {remaining_attempts}");
743            if remaining_attempts == 0 {
744                // We're out of luck.
745                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        // Create the list of missing event ids after the syncs.
756        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        // Try checking if the missing notifications could be invites.
765        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    /// Given a (decrypted or not) event, figure out whether it should be
806    /// filtered out for other client-side reasons (such as the sender being
807    /// ignored, for instance), and returns the corresponding
808    /// [`NotificationStatus`].
809    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            // The event shouldn't notify: return early.
820            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    /// Get a list of full notifications, given a room id and event ids.
834    ///
835    /// This will run a small sliding sync to retrieve the content of the
836    /// events, along with extra data to form a rich notification context.
837    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            // At this point it should have been added by the sync, if it's not,
847            // give up.
848            let Some(room) = self.client.get_room(&room_id) else { return Err(Error::UnknownRoom) };
849
850            let Some(raw_event) = raw_event else {
851                // The event was not found, so we can't build a notification.
852                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                    // Check if the event is redacted first
859                    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                    // Timeline events may be encrypted, so make sure they get
874                    // decrypted first.
875                    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                            // The event was either not encrypted in the first
886                            // place, or we couldn't decrypt it after retrying.
887                            // Use the raw event as is.
888                            match room.event_push_actions(timeline_event).await {
889                                Ok(push_actions) => (raw_event.clone(), push_actions),
890                                Err(err) => {
891                                    // Could not get push actions.
892                                    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                    // Invite events can't be encrypted, so they should be in
907                    // clear text.
908                    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    /// Retrieve a notification using a `/context` query.
930    ///
931    /// This is for clients that are already running other sliding syncs in the
932    /// same process, so that most of the contextual information for the
933    /// notification should already be there. In particular, the room containing
934    /// the event MUST be known (via a sliding sync for invites, or another
935    /// sliding sync).
936    ///
937    /// An error result means that we couldn't resolve the notification; in that
938    /// case, a dummy notification may be displayed instead. A `None` result
939    /// means the notification has been filtered out by the user's push rules.
940    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        // See above comment.
948        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        // Check if the event is redacted
958        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
981/// The outcome of an attempt at decrypting a notified event.
982enum DecryptionAttempt {
983    /// The event could be decrypted.
984    Decrypted(TimelineEvent),
985
986    /// The event could not be decrypted because the room key is missing; it may
987    /// still arrive.
988    MissingRoomKey,
989
990    /// The event could not be decrypted, and waiting longer is unlikely to
991    /// help.
992    Unrecoverable,
993}
994
995/// Attempt to decrypt an encrypted timeline event of `room`.
996async fn try_decrypt(
997    room: &Room,
998    raw_event: &Raw<AnySyncTimelineEvent>,
999    push_ctx: Option<&PushContext>,
1000) -> Result<DecryptionAttempt, matrix_sdk::Error> {
1001    // Note: We specify the cast type in case the
1002    // `experimental-encrypted-state-events` feature is enabled, which provides
1003    // multiple cast implementations.
1004    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    // Check if the event is a message-like event but has no original content
1038    // (i.e., redacted)
1039    match event {
1040        AnySyncTimelineEvent::MessageLike(msg) => msg.is_redacted(),
1041        _ => false,
1042    }
1043}
1044
1045#[derive(Debug)]
1046pub enum NotificationStatus {
1047    /// The event has been found and was not filtered out.
1048    Event(Box<NotificationItem>),
1049    /// The event couldn't be found in the network queries used to find it.
1050    EventNotFound,
1051    /// The event has been filtered out, either because of the user's push
1052    /// rules, or because the user which triggered it is ignored by the current
1053    /// user.
1054    EventFilteredOut,
1055    /// The event has been redacted and has no meaningful content.
1056    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/// The Notification event as it was fetched from remote for the given
1068/// `event_id`, represented as Raw but decrypted, thus only whether it is an
1069/// invite or regular Timeline event has been determined.
1070#[derive(Debug, Clone)]
1071pub enum RawNotificationEvent {
1072    /// The raw event for a timeline event
1073    Timeline(Raw<AnySyncTimelineEvent>),
1074    /// The notification contains an invitation with the given
1075    /// StrippedRoomMemberEvent (in raw here)
1076    Invite(Raw<StrippedRoomMemberEvent>),
1077}
1078
1079/// The deserialized Event as it was fetched from remote for the given
1080/// `event_id` and after decryption (if possible).
1081#[derive(Debug)]
1082pub enum NotificationEvent {
1083    /// The Notification was for a TimelineEvent
1084    Timeline(Box<AnySyncTimelineEvent>),
1085    /// The Notification is an invite with the given stripped room event data
1086    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    /// Returns the root event id of the thread the notification event is in, if
1098    /// any.
1099    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/// A notification with its full content.
1118#[derive(Debug)]
1119pub struct NotificationItem {
1120    /// Underlying Ruma event.
1121    pub event: NotificationEvent,
1122
1123    /// The raw of the underlying event.
1124    pub raw_event: RawNotificationEvent,
1125
1126    /// Display name of the sender.
1127    pub sender_display_name: Option<String>,
1128    /// Avatar URL of the sender.
1129    pub sender_avatar_url: Option<String>,
1130    /// Is the sender's name ambiguous?
1131    pub is_sender_name_ambiguous: bool,
1132
1133    /// Room computed display name.
1134    pub room_computed_display_name: String,
1135    /// Room avatar URL.
1136    pub room_avatar_url: Option<String>,
1137    /// Room canonical alias.
1138    pub room_canonical_alias: Option<String>,
1139    /// Room topic.
1140    pub room_topic: Option<String>,
1141    /// Room join rule.
1142    ///
1143    /// Set to `None` if the join rule for this room is not available.
1144    pub room_join_rule: Option<JoinRule>,
1145    /// Is this room encrypted?
1146    pub is_room_encrypted: Option<bool>,
1147    /// Is this room considered a direct message?
1148    pub is_direct_message_room: bool,
1149    /// Numbers of members who joined the room.
1150    pub joined_members_count: u64,
1151    /// Number of service members in the room.
1152    pub service_members: Vec<String>,
1153    pub active_service_members_count: u64,
1154    /// Is the room a space?
1155    pub is_space: bool,
1156
1157    /// Is it a noisy notification? (i.e. does any push action contain a sound
1158    /// action)
1159    ///
1160    /// It is set if and only if the push actions could be determined.
1161    pub is_noisy: Option<bool>,
1162    pub has_mention: Option<bool>,
1163    pub thread_id: Option<OwnedEventId>,
1164
1165    /// The push actions for this notification (notify, sound, highlight, etc.).
1166    pub actions: Option<Vec<Action>>,
1167
1168    /// Whether the room this notification is from is a DM or not.
1169    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    /// Returns whether this room is public or not, based on the join rule.
1283    ///
1284    /// Maybe return `None` if the join rule is not available.
1285    pub fn is_public(&self) -> Option<bool> {
1286        self.room_join_rule.as_ref().map(|rule| matches!(rule, JoinRule::Public))
1287    }
1288}
1289
1290/// An error for the [`NotificationClient`].
1291#[derive(Debug, Error)]
1292pub enum Error {
1293    #[error(transparent)]
1294    BuildingLocalClient(ClientBuildError),
1295
1296    /// The room associated to this event wasn't found.
1297    #[error("unknown room for a notification")]
1298    UnknownRoom,
1299
1300    /// The Ruma event contained within this notification couldn't be parsed.
1301    #[error("invalid ruma event")]
1302    InvalidRumaEvent,
1303
1304    /// When calling `get_notification_with_sliding_sync`, the room was missing
1305    /// in the response.
1306    #[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    /// An error forwarded from the client.
1313    #[error(transparent)]
1314    SdkError(#[from] matrix_sdk::Error),
1315
1316    /// An error forwarded from the underlying state store.
1317    #[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        // Check we don't receive the invite for a different room, even if it
1395        // was included in the sync response
1396        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        // Now try fetching the invite for the previously ignored room
1408        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        // Check we did receive an event
1417        assert!(!result.is_empty());
1418
1419        // Try to assert it's the same event (since we don't have an event id)
1420        // We can check its room, sender and membership state
1421        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        // Create a redacted message event (no content)
1438        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                // Success - redacted event was properly detected
1472            }
1473            other => panic!("Expected EventRedacted, got {:?}", other),
1474        }
1475    }
1476}