feat(room): add JoinRequest subscriptions

bindings/matrix-sdk-ffi/src/room.rs

@@ -1,7 +1,7 @@
 use std::{collections::HashMap, pin::pin, sync::Arc};
 
 use anyhow::{Context, Result};
-use futures_util::StreamExt;
+use futures_util::{pin_mut, StreamExt};
 use matrix_sdk::{
     crypto::LocalTrust,
     event_cache::paginator::PaginatorError,
@@ -911,6 +911,108 @@ impl Room {
         room_event_cache.clear().await?;
         Ok(())
     }
+
+    /// Subscribes to requests to join this room (knock member events), using a
+    /// `listener` to be notified of the changes.
+    ///
+    /// The current requests to join the room will be emitted immediately
+    /// when subscribing, along with a [`TaskHandle`] to cancel the
+    /// subscription.
+    pub async fn subscribe_to_knock_requests(
+        self: Arc<Self>,
+        listener: Box<dyn KnockRequestsListener>,
+    ) -> Result<Arc<TaskHandle>, ClientError> {
+        let stream = self.inner.subscribe_to_knock_requests().await?;
+
+        let handle = Arc::new(TaskHandle::new(RUNTIME.spawn(async move {
+            pin_mut!(stream);
+            while let Some(requests) = stream.next().await {
+                listener.call(requests.into_iter().map(Into::into).collect());
+            }
+        })));
+
+        Ok(handle)
+    }
+}
+
+impl From<matrix_sdk::room::knock_requests::KnockRequest> for KnockRequest {
+    fn from(request: matrix_sdk::room::knock_requests::KnockRequest) -> Self {
+        Self {
+            event_id: request.event_id.to_string(),
+            user_id: request.member_info.user_id.to_string(),
+            room_id: request.room_id().to_string(),
+            display_name: request.member_info.display_name.clone(),
+            avatar_url: request.member_info.avatar_url.as_ref().map(|url| url.to_string()),
+            reason: request.member_info.reason.clone(),
+            timestamp: request.timestamp.map(|ts| ts.into()),
+            is_seen: request.is_seen,
+            actions: Arc::new(KnockRequestActions { inner: request }),
+        }
+    }
+}
+
+/// A listener for receiving new requests to a join a room.
+#[matrix_sdk_ffi_macros::export(callback_interface)]
+pub trait KnockRequestsListener: Send + Sync {
+    fn call(&self, join_requests: Vec<KnockRequest>);
+}
+
+/// An FFI representation of a request to join a room.
+#[derive(Debug, Clone, uniffi::Record)]
+pub struct KnockRequest {
+    /// The event id of the event that contains the `knock` membership change.
+    pub event_id: String,
+    /// The user id of the user who's requesting to join the room.
+    pub user_id: String,
+    /// The room id of the room whose access was requested.
+    pub room_id: String,
+    /// The optional display name of the user who's requesting to join the room.
+    pub display_name: Option<String>,
+    /// The optional avatar url of the user who's requesting to join the room.
+    pub avatar_url: Option<String>,
+    /// An optional reason why the user wants join the room.
+    pub reason: Option<String>,
+    /// The timestamp when this request was created.
+    pub timestamp: Option<u64>,
+    /// Whether the knock request has been marked as `seen` so it can be
+    /// filtered by the client.
+    pub is_seen: bool,
+    /// A set of actions to perform for this knock request.
+    pub actions: Arc<KnockRequestActions>,
+}
+
+/// A set of actions to perform for a knock request.
+#[derive(Debug, Clone, uniffi::Object)]
+pub struct KnockRequestActions {
+    inner: matrix_sdk::room::knock_requests::KnockRequest,
+}
+
+#[matrix_sdk_ffi_macros::export]
+impl KnockRequestActions {
+    /// Accepts the knock request by inviting the user to the room.
+    pub async fn accept(&self) -> Result<(), ClientError> {
+        self.inner.accept().await.map_err(Into::into)
+    }
+
+    /// Declines the knock request by kicking the user from the room with an
+    /// optional reason.
+    pub async fn decline(&self, reason: Option<String>) -> Result<(), ClientError> {
+        self.inner.decline(reason.as_deref()).await.map_err(Into::into)
+    }
+
+    /// Declines the knock request by banning the user from the room with an
+    /// optional reason.
+    pub async fn decline_and_ban(&self, reason: Option<String>) -> Result<(), ClientError> {
+        self.inner.decline_and_ban(reason.as_deref()).await.map_err(Into::into)
+    }
+
+    /// Marks the knock request as 'seen'.
+    ///
+    /// **IMPORTANT**: this won't update the current reference to this request,
+    /// a new one with the updated value should be emitted instead.
+    pub async fn mark_as_seen(&self) -> Result<(), ClientError> {
+        self.inner.mark_as_seen().await.map_err(Into::into)
+    }
 }
 
 /// Generates a `matrix.to` permalink to the given room alias.

crates/matrix-sdk-base/Cargo.toml

@@ -51,7 +51,7 @@ assert_matches2 = { workspace = true, optional = true }
 async-trait = { workspace = true }
 bitflags = { version = "2.6.0", features = ["serde"] }
 decancer = "3.2.8"
-eyeball = { workspace = true }
+eyeball = { workspace = true, features = ["async-lock"] }
 eyeball-im = { workspace = true }
 futures-util = { workspace = true }
 growable-bloom-filter = { workspace = true }

crates/matrix-sdk-base/src/deserialized_responses.rs

@@ -30,7 +30,7 @@ use ruma::{
         StateEventContent, StaticStateEventContent, StrippedStateEvent, SyncStateEvent,
     },
     serde::Raw,
-    EventId, MilliSecondsSinceUnixEpoch, OwnedEventId, OwnedRoomId, OwnedUserId, UserId,
+    EventId, MilliSecondsSinceUnixEpoch, OwnedEventId, OwnedRoomId, OwnedUserId, UInt, UserId,
 };
 use serde::Serialize;
 use unicode_normalization::UnicodeNormalization;
@@ -476,6 +476,23 @@ impl MemberEvent {
                 .unwrap_or_else(|| self.user_id().localpart()),
         )
     }
+
+    /// The optional reason why the membership changed.
+    pub fn reason(&self) -> Option<&str> {
+        match self {
+            MemberEvent::Sync(SyncStateEvent::Original(c)) => c.content.reason.as_deref(),
+            MemberEvent::Stripped(e) => e.content.reason.as_deref(),
+            _ => None,
+        }
+    }
+
+    /// The optional timestamp for this member event.
+    pub fn timestamp(&self) -> Option<UInt> {
+        match self {
+            MemberEvent::Sync(SyncStateEvent::Original(c)) => Some(c.origin_server_ts.0),
+            _ => None,
+        }
+    }
 }
 
 impl SyncOrStrippedState<RoomPowerLevelsEventContent> {

crates/matrix-sdk-base/src/rooms/normal.rs

@@ -22,7 +22,7 @@ use std::{
 
 use as_variant::as_variant;
 use bitflags::bitflags;
-use eyeball::{SharedObservable, Subscriber};
+use eyeball::{AsyncLock, ObservableWriteGuard, SharedObservable, Subscriber};
 use futures_util::{Stream, StreamExt};
 #[cfg(feature = "experimental-sliding-sync")]
 use matrix_sdk_common::deserialized_responses::TimelineEventKind;
@@ -52,7 +52,7 @@ use ruma::{
         },
         tag::{TagEventContent, Tags},
         AnyRoomAccountDataEvent, AnyStrippedStateEvent, AnySyncStateEvent,
-        RoomAccountDataEventType, SyncStateEvent,
+        RoomAccountDataEventType, StateEventType, SyncStateEvent,
     },
     room::RoomType,
     serde::Raw,
@@ -77,7 +77,8 @@ use crate::{
     read_receipts::RoomReadReceipts,
     store::{DynStateStore, Result as StoreResult, StateStoreExt},
     sync::UnreadNotificationsCount,
-    Error, MinimalStateEvent, OriginalMinimalStateEvent, RoomMemberships,
+    Error, MinimalStateEvent, OriginalMinimalStateEvent, RoomMemberships, StateStoreDataKey,
+    StateStoreDataValue, StoreError,
 };
 
 /// Indicates that a notable update of `RoomInfo` has been applied, and why.
@@ -167,6 +168,12 @@ pub struct Room {
     /// to disk but held in memory.
     #[cfg(all(feature = "e2e-encryption", feature = "experimental-sliding-sync"))]
     pub latest_encrypted_events: Arc<SyncRwLock<RingBuffer<Raw<AnySyncTimelineEvent>>>>,
+
+    /// A map for ids of room membership events in the knocking state linked to
+    /// the user id of the user affected by the member event, that the current
+    /// user has marked as seen so they can be ignored.
+    pub seen_knock_request_ids_map:
+        SharedObservable<Option<BTreeMap<OwnedEventId, OwnedUserId>>, AsyncLock>,
 }
 
 /// The room summary containing member counts and members that should be used to
@@ -289,6 +296,7 @@ impl Room {
                 Self::MAX_ENCRYPTED_EVENTS,
             ))),
             room_info_notable_update_sender,
+            seen_knock_request_ids_map: SharedObservable::new_async(None),
         }
     }
 
@@ -1169,6 +1177,88 @@ impl Room {
     pub fn pinned_event_ids(&self) -> Option<Vec<OwnedEventId>> {
         self.inner.read().pinned_event_ids()
     }
+
+    /// Mark a list of requests to join the room as seen, given their state
+    /// event ids.
+    pub async fn mark_knock_requests_as_seen(&self, user_ids: &[OwnedUserId]) -> StoreResult<()> {
+        let raw_user_ids: Vec<&str> = user_ids.iter().map(|id| id.as_str()).collect();
+        let member_raw_events = self
+            .store
+            .get_state_events_for_keys(self.room_id(), StateEventType::RoomMember, &raw_user_ids)
+            .await?;
+        let mut event_to_user_ids = Vec::with_capacity(member_raw_events.len());
+
+        // Map the list of events ids to their user ids, if they are event ids for knock
+        // membership events. Log an error and continue otherwise.
+        for raw_event in member_raw_events {
+            let event = raw_event.cast::<RoomMemberEventContent>().deserialize()?;
+            match event {
+                SyncOrStrippedState::Sync(SyncStateEvent::Original(event)) => {
+                    if event.content.membership == MembershipState::Knock {
+                        event_to_user_ids.push((event.event_id, event.state_key))
+                    } else {
+                        warn!("Could not mark knock event as seen: event {} for user {} is not in Knock membership state.", event.event_id, event.state_key);
+                    }
+                }
+                _ => warn!(
+                    "Could not mark knock event as seen: event for user {} is not valid.",
+                    event.state_key()
+                ),
+            }
+        }
+
+        let mut current_seen_events_guard = self.seen_knock_request_ids_map.write().await;
+        // We're not calling `get_seen_join_request_ids` here because we need to keep
+        // the Mutex's guard until we've updated the data
+        let mut current_seen_events = if current_seen_events_guard.is_none() {
+            self.load_cached_knock_request_ids().await?
+        } else {
+            current_seen_events_guard.clone().unwrap()
+        };
+
+        current_seen_events.extend(event_to_user_ids);
+
+        ObservableWriteGuard::set(
+            &mut current_seen_events_guard,
+            Some(current_seen_events.clone()),
+        );
+
+        self.store
+            .set_kv_data(
+                StateStoreDataKey::SeenKnockRequests(self.room_id()),
+                StateStoreDataValue::SeenKnockRequests(current_seen_events),
+            )
+            .await?;
+
+        Ok(())
+    }
+
+    /// Get the list of seen knock request event ids in this room.
+    pub async fn get_seen_knock_request_ids(
+        &self,
+    ) -> Result<BTreeMap<OwnedEventId, OwnedUserId>, StoreError> {
+        let mut guard = self.seen_knock_request_ids_map.write().await;
+        if guard.is_none() {
+            ObservableWriteGuard::set(
+                &mut guard,
+                Some(self.load_cached_knock_request_ids().await?),
+            );
+        }
+        Ok(guard.clone().unwrap_or_default())
+    }
+
+    /// This loads the current list of seen knock request ids from the state
+    /// store.
+    async fn load_cached_knock_request_ids(
+        &self,
+    ) -> StoreResult<BTreeMap<OwnedEventId, OwnedUserId>> {
+        Ok(self
+            .store
+            .get_kv_data(StateStoreDataKey::SeenKnockRequests(self.room_id()))
+            .await?
+            .and_then(|v| v.into_seen_knock_requests())
+            .unwrap_or_default())
+    }
 }
 
 // See https://github.com/matrix-org/matrix-rust-sdk/pull/3749#issuecomment-2312939823.
@@ -1348,6 +1438,11 @@ impl RoomInfo {
         self.members_synced = false;
     }
 
+    /// Returns whether the room members are synced.
+    pub fn are_members_synced(&self) -> bool {
+        self.members_synced
+    }
+
     /// Mark this Room as still missing some state information.
     pub fn mark_state_partially_synced(&mut self) {
         self.sync_info = SyncInfo::PartiallySynced;

crates/matrix-sdk-base/src/store/memory_store.rs

@@ -82,6 +82,7 @@ struct MemoryStoreInner {
     custom: HashMap<Vec<u8>, Vec<u8>>,
     send_queue_events: BTreeMap<OwnedRoomId, Vec<QueuedRequest>>,
     dependent_send_queue_events: BTreeMap<OwnedRoomId, Vec<DependentQueuedRequest>>,
+    seen_knock_requests: BTreeMap<OwnedRoomId, BTreeMap<OwnedEventId, OwnedUserId>>,
 }
 
 /// In-memory, non-persistent implementation of the `StateStore`.
@@ -168,6 +169,11 @@ impl StateStore for MemoryStore {
             StateStoreDataKey::ComposerDraft(room_id) => {
                 inner.composer_drafts.get(room_id).cloned().map(StateStoreDataValue::ComposerDraft)
             }
+            StateStoreDataKey::SeenKnockRequests(room_id) => inner
+                .seen_knock_requests
+                .get(room_id)
+                .cloned()
+                .map(StateStoreDataValue::SeenKnockRequests),
         })
     }
 
@@ -222,6 +228,14 @@ impl StateStore for MemoryStore {
                         .expect("Session data not containing server capabilities"),
                 );
             }
+            StateStoreDataKey::SeenKnockRequests(room_id) => {
+                inner.seen_knock_requests.insert(
+                    room_id.to_owned(),
+                    value
+                        .into_seen_knock_requests()
+                        .expect("Session data is not a set of seen join request ids"),
+                );
+            }
         }
 
         Ok(())
@@ -245,6 +259,9 @@ impl StateStore for MemoryStore {
             StateStoreDataKey::ComposerDraft(room_id) => {
                 inner.composer_drafts.remove(room_id);
             }
+            StateStoreDataKey::SeenKnockRequests(room_id) => {
+                inner.seen_knock_requests.remove(room_id);
+            }
         }
         Ok(())
     }