feat(sdk): Use state_after in sync v2 (#5488)

It is supposed to be an improvement over `state`, since it allows the
server to send updates to the state that might not be reflected in the
timeline.

This is also the same behavior as in Simplified Sliding Sync.

This is MSC4222 that was accepted and is about to get merged in the
spec.

---------

Signed-off-by: Kévin Commaille <[email protected]>

Author: zecakeh

Cargo.toml

@@ -74,6 +74,7 @@ ruma = { git = "https://github.com/ruma/ruma", rev = "a626e6b8521bcaa01bf8b7c416
     "unstable-msc4140",
     "unstable-msc4143",
     "unstable-msc4171",
+    "unstable-msc4222",
     "unstable-msc4278",
     "unstable-msc4286",
     "unstable-msc4306"

crates/matrix-sdk-base/CHANGELOG.md

@@ -7,6 +7,10 @@ All notable changes to this project will be documented in this file.
 ## [Unreleased] - ReleaseDate
 
 ### Features
+- [**breaking**] The `state` field of `JoinedRoomUpdate` and `LeftRoomUpdate`
+  now uses the `State` enum, depending on whether the state changes were
+  received in the `state` field or the `state_after` field.
+  ([#5488](https://github.com/matrix-org/matrix-rust-sdk/pull/5488))
 - [**breaking**] `RoomCreateWithCreatorEventContent` has a new field
   `additional_creators` that allows to specify additional room creators beside
   the user sending the `m.room.create` event, introduced with room version 12.

crates/matrix-sdk-base/src/response_processors/room/msc4186/mod.rs

@@ -48,7 +48,7 @@ use crate::{
     Result, Room, RoomHero, RoomInfo, RoomInfoNotableUpdate, RoomInfoNotableUpdateReasons,
     RoomState,
     store::BaseStateStore,
-    sync::{InvitedRoomUpdate, JoinedRoomUpdate, KnockedRoomUpdate, LeftRoomUpdate},
+    sync::{InvitedRoomUpdate, JoinedRoomUpdate, KnockedRoomUpdate, LeftRoomUpdate, State},
 };
 
 /// Represent any kind of room updates.
@@ -80,8 +80,8 @@ pub async fn update_any_room(
     // Don't read state events from the `timeline` field, because they might be
     // incomplete or staled already. We must only read state events from
     // `required_state`.
-    let (raw_state_events, state_events) =
-        state_events::sync::collect(&room_response.required_state);
+    let state = State::from_msc4186(room_response.required_state.clone());
+    let (raw_state_events, state_events) = state.collect(&[]);
 
     let state_store = notification.state_store;
 
@@ -190,7 +190,7 @@ pub async fn update_any_room(
                 room_info,
                 RoomUpdateKind::Joined(JoinedRoomUpdate::new(
                     timeline,
-                    raw_state_events,
+                    state,
                     room_account_data.cloned().unwrap_or_default(),
                     ephemeral,
                     notification_count,
@@ -203,7 +203,7 @@ pub async fn update_any_room(
             room_info,
             RoomUpdateKind::Left(LeftRoomUpdate::new(
                 timeline,
-                raw_state_events,
+                state,
                 room_account_data.cloned().unwrap_or_default(),
                 ambiguity_changes,
             )),
@@ -552,3 +552,11 @@ pub(crate) async fn cache_latest_events(
     // the latest is last
     room.latest_encrypted_events.write().unwrap().extend(encrypted_events.into_iter().rev());
 }
+
+impl State {
+    /// Construct a [`State`] from the state changes for a joined or left room
+    /// from a response of the Simplified Sliding Sync endpoint.
+    fn from_msc4186(events: Vec<Raw<AnySyncStateEvent>>) -> Self {
+        Self::After(events)
+    }
+}

crates/matrix-sdk-base/src/response_processors/room/sync_v2.rs

@@ -16,7 +16,9 @@ use std::collections::{BTreeMap, BTreeSet};
 
 use ruma::{
     OwnedRoomId, OwnedUserId, RoomId,
-    api::client::sync::sync_events::v3::{InvitedRoom, JoinedRoom, KnockedRoom, LeftRoom, State},
+    api::client::sync::sync_events::v3::{
+        InvitedRoom, JoinedRoom, KnockedRoom, LeftRoom, State as RumaState,
+    },
 };
 use tokio::sync::broadcast::Sender;
 use tracing::error;
@@ -29,7 +31,7 @@ use super::{
 };
 use crate::{
     Result, RoomInfoNotableUpdate, RoomState,
-    sync::{InvitedRoomUpdate, JoinedRoomUpdate, KnockedRoomUpdate, LeftRoomUpdate},
+    sync::{InvitedRoomUpdate, JoinedRoomUpdate, KnockedRoomUpdate, LeftRoomUpdate, State},
 };
 
 /// Process updates of a joined room.
@@ -64,39 +66,18 @@ pub async fn update_joined_room(
 
     let mut new_user_ids = BTreeSet::new();
 
-    let state = match joined_room.state {
-        State::Before(state) => {
-            let (raw_state_events, state_events) = state_events::sync::collect(&state.events);
-            state_events::sync::dispatch(
-                context,
-                (&raw_state_events, &state_events),
-                &mut room_info,
-                ambiguity_cache,
-                &mut new_user_ids,
-                state_store,
-            )
-            .await?;
-
-            let (raw_state_events_from_timeline, state_events_from_timeline) =
-                state_events::sync::collect_from_timeline(&joined_room.timeline.events);
-            state_events::sync::dispatch(
-                context,
-                (&raw_state_events_from_timeline, &state_events_from_timeline),
-                &mut room_info,
-                ambiguity_cache,
-                &mut new_user_ids,
-                state_store,
-            )
-            .await?;
-
-            state
-        }
-        // We shouldn't receive other variants because they are opt-in.
-        state => {
-            error!("Unsupported State variant received for joined room: {state:?}");
-            Default::default()
-        }
-    };
+    let state = State::from_sync_v2(joined_room.state);
+    let (raw_state_events, state_events) = state.collect(&joined_room.timeline.events);
+
+    state_events::sync::dispatch(
+        context,
+        (&raw_state_events, &state_events),
+        &mut room_info,
+        ambiguity_cache,
+        &mut new_user_ids,
+        state_store,
+    )
+    .await?;
 
     ephemeral_events::dispatch(context, &joined_room.ephemeral.events, room_id);
 
@@ -155,7 +136,7 @@ pub async fn update_joined_room(
 
     Ok(JoinedRoomUpdate::new(
         timeline,
-        state.events,
+        state,
         joined_room.account_data.events,
         joined_room.ephemeral.events,
         notification_count,
@@ -189,39 +170,18 @@ pub async fn update_left_room(
     room_info.mark_state_partially_synced();
     room_info.handle_encryption_state(requested_required_states.for_room(room_id));
 
-    let state = match left_room.state {
-        State::Before(state) => {
-            let (raw_state_events, state_events) = state_events::sync::collect(&state.events);
-            state_events::sync::dispatch(
-                context,
-                (&raw_state_events, &state_events),
-                &mut room_info,
-                ambiguity_cache,
-                &mut (),
-                state_store,
-            )
-            .await?;
-
-            let (raw_state_events_from_timeline, state_events_from_timeline) =
-                state_events::sync::collect_from_timeline(&left_room.timeline.events);
-            state_events::sync::dispatch(
-                context,
-                (&raw_state_events_from_timeline, &state_events_from_timeline),
-                &mut room_info,
-                ambiguity_cache,
-                &mut (),
-                state_store,
-            )
-            .await?;
-
-            state
-        }
-        // We shouldn't receive other variants because they are opt-in.
-        state => {
-            error!("Unsupported State variant received for left room: {state:?}");
-            Default::default()
-        }
-    };
+    let state = State::from_sync_v2(left_room.state);
+    let (raw_state_events, state_events) = state.collect(&left_room.timeline.events);
+
+    state_events::sync::dispatch(
+        context,
+        (&raw_state_events, &state_events),
+        &mut room_info,
+        ambiguity_cache,
+        &mut (),
+        state_store,
+    )
+    .await?;
 
     let timeline = timeline::build(
         context,
@@ -241,12 +201,7 @@ pub async fn update_left_room(
 
     let ambiguity_changes = ambiguity_cache.changes.remove(room_id).unwrap_or_default();
 
-    Ok(LeftRoomUpdate::new(
-        timeline,
-        state.events,
-        left_room.account_data.events,
-        ambiguity_changes,
-    ))
+    Ok(LeftRoomUpdate::new(timeline, state, left_room.account_data.events, ambiguity_changes))
 }
 
 /// Process updates of an invited room.
@@ -320,3 +275,19 @@ pub async fn update_knocked_room(
 
     Ok(knocked_room)
 }
+
+impl State {
+    /// Construct a [`State`] from the state changes for a joined or left room
+    /// from a response of the sync v2 endpoint.
+    fn from_sync_v2(state: RumaState) -> Self {
+        match state {
+            RumaState::Before(state) => Self::Before(state.events),
+            RumaState::After(state) => Self::After(state.events),
+            // We shouldn't receive other variants because they are opt-in.
+            state => {
+                error!("Unsupported State variant received for joined room: {state:?}");
+                Self::default()
+            }
+        }
+    }
+}

crates/matrix-sdk-base/src/response_processors/state_events.rs

@@ -45,29 +45,31 @@ pub mod sync {
     use crate::{
         RoomInfo,
         store::{BaseStateStore, Result as StoreResult, ambiguity_map::AmbiguityCache},
+        sync::State,
     };
 
-    /// Collect [`AnySyncStateEvent`] to [`AnySyncStateEvent`].
-    pub fn collect(
-        raw_events: &[Raw<AnySyncStateEvent>],
-    ) -> (Vec<Raw<AnySyncStateEvent>>, Vec<AnySyncStateEvent>) {
-        super::collect(raw_events)
-    }
-
-    /// Collect [`AnySyncTimelineEvent`] to [`AnySyncStateEvent`].
-    ///
-    /// A [`AnySyncTimelineEvent`] can represent either message-like events or
-    /// state events. The message-like events are filtered out.
-    pub fn collect_from_timeline(
-        raw_events: &[Raw<AnySyncTimelineEvent>],
-    ) -> (Vec<Raw<AnySyncStateEvent>>, Vec<AnySyncStateEvent>) {
-        super::collect(raw_events.iter().filter_map(|raw_event| {
-            // Only state events have a `state_key` field.
-            match raw_event.get_field::<&str>("state_key") {
-                Ok(Some(_)) => Some(raw_event.cast_ref_unchecked()),
-                _ => None,
+    impl State {
+        /// Collect all the state changes to update the local state, from this
+        /// [`State`] and from the given timeline, if necessary.
+        ///
+        /// The events that fail to deserialize are logged and filtered out.
+        pub(crate) fn collect(
+            &self,
+            timeline: &[Raw<AnySyncTimelineEvent>],
+        ) -> (Vec<Raw<AnySyncStateEvent>>, Vec<AnySyncStateEvent>) {
+            match self {
+                Self::Before(events) => {
+                    super::collect(events.iter().chain(timeline.iter().filter_map(|raw_event| {
+                        // Only state events have a `state_key` field.
+                        match raw_event.get_field::<&str>("state_key") {
+                            Ok(Some(_)) => Some(raw_event.cast_ref_unchecked()),
+                            _ => None,
+                        }
+                    })))
+                }
+                Self::After(events) => super::collect(events),
             }
-        }))
+        }
     }
 
     /// Dispatch the sync state events.