refactor: Fix errors in api/client/sync

src/api/client/sync/v3/joined.rs

@@ -23,17 +23,21 @@ use ruma::{
 	OwnedRoomId, OwnedUserId, RoomId, UserId,
 	api::client::sync::sync_events::{
 		UnreadNotificationsCount,
-		v3::{Ephemeral, JoinedRoom, RoomAccountData, RoomSummary, State as RoomState, Timeline},
+		v3::{
+			Ephemeral, JoinedRoom, RoomAccountData, RoomSummary, State as RoomState, StateEvents,
+			Timeline,
+		},
 	},
+	assign,
 	events::{
-		AnyRawAccountDataEvent, StateEventType,
+		StateEventType,
 		TimelineEventType::*,
 		room::member::{MembershipState, RoomMemberEventContent},
 	},
 	serde::Raw,
 	uint,
 };
-use service::rooms::short::ShortStateHash;
+use service::{account_data::AnyRawAccountDataEvent, rooms::short::ShortStateHash};
 
 use super::{load_timeline, share_encrypted_room};
 use crate::client::{
@@ -92,17 +96,15 @@ pub(super) async fn load_joined_room(
 		);
 	}
 
-	let joined_room = JoinedRoom {
+	let joined_room = assign!(JoinedRoom::new(), {
 		account_data,
 		summary: summary.unwrap_or_default(),
 		unread_notifications: notification_counts.unwrap_or_default(),
 		timeline,
-		state: RoomState {
-			events: state_events.into_iter().map(Event::into_format).collect(),
-		},
+		state: RoomState::Before(StateEvents::with_events(state_events.into_iter().map(Event::into_format).collect())),
 		ephemeral,
 		unread_thread_notifications: BTreeMap::new(),
-	};
+	});
 
 	Ok((joined_room, device_list_updates))
 }
@@ -126,7 +128,7 @@ async fn build_account_data(
 		.collect()
 		.await;
 
-	Ok(RoomAccountData { events: account_data_changes })
+	Ok(assign!(RoomAccountData::new(), { events: account_data_changes }))
 }
 
 /// Collect new ephemeral events.
@@ -233,7 +235,7 @@ async fn build_ephemeral(
 	edus.extend(typing_event);
 	edus.extend(private_read_event);
 
-	Ok(Ephemeral { events: edus })
+	Ok(assign!(Ephemeral::new(), { events: edus }))
 }
 
 /// A struct to hold the state events, timeline, and other data which is
@@ -318,11 +320,11 @@ async fn build_state_and_timeline(
 
 	Ok(StateAndTimeline {
 		state_events,
-		timeline: Timeline {
+		timeline: assign!(Timeline::new(), {
 			limited,
 			prev_batch: prev_batch.as_ref().map(ToString::to_string),
 			events: filtered_timeline,
-		},
+		}),
 		summary,
 		notification_counts,
 		device_list_updates,
@@ -580,10 +582,10 @@ async fn build_notification_counts(
 
 		trace!(%notification_count, %highlight_count, "syncing new notification counts");
 
-		Ok(Some(UnreadNotificationsCount {
+		Ok(Some(assign!(UnreadNotificationsCount::new(), {
 			notification_count: Some(notification_count),
 			highlight_count: Some(highlight_count),
-		}))
+		})))
 	} else {
 		Ok(None)
 	}
@@ -698,13 +700,13 @@ async fn build_room_summary(
 		"syncing updated summary"
 	);
 
-	Ok(Some(RoomSummary {
+	Ok(Some(assign!(RoomSummary::new(), {
 		heroes: heroes
 			.map(|heroes| heroes.into_iter().collect())
 			.unwrap_or_default(),
 		joined_member_count: Some(ruma_from_u64(joined_member_count)),
 		invited_member_count: Some(ruma_from_u64(invited_member_count)),
-	}))
+	})))
 }
 
 /// Fetch the user IDs to include in the `m.heroes` property of the room
@@ -718,18 +720,10 @@ async fn build_heroes(
 	const MAX_HERO_COUNT: usize = 5;
 
 	// fetch joined members from the state cache first
-	let joined_members_stream = services
-		.rooms
-		.state_cache
-		.room_members(room_id)
-		.map(ToOwned::to_owned);
+	let joined_members_stream = services.rooms.state_cache.room_members(room_id);
 
 	// then fetch invited members
-	let invited_members_stream = services
-		.rooms
-		.state_cache
-		.room_members_invited(room_id)
-		.map(ToOwned::to_owned);
+	let invited_members_stream = services.rooms.state_cache.room_members_invited(room_id);
 
 	// then as a last resort fetch every membership event
 	let all_members_stream = services
@@ -796,7 +790,6 @@ async fn build_device_list_updates(
 		.users
 		.room_keys_changed(room_id, last_sync_end_count, Some(current_count))
 		.map(at!(0))
-		.map(ToOwned::to_owned)
 		.ready_for_each(|user_id| {
 			device_list_updates.changed.insert(user_id);
 		})

src/api/client/sync/v3/left.rs

@@ -7,7 +7,10 @@ use conduwuit::{
 use futures::{StreamExt, future::join};
 use ruma::{
 	EventId, OwnedRoomId, RoomId,
-	api::client::sync::sync_events::v3::{LeftRoom, RoomAccountData, State, Timeline},
+	api::client::sync::sync_events::v3::{
+		LeftRoom, RoomAccountData, State, StateEvents, Timeline,
+	},
+	assign,
 	events::{StateEventType, TimelineEventType},
 	uint,
 };
@@ -178,17 +181,15 @@ pub(super) async fn load_left_room(
 		.collect::<Vec<_>>()
 		.await;
 
-	Ok(Some(LeftRoom {
-		account_data: RoomAccountData { events: Vec::new() },
-		timeline: Timeline {
+	Ok(Some(assign!(LeftRoom::new(), {
+		account_data: RoomAccountData::new(),
+		timeline: assign!(Timeline::new(), {
 			limited: timeline.limited,
 			prev_batch: Some(current_count.to_string()),
 			events: raw_timeline_pdus,
-		},
-		state: State {
-			events: state_events.into_iter().map(Event::into_format).collect(),
-		},
-	}))
+		}),
+		state: State::Before(StateEvents::with_events(state_events.into_iter().map(Event::into_format).collect())),
+	})))
 }
 
 async fn build_left_state_and_timeline(
@@ -317,7 +318,7 @@ fn create_dummy_leave_event(
 	// clients. perhaps a database table could be created to hold these dummy
 	// events, or they could be stored as outliers?
 	PduEvent {
-		event_id: EventId::new(services.globals.server_name()),
+		event_id: EventId::new_v1(services.globals.server_name()),
 		sender: syncing_user.to_owned(),
 		origin: None,
 		origin_server_ts: utils::millis_since_unix_epoch()

src/api/client/sync/v3/mod.rs

@@ -11,7 +11,7 @@ use std::{
 use axum::extract::State;
 use axum_client_ip::ClientIp;
 use conduwuit::{
-	Result, at, extract_variant,
+	Err, Result, at, extract_variant,
 	utils::{
 		ReadyExt, TryFutureExtExt,
 		stream::{BroadbandExt, Tools, WidebandExt},
@@ -19,10 +19,7 @@ use conduwuit::{
 	warn,
 };
 use conduwuit_service::Services;
-use futures::{
-	FutureExt, StreamExt, TryFutureExt,
-	future::{OptionFuture, join3, join4, join5},
-};
+use futures::{FutureExt, StreamExt, TryFutureExt, future::OptionFuture};
 use ruma::{
 	DeviceId, OwnedUserId, RoomId, UserId,
 	api::client::{
@@ -36,13 +33,14 @@ use ruma::{
 		},
 		uiaa::UiaaResponse,
 	},
-	events::{
-		AnyRawAccountDataEvent,
-		presence::{PresenceEvent, PresenceEventContent},
-	},
+	assign,
+	events::presence::{PresenceEvent, PresenceEventContent},
 	serde::Raw,
 };
-use service::rooms::lazy_loading::{self, MemberSet, Options as _};
+use service::{
+	account_data::AnyRawAccountDataEvent,
+	rooms::lazy_loading::{self, MemberSet, Options as _},
+};
 
 use super::{load_timeline, share_encrypted_room};
 use crate::{
@@ -82,10 +80,10 @@ impl DeviceListUpdates {
 
 impl From<DeviceListUpdates> for DeviceLists {
 	fn from(val: DeviceListUpdates) -> Self {
-		Self {
+		assign!(Self::new(), {
 			changed: val.changed.into_iter().collect(),
 			left: val.left.into_iter().collect(),
-		}
+		})
 	}
 }
 
@@ -253,6 +251,8 @@ pub(crate) async fn build_sync_events(
 			.get_filter(syncing_user, filter_id)
 			.await
 			.unwrap_or_default(),
+		// error out for unknown filter types
+		| _ => return Err!(Request(InvalidParam("Unknown filter type"))),
 	});
 
 	let context = SyncContext {
@@ -268,7 +268,6 @@ pub(crate) async fn build_sync_events(
 		.rooms
 		.state_cache
 		.rooms_joined(syncing_user)
-		.map(ToOwned::to_owned)
 		.broad_filter_map(|room_id| async {
 			let joined_room = load_joined_room(services, context, room_id.clone()).await;
 
@@ -332,9 +331,9 @@ pub(crate) async fn build_sync_events(
 
 			// only sync this invite if it was sent after the last /sync call
 			if last_sync_end_count < invite_count {
-				let invited_room = InvitedRoom {
-					invite_state: InviteState { events: invite_state },
-				};
+				let invited_room = assign!(InvitedRoom::new(), {
+					invite_state: InviteState::from(invite_state),
+				});
 
 				invited_rooms.insert(room_id, invited_room);
 			}
@@ -355,9 +354,9 @@ pub(crate) async fn build_sync_events(
 
 			// only sync this knock if it was sent after the last /sync call
 			if last_sync_end_count < knock_count {
-				let knocked_room = KnockedRoom {
-					knock_state: KnockState { events: knock_state },
-				};
+				let knocked_room = assign!(KnockedRoom::new(), {
+					knock_state: assign!(KnockState::new(), { events: knock_state }),
+				});
 
 				knocked_rooms.insert(room_id, knocked_room);
 			}
@@ -376,13 +375,6 @@ pub(crate) async fn build_sync_events(
 		.ready_filter_map(|e| extract_variant!(e, AnyRawAccountDataEvent::Global))
 		.collect();
 
-	// Look for device list updates of this account
-	let keys_changed = services
-		.users
-		.keys_changed(syncing_user, last_sync_end_count, Some(current_count))
-		.map(ToOwned::to_owned)
-		.collect::<HashSet<_>>();
-
 	let to_device_events = services
 		.users
 		.get_to_device_events(
@@ -394,36 +386,57 @@ pub(crate) async fn build_sync_events(
 		.map(at!(1))
 		.collect::<Vec<_>>();
 
+	// Look for device list updates of this account
+	let keys_changed = services
+		.users
+		.keys_changed(syncing_user, last_sync_end_count, Some(current_count))
+		.collect::<HashSet<_>>();
+
 	let device_one_time_keys_count = services
 		.users
 		.count_one_time_keys(syncing_user, syncing_device);
 
-	// Remove all to-device events the device received *last time*
-	let remove_to_device_events =
-		services
-			.users
-			.remove_to_device_events(syncing_user, syncing_device, last_sync_end_count);
+	let (
+		(joined_rooms, mut device_list_updates),
+		left_rooms,
+		invited_rooms,
+		knocked_rooms,
+		presence_updates,
+		account_data,
+		to_device_events,
+		keys_changed,
+		device_one_time_keys_count,
+	) = async {
+		futures::join!(
+			joined_rooms,
+			left_rooms,
+			invited_rooms,
+			knocked_rooms,
+			presence_updates,
+			account_data,
+			to_device_events,
+			keys_changed,
+			device_one_time_keys_count
+		)
+	}
+	.boxed()
+	.await;
 
-	let rooms = join4(joined_rooms, left_rooms, invited_rooms, knocked_rooms);
-	let ephemeral = join3(remove_to_device_events, to_device_events, presence_updates);
-	let top = join5(account_data, ephemeral, device_one_time_keys_count, keys_changed, rooms)
-		.boxed()
+	// Remove all to-device events the device received *last time*
+	services
+		.users
+		.remove_to_device_events(syncing_user, syncing_device, last_sync_end_count)
 		.await;
 
-	let (account_data, ephemeral, device_one_time_keys_count, keys_changed, rooms) = top;
-	let ((), to_device_events, presence_updates) = ephemeral;
-	let (joined_rooms, left_rooms, invited_rooms, knocked_rooms) = rooms;
-	let (joined_rooms, mut device_list_updates) = joined_rooms;
 	device_list_updates.changed.extend(keys_changed);
 
-	let response = sync_events::v3::Response {
-		account_data: GlobalAccountData { events: account_data },
+	let response = assign!(sync_events::v3::Response::new(current_count.to_string()), {
+		account_data: assign!(GlobalAccountData::new(), { events: account_data }),
 		device_lists: device_list_updates.into(),
 		device_one_time_keys_count,
 		// Fallback keys are not yet supported
 		device_unused_fallback_key_types: None,
-		next_batch: current_count.to_string(),
-		presence: Presence {
+		presence: assign!(Presence::new(), {
 			events: presence_updates
 				.into_iter()
 				.flat_map(IntoIterator::into_iter)
@@ -431,15 +444,15 @@ pub(crate) async fn build_sync_events(
 				.map(|ref event| Raw::new(event))
 				.filter_map(Result::ok)
 				.collect(),
-		},
-		rooms: Rooms {
+		}),
+		rooms: assign!(Rooms::new(), {
 			leave: left_rooms,
 			join: joined_rooms,
 			invite: invited_rooms,
 			knock: knocked_rooms,
-		},
-		to_device: ToDevice { events: to_device_events },
-	};
+		}),
+		to_device: assign!(ToDevice::new(), { events: to_device_events }),
+	});
 
 	Ok(response)
 }