syncv3: support per room account data

Cargo.lock

@@ -2975,7 +2975,7 @@ dependencies = [
 [[package]]
 name = "ruma"
 version = "0.10.1"
-source = "git+https://github.com/girlbossceo/ruwuma?rev=1de0f493e8aab7e65ea78e3a079a3de10167c777#1de0f493e8aab7e65ea78e3a079a3de10167c777"
+source = "git+https://github.com/girlbossceo/ruwuma?rev=d7ddcd036f81edb257ab9371f9cadd46444e8a90#d7ddcd036f81edb257ab9371f9cadd46444e8a90"
 dependencies = [
  "assign",
  "js_int",
@@ -2997,7 +2997,7 @@ dependencies = [
 [[package]]
 name = "ruma-appservice-api"
 version = "0.10.0"
-source = "git+https://github.com/girlbossceo/ruwuma?rev=1de0f493e8aab7e65ea78e3a079a3de10167c777#1de0f493e8aab7e65ea78e3a079a3de10167c777"
+source = "git+https://github.com/girlbossceo/ruwuma?rev=d7ddcd036f81edb257ab9371f9cadd46444e8a90#d7ddcd036f81edb257ab9371f9cadd46444e8a90"
 dependencies = [
  "js_int",
  "ruma-common",
@@ -3009,7 +3009,7 @@ dependencies = [
 [[package]]
 name = "ruma-client-api"
 version = "0.18.0"
-source = "git+https://github.com/girlbossceo/ruwuma?rev=1de0f493e8aab7e65ea78e3a079a3de10167c777#1de0f493e8aab7e65ea78e3a079a3de10167c777"
+source = "git+https://github.com/girlbossceo/ruwuma?rev=d7ddcd036f81edb257ab9371f9cadd46444e8a90#d7ddcd036f81edb257ab9371f9cadd46444e8a90"
 dependencies = [
  "as_variant",
  "assign",
@@ -3032,7 +3032,7 @@ dependencies = [
 [[package]]
 name = "ruma-common"
 version = "0.13.0"
-source = "git+https://github.com/girlbossceo/ruwuma?rev=1de0f493e8aab7e65ea78e3a079a3de10167c777#1de0f493e8aab7e65ea78e3a079a3de10167c777"
+source = "git+https://github.com/girlbossceo/ruwuma?rev=d7ddcd036f81edb257ab9371f9cadd46444e8a90#d7ddcd036f81edb257ab9371f9cadd46444e8a90"
 dependencies = [
  "as_variant",
  "base64 0.22.1",
@@ -3062,7 +3062,7 @@ dependencies = [
 [[package]]
 name = "ruma-events"
 version = "0.28.1"
-source = "git+https://github.com/girlbossceo/ruwuma?rev=1de0f493e8aab7e65ea78e3a079a3de10167c777#1de0f493e8aab7e65ea78e3a079a3de10167c777"
+source = "git+https://github.com/girlbossceo/ruwuma?rev=d7ddcd036f81edb257ab9371f9cadd46444e8a90#d7ddcd036f81edb257ab9371f9cadd46444e8a90"
 dependencies = [
  "as_variant",
  "indexmap 2.4.0",
@@ -3086,7 +3086,7 @@ dependencies = [
 [[package]]
 name = "ruma-federation-api"
 version = "0.9.0"
-source = "git+https://github.com/girlbossceo/ruwuma?rev=1de0f493e8aab7e65ea78e3a079a3de10167c777#1de0f493e8aab7e65ea78e3a079a3de10167c777"
+source = "git+https://github.com/girlbossceo/ruwuma?rev=d7ddcd036f81edb257ab9371f9cadd46444e8a90#d7ddcd036f81edb257ab9371f9cadd46444e8a90"
 dependencies = [
  "bytes",
  "http",
@@ -3104,7 +3104,7 @@ dependencies = [
 [[package]]
 name = "ruma-identifiers-validation"
 version = "0.9.5"
-source = "git+https://github.com/girlbossceo/ruwuma?rev=1de0f493e8aab7e65ea78e3a079a3de10167c777#1de0f493e8aab7e65ea78e3a079a3de10167c777"
+source = "git+https://github.com/girlbossceo/ruwuma?rev=d7ddcd036f81edb257ab9371f9cadd46444e8a90#d7ddcd036f81edb257ab9371f9cadd46444e8a90"
 dependencies = [
  "js_int",
  "thiserror",
@@ -3113,7 +3113,7 @@ dependencies = [
 [[package]]
 name = "ruma-identity-service-api"
 version = "0.9.0"
-source = "git+https://github.com/girlbossceo/ruwuma?rev=1de0f493e8aab7e65ea78e3a079a3de10167c777#1de0f493e8aab7e65ea78e3a079a3de10167c777"
+source = "git+https://github.com/girlbossceo/ruwuma?rev=d7ddcd036f81edb257ab9371f9cadd46444e8a90#d7ddcd036f81edb257ab9371f9cadd46444e8a90"
 dependencies = [
  "js_int",
  "ruma-common",
@@ -3123,7 +3123,7 @@ dependencies = [
 [[package]]
 name = "ruma-macros"
 version = "0.13.0"
-source = "git+https://github.com/girlbossceo/ruwuma?rev=1de0f493e8aab7e65ea78e3a079a3de10167c777#1de0f493e8aab7e65ea78e3a079a3de10167c777"
+source = "git+https://github.com/girlbossceo/ruwuma?rev=d7ddcd036f81edb257ab9371f9cadd46444e8a90#d7ddcd036f81edb257ab9371f9cadd46444e8a90"
 dependencies = [
  "once_cell",
  "proc-macro-crate",
@@ -3138,7 +3138,7 @@ dependencies = [
 [[package]]
 name = "ruma-push-gateway-api"
 version = "0.9.0"
-source = "git+https://github.com/girlbossceo/ruwuma?rev=1de0f493e8aab7e65ea78e3a079a3de10167c777#1de0f493e8aab7e65ea78e3a079a3de10167c777"
+source = "git+https://github.com/girlbossceo/ruwuma?rev=d7ddcd036f81edb257ab9371f9cadd46444e8a90#d7ddcd036f81edb257ab9371f9cadd46444e8a90"
 dependencies = [
  "js_int",
  "ruma-common",
@@ -3150,7 +3150,7 @@ dependencies = [
 [[package]]
 name = "ruma-server-util"
 version = "0.3.0"
-source = "git+https://github.com/girlbossceo/ruwuma?rev=1de0f493e8aab7e65ea78e3a079a3de10167c777#1de0f493e8aab7e65ea78e3a079a3de10167c777"
+source = "git+https://github.com/girlbossceo/ruwuma?rev=d7ddcd036f81edb257ab9371f9cadd46444e8a90#d7ddcd036f81edb257ab9371f9cadd46444e8a90"
 dependencies = [
  "headers",
  "http",
@@ -3163,7 +3163,7 @@ dependencies = [
 [[package]]
 name = "ruma-signatures"
 version = "0.15.0"
-source = "git+https://github.com/girlbossceo/ruwuma?rev=1de0f493e8aab7e65ea78e3a079a3de10167c777#1de0f493e8aab7e65ea78e3a079a3de10167c777"
+source = "git+https://github.com/girlbossceo/ruwuma?rev=d7ddcd036f81edb257ab9371f9cadd46444e8a90#d7ddcd036f81edb257ab9371f9cadd46444e8a90"
 dependencies = [
  "base64 0.22.1",
  "ed25519-dalek",
@@ -3179,7 +3179,7 @@ dependencies = [
 [[package]]
 name = "ruma-state-res"
 version = "0.11.0"
-source = "git+https://github.com/girlbossceo/ruwuma?rev=1de0f493e8aab7e65ea78e3a079a3de10167c777#1de0f493e8aab7e65ea78e3a079a3de10167c777"
+source = "git+https://github.com/girlbossceo/ruwuma?rev=d7ddcd036f81edb257ab9371f9cadd46444e8a90#d7ddcd036f81edb257ab9371f9cadd46444e8a90"
 dependencies = [
  "itertools 0.12.1",
  "js_int",

Cargo.toml

@@ -314,7 +314,7 @@ version = "0.1.2"
 [workspace.dependencies.ruma]
 git = "https://github.com/girlbossceo/ruwuma"
 #branch = "conduwuit-changes"
-rev = "1de0f493e8aab7e65ea78e3a079a3de10167c777"
+rev = "d7ddcd036f81edb257ab9371f9cadd46444e8a90"
 features = [
     "compat",
     "rand",

src/api/client/sync.rs

@@ -25,9 +25,16 @@ use ruma::{
 			DeviceLists, UnreadNotificationsCount,
 		},
 		uiaa::UiaaResponse,
-	}, events::{
+	},
-		presence::PresenceEvent, room::member::{MembershipState, RoomMemberEventContent}, StateEventType, TimelineEventType
+	events::{
-	}, room::RoomType, serde::Raw, state_res::Event, uint, DeviceId, EventId, MilliSecondsSinceUnixEpoch, OwnedRoomId, OwnedUserId, RoomId, UInt, UserId
+		presence::PresenceEvent,
+		room::member::{MembershipState, RoomMemberEventContent},
+		AnyRawAccountDataEvent, StateEventType, TimelineEventType,
+	},
+	room::RoomType,
+	serde::Raw,
+	state_res::Event,
+	uint, DeviceId, EventId, MilliSecondsSinceUnixEpoch, OwnedRoomId, OwnedUserId, RoomId, UInt, UserId,
 };
 use tracing::{Instrument as _, Span};
 
@@ -46,6 +53,15 @@ const DEFAULT_BUMP_TYPES: &[TimelineEventType] = &[
 	TimelineEventType::Beacon,
 ];
 
+macro_rules! extract_variant {
+	($e:expr, $variant:path) => {
+		match $e {
+			$variant(value) => Some(value),
+			_ => None,
+		}
+	};
+}
+
 /// # `GET /_matrix/client/r0/sync`
 ///
 /// Synchronize the client's state with the latest state on the server.
@@ -284,11 +300,7 @@ pub(crate) async fn sync_events_route(
 				.account_data
 				.changes_since(None, &sender_user, since)?
 				.into_iter()
-				.filter_map(|(_, v)| {
+				.filter_map(|e| extract_variant!(e, AnyRawAccountDataEvent::Global))
-					serde_json::from_str(v.json().get())
-						.map_err(|_| Error::bad_database("Invalid account event in database."))
-						.ok()
-				})
 				.collect(),
 		},
 		device_lists: DeviceLists {
@@ -981,11 +993,7 @@ async fn load_joined_room(
 				.account_data
 				.changes_since(Some(room_id), sender_user, since)?
 				.into_iter()
-				.filter_map(|(_, v)| {
+				.filter_map(|e| extract_variant!(e, AnyRawAccountDataEvent::Room))
-					serde_json::from_str(v.json().get())
-						.map_err(|_| Error::bad_database("Invalid account event in database."))
-						.ok()
-				})
 				.collect(),
 		},
 		summary: RoomSummary {
@@ -1160,6 +1168,33 @@ pub(crate) async fn sync_events_v4_route(
 	let mut device_list_changes = HashSet::new();
 	let mut device_list_left = HashSet::new();
 
+	let mut account_data = sync_events::v4::AccountData {
+		global: Vec::new(),
+		rooms: BTreeMap::new(),
+	};
+	if body.extensions.account_data.enabled.unwrap_or(false) {
+		account_data.global = services
+			.account_data
+			.changes_since(None, &sender_user, globalsince)?
+			.into_iter()
+			.filter_map(|e| extract_variant!(e, AnyRawAccountDataEvent::Global))
+			.collect();
+
+		if let Some(rooms) = body.extensions.account_data.rooms {
+			for room in rooms {
+				account_data.rooms.insert(
+					room.clone(),
+					services
+						.account_data
+						.changes_since(Some(&room), &sender_user, globalsince)?
+						.into_iter()
+						.filter_map(|e| extract_variant!(e, AnyRawAccountDataEvent::Room))
+						.collect(),
+				);
+			}
+		}
+	}
+
 	if body.extensions.e2ee.enabled.unwrap_or(false) {
 		// Look for device list updates of this account
 		device_list_changes.extend(
@@ -1464,7 +1499,17 @@ pub(crate) async fn sync_events_v4_route(
 		let (timeline_pdus, limited) =
 			load_timeline(&services, &sender_user, room_id, roomsincecount, *timeline_limit)?;
 
-		if roomsince != &0 && timeline_pdus.is_empty() {
+		account_data.rooms.insert(
+			room_id.clone(),
+			services
+				.account_data
+				.changes_since(Some(room_id), &sender_user, *roomsince)?
+				.into_iter()
+				.filter_map(|e| extract_variant!(e, AnyRawAccountDataEvent::Room))
+				.collect(),
+		);
+
+		if roomsince != &0 && timeline_pdus.is_empty() && account_data.rooms.get(room_id).is_some_and(Vec::is_empty) {
 			continue;
 		}
 
@@ -1504,7 +1549,7 @@ pub(crate) async fn sync_events_v4_route(
 
 		let mut timestamp: Option<_> = None;
 		for (_, pdu) in timeline_pdus {
-		    timestamp = Some(MilliSecondsSinceUnixEpoch(pdu.origin_server_ts));
+			timestamp = Some(MilliSecondsSinceUnixEpoch(pdu.origin_server_ts));
 			if DEFAULT_BUMP_TYPES.contains(pdu.event_type()) {
 				break;
 			}
@@ -1677,23 +1722,7 @@ pub(crate) async fn sync_events_v4_route(
 				// Fallback keys are not yet supported
 				device_unused_fallback_key_types: None,
 			},
-			account_data: sync_events::v4::AccountData {
+			account_data,
-				global: if body.extensions.account_data.enabled.unwrap_or(false) {
-					services
-						.account_data
-						.changes_since(None, &sender_user, globalsince)?
-						.into_iter()
-						.filter_map(|(_, v)| {
-							serde_json::from_str(v.json().get())
-								.map_err(|_| Error::bad_database("Invalid account event in database."))
-								.ok()
-						})
-						.collect()
-				} else {
-					Vec::new()
-				},
-				rooms: BTreeMap::new(),
-			},
 			receipts: sync_events::v4::Receipts {
 				rooms: BTreeMap::new(),
 			},

src/service/account_data/data.rs

@@ -1,10 +1,10 @@
 use std::{collections::HashMap, sync::Arc};
 
-use conduit::{utils, warn, Error, Result};
+use conduit::{Error, Result};
 use database::Map;
 use ruma::{
 	api::client::error::ErrorKind,
-	events::{AnyEphemeralRoomEvent, RoomAccountDataEventType},
+	events::{AnyGlobalAccountDataEvent, AnyRawAccountDataEvent, AnyRoomAccountDataEvent, RoomAccountDataEventType},
 	serde::Raw,
 	RoomId, UserId,
 };
@@ -110,7 +110,7 @@ impl Data {
 	/// Returns all changes to the account data that happened after `since`.
 	pub(super) fn changes_since(
 		&self, room_id: Option<&RoomId>, user_id: &UserId, since: u64,
-	) -> Result<HashMap<RoomAccountDataEventType, Raw<AnyEphemeralRoomEvent>>> {
+	) -> Result<Vec<AnyRawAccountDataEvent>> {
 		let mut userdata = HashMap::new();
 
 		let mut prefix = room_id
@@ -132,25 +132,21 @@ impl Data {
 			.take_while(move |(k, _)| k.starts_with(&prefix))
 			.map(|(k, v)| {
 				Ok::<_, Error>((
-					RoomAccountDataEventType::from(
+					k,
-						utils::string_from_bytes(
+					match room_id {
-							k.rsplit(|&b| b == 0xFF)
+						None => serde_json::from_slice::<Raw<AnyGlobalAccountDataEvent>>(&v)
-								.next()
+							.map(AnyRawAccountDataEvent::Global)
-								.ok_or_else(|| Error::bad_database("RoomUserData ID in db is invalid."))?,
+							.map_err(|_| Error::bad_database("Database contains invalid account data."))?,
-						)
+						Some(_) => serde_json::from_slice::<Raw<AnyRoomAccountDataEvent>>(&v)
-						.map_err(|e| {
+							.map(AnyRawAccountDataEvent::Room)
-							warn!("RoomUserData ID in database is invalid: {}", e);
+							.map_err(|_| Error::bad_database("Database contains invalid account data."))?,
-							Error::bad_database("RoomUserData ID in db is invalid.")
+					},
-						})?,
-					),
-					serde_json::from_slice::<Raw<AnyEphemeralRoomEvent>>(&v)
-						.map_err(|_| Error::bad_database("Database contains invalid account data."))?,
 				))
 			}) {
 			let (kind, data) = r?;
 			userdata.insert(kind, data);
 		}
 
-		Ok(userdata)
+		Ok(userdata.into_values().collect())
 	}
 }

src/service/account_data/mod.rs

@@ -1,12 +1,11 @@
 mod data;
 
-use std::{collections::HashMap, sync::Arc};
+use std::sync::Arc;
 
 use conduit::Result;
 use data::Data;
 use ruma::{
-	events::{AnyEphemeralRoomEvent, RoomAccountDataEventType},
+	events::{AnyRawAccountDataEvent, RoomAccountDataEventType},
-	serde::Raw,
 	RoomId, UserId,
 };
 
@@ -47,7 +46,7 @@ impl Service {
 	#[tracing::instrument(skip_all, name = "since", level = "debug")]
 	pub fn changes_since(
 		&self, room_id: Option<&RoomId>, user_id: &UserId, since: u64,
-	) -> Result<HashMap<RoomAccountDataEventType, Raw<AnyEphemeralRoomEvent>>> {
+	) -> Result<Vec<AnyRawAccountDataEvent>> {
 		self.db.changes_since(room_id, user_id, since)
 	}
 }