sdk-base: add Room::pinned_events(&self) to subscribe to pinned eve…

…nt ids in a room.

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

@@ -45,6 +45,7 @@ use ruma::{
         push_rules::{PushRulesEvent, PushRulesEventContent},
         room::{
             member::{MembershipState, RoomMemberEventContent, SyncRoomMemberEvent},
+            pinned_events::SyncRoomPinnedEventsEvent,
             power_levels::{
                 RoomPowerLevelsEvent, RoomPowerLevelsEventContent, StrippedRoomPowerLevelsEvent,
             },
@@ -549,6 +550,12 @@ impl BaseClient {
                 }
 
                 handle_room_member_event_for_profiles(&room_info.room_id, member, changes);
+            } else if let AnySyncStateEvent::RoomPinnedEvents(
+                SyncRoomPinnedEventsEvent::Original(ev),
+            ) = &event
+            {
+                let new_pinned_event_ids = ev.content.pinned.clone();
+                changes.pinned_events.insert(room_info.room_id.to_owned(), new_pinned_event_ids);
             }
 
             state_events
@@ -1173,6 +1180,15 @@ impl BaseClient {
             }
         }
 
+        for (room_id, pinned_event_ids) in changes.pinned_events.clone() {
+            if let Some(room) = self.store.room(&room_id) {
+                room.pinned_event_ids.update(|ids| {
+                    ids.clear();
+                    ids.append(&mut pinned_event_ids.clone());
+                });
+            }
+        }
+
         for (room_id, room_info) in &changes.room_infos {
             if let Some(room) = self.store.room(room_id) {
                 let room_info_notable_update_reasons =

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

@@ -132,6 +132,10 @@ 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>>>>,
+
+    /// The list of pinned event ids in this room in an observable format so
+    /// they can be subscribed to.
+    pub pinned_event_ids: SharedObservable<Vec<OwnedEventId>>,
 }
 
 /// The room summary containing member counts and members that should be used to
@@ -240,6 +244,7 @@ impl Room {
                 Self::MAX_ENCRYPTED_EVENTS,
             ))),
             room_info_notable_update_sender,
+            pinned_event_ids: SharedObservable::new(Vec::new()),
         }
     }
 
@@ -947,6 +952,11 @@ impl Room {
     pub fn recency_stamp(&self) -> Option<u64> {
         self.inner.read().recency_stamp
     }
+
+    /// Subscribe to the list of event ids for pinned events in this room.
+    pub fn pinned_events(&self) -> Subscriber<Vec<OwnedEventId>> {
+        self.pinned_event_ids.subscribe()
+    }
 }
 
 /// The underlying pure data structure for joined and left rooms.

crates/matrix-sdk-base/src/sliding_sync/mod.rs

@@ -844,6 +844,7 @@ mod tests {
     use std::{
         collections::{BTreeMap, HashSet},
         sync::{Arc, RwLock as SyncRwLock},
+        time::Duration,
     };
 
     use assert_matches::assert_matches;
@@ -860,15 +861,17 @@ mod tests {
                 member::{MembershipState, RoomMemberEventContent},
                 message::SyncRoomMessageEvent,
                 name::RoomNameEventContent,
+                pinned_events::RoomPinnedEventsEventContent,
             },
             AnySyncMessageLikeEvent, AnySyncTimelineEvent, GlobalAccountDataEventContent,
             StateEventContent,
         },
-        mxc_uri, owned_mxc_uri, owned_user_id, room_alias_id, room_id,
+        mxc_uri, owned_event_id, owned_mxc_uri, owned_user_id, room_alias_id, room_id,
         serde::Raw,
         uint, user_id, JsOption, MxcUri, OwnedRoomId, OwnedUserId, RoomAliasId, RoomId, UserId,
     };
     use serde_json::json;
+    use tokio::time::timeout;
 
     use super::{cache_latest_events, http};
     use crate::{
@@ -2178,6 +2181,59 @@ mod tests {
         );
     }
 
+    #[async_test]
+    async fn test_pinned_events_are_updated_on_sync() {
+        let user_a_id = user_id!("@a:e.uk");
+        let client = logged_in_base_client(Some(user_a_id)).await;
+        let room_id = room_id!("!r:e.uk");
+        let pinned_event_id = owned_event_id!("$an-id:e.uk");
+
+        // Create room
+        let mut room_response = http::response::Room::new();
+        set_room_joined(&mut room_response, user_a_id);
+        let response = response_with_room(room_id, room_response);
+        client.process_sliding_sync(&response, &(), true).await.expect("Failed to process sync");
+
+        // The newly created room has no pinned event ids
+        let room = client.get_room(room_id).unwrap();
+        let mut pinned_events_subscriber = room.pinned_events();
+        assert!(pinned_events_subscriber.get().is_empty());
+
+        // Load new pinned event id
+        let mut room_response = http::response::Room::new();
+        room_response.required_state.push(make_state_event(
+            user_a_id,
+            "",
+            RoomPinnedEventsEventContent::new(vec![pinned_event_id.clone()]),
+            None,
+        ));
+        let response = response_with_room(room_id, room_response);
+        client.process_sliding_sync(&response, &(), true).await.expect("Failed to process sync");
+
+        let pinned_event_ids = timeout(Duration::from_secs(1), pinned_events_subscriber.next())
+            .await
+            .unwrap()
+            .expect("No pinned event ids found");
+        assert_eq!(!pinned_event_ids.len(), 1);
+        assert_eq!(pinned_event_ids[0], pinned_event_id);
+
+        // Pinned event ids are now empty
+        let mut room_response = http::response::Room::new();
+        room_response.required_state.push(make_state_event(
+            user_a_id,
+            "",
+            RoomPinnedEventsEventContent::new(Vec::new()),
+            None,
+        ));
+        let response = response_with_room(room_id, room_response);
+        client.process_sliding_sync(&response, &(), true).await.expect("Failed to process sync");
+        let pinned_event_ids = timeout(Duration::from_secs(1), pinned_events_subscriber.next())
+            .await
+            .unwrap()
+            .expect("No pinned event ids found");
+        assert!(pinned_event_ids.is_empty());
+    }
+
     async fn choose_event_to_cache(events: &[SyncTimelineEvent]) -> Option<SyncTimelineEvent> {
         let room = make_room();
         let mut room_info = room.clone_info();

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

@@ -340,6 +340,9 @@ pub struct StateChanges {
     /// A map from room id to a map of a display name and a set of user ids that
     /// share that display name in the given room.
     pub ambiguity_maps: BTreeMap<OwnedRoomId, BTreeMap<String, BTreeSet<OwnedUserId>>>,
+
+    /// A map of `RoomId` to pinned `OwnedEventId`s for each room.
+    pub pinned_events: BTreeMap<OwnedRoomId, Vec<OwnedEventId>>,
 }
 
 impl StateChanges {

crates/matrix-sdk-sqlite/src/state_store.rs

@@ -1007,6 +1007,7 @@ impl StateStore for SqliteStateStore {
                     redactions,
                     stripped_state,
                     ambiguity_maps,
+                    pinned_events: _,
                 } = changes;
 
                 if let Some(sync_token) = sync_token {