fix(fast-joins): Use fast joins only for remote joins

src/admin/user/commands.rs

@@ -1,8 +1,8 @@
 use std::{collections::BTreeMap, fmt::Write as _};
 
 use api::client::{
-	full_user_deactivate, join_room_by_id_helper, leave_all_rooms, leave_room, update_avatar_url,
-	update_displayname,
+	full_user_deactivate, join_room_by_id_helper, leave_all_rooms, leave_room, remote_leave_room,
+	update_avatar_url, update_displayname,
 };
 use conduwuit::{
 	Err, Result, debug, debug_warn, error, info, is_equal_to,
@@ -924,3 +924,29 @@ pub(super) async fn redact_event(&self, event_id: OwnedEventId) -> Result {
 	))
 	.await
 }
+
+#[admin_command]
+pub(super) async fn force_leave_remote_room(
+	&self,
+	user_id: String,
+	room_id: OwnedRoomOrAliasId,
+) -> Result {
+	let user_id = parse_local_user_id(self.services, &user_id)?;
+	let (room_id, _) = self
+		.services
+		.rooms
+		.alias
+		.resolve_with_servers(&room_id, None)
+		.await?;
+
+	assert!(
+		self.services.globals.user_is_local(&user_id),
+		"Parsed user_id must be a local user"
+	);
+	remote_leave_room(self.services, &user_id, &room_id, None)
+		.boxed()
+		.await?;
+
+	self.write_str(&format!("{user_id} has been joined to {room_id}.",))
+		.await
+}

src/admin/user/mod.rs

@@ -103,6 +103,12 @@ pub enum UserCommand {
 		room_id: OwnedRoomOrAliasId,
 	},
 
+	/// - Manually leave a remote room for a local user.
+	ForceLeaveRemoteRoom {
+		user_id: String,
+		room_id: OwnedRoomOrAliasId,
+	},
+
 	/// - Forces the specified user to drop their power levels to the room
 	///   default, if their permissions allow and the auth check permits
 	ForceDemote {

src/api/client/membership/join.rs

@@ -3,7 +3,7 @@ use std::{borrow::Borrow, collections::HashMap, iter::once, sync::Arc};
 use axum::extract::State;
 use axum_client_ip::InsecureClientIp;
 use conduwuit::{
-	Err, Result, debug, debug_info, debug_warn, err, error, info,
+	Err, Result, debug, debug_error, debug_info, debug_warn, err, error, info,
 	matrix::{
 		StateKey,
 		event::{gen_event_id, gen_event_id_canonical_json},
@@ -20,14 +20,14 @@ use conduwuit::{
 };
 use futures::{FutureExt, StreamExt};
 use ruma::{
-	CanonicalJsonObject, CanonicalJsonValue, OwnedRoomId, OwnedServerName, OwnedUserId, RoomId,
-	RoomVersionId, UserId,
+	CanonicalJsonObject, CanonicalJsonValue, OwnedEventId, OwnedRoomId, OwnedServerName,
+	OwnedUserId, RoomId, RoomVersionId, UserId,
 	api::{
 		client::{
 			error::ErrorKind,
 			membership::{ThirdPartySigned, join_room_by_id, join_room_by_id_or_alias},
 		},
-		federation::{self},
+		federation::{self, event::get_room_state},
 	},
 	canonical_json::to_canonical_value,
 	events::{
@@ -156,31 +156,34 @@ pub(crate) async fn join_room_by_id_or_alias_route(
 			.await?;
 
 			let mut servers = body.via.clone();
-			servers.extend(
-				services
-					.rooms
-					.state_cache
-					.servers_invite_via(&room_id)
-					.map(ToOwned::to_owned)
-					.collect::<Vec<_>>()
-					.await,
-			);
+			if servers.is_empty() {
+				debug!("No via servers provided for join, injecting some.");
+				servers.extend(
+					services
+						.rooms
+						.state_cache
+						.servers_invite_via(&room_id)
+						.map(ToOwned::to_owned)
+						.collect::<Vec<_>>()
+						.await,
+				);
 
-			servers.extend(
-				services
-					.rooms
-					.state_cache
-					.invite_state(sender_user, &room_id)
-					.await
-					.unwrap_or_default()
-					.iter()
-					.filter_map(|event| event.get_field("sender").ok().flatten())
-					.filter_map(|sender: &str| UserId::parse(sender).ok())
-					.map(|user| user.server_name().to_owned()),
-			);
+				servers.extend(
+					services
+						.rooms
+						.state_cache
+						.invite_state(sender_user, &room_id)
+						.await
+						.unwrap_or_default()
+						.iter()
+						.filter_map(|event| event.get_field("sender").ok().flatten())
+						.filter_map(|sender: &str| UserId::parse(sender).ok())
+						.map(|user| user.server_name().to_owned()),
+				);
 
-			if let Some(server) = room_id.server_name() {
-				servers.push(server.to_owned());
+				if let Some(server) = room_id.server_name() {
+					servers.push(server.to_owned());
+				}
 			}
 
 			servers.sort_unstable();
@@ -444,11 +447,11 @@ async fn join_room_by_id_helper_remote(
 	// It has enough fields to be called a proper event now
 	let mut join_event = join_event_stub;
 
-	info!("Asking {remote_server} for send_join in room {room_id}");
+	info!("Asking {remote_server} for send_join in room {room_id} using snazzy fast joins");
 	let send_join_request = federation::membership::create_join_event::v2::Request {
 		room_id: room_id.to_owned(),
 		event_id: event_id.clone(),
-		omit_members: false,
+		omit_members: true,
 		pdu: services
 			.sending
 			.convert_to_outgoing_federation_event(join_event.clone())
@@ -685,6 +688,133 @@ async fn join_room_by_id_helper_remote(
 		.state
 		.set_room_state(room_id, statehash_after_join, &state_lock);
 
+	info!("Fetching the rest of the room state \"in the background\"");
+	// TODO: Actually do this in the background
+	// tokio::spawn(fetch_state_in_background(
+	// 	services.clone(),
+	// 	room_id.to_owned(),
+	// 	event_id.clone(),
+	// 	servers.to_vec(),
+	// ));
+	fetch_state_in_background(services, room_id.to_owned(), event_id, servers.to_vec())
+		.boxed()
+		.await?;
+	info!("Join completed");
+	Ok(())
+}
+
+async fn fetch_state_in_background(
+	services: &Services,
+	room_id: OwnedRoomId,
+	event_id: OwnedEventId,
+	via: Vec<OwnedServerName>,
+) -> Result<()> {
+	let room_version = services.rooms.state.get_room_version(&room_id).await?;
+
+	let mut state: HashMap<u64, OwnedEventId> = HashMap::new();
+
+	let mut cur_response = None;
+	for server_name in via {
+		cur_response = services
+			.sending
+			.send_federation_request(&server_name, get_room_state::v1::Request {
+				room_id: room_id.clone(),
+				event_id: event_id.clone(),
+			})
+			.await
+			.ok();
+		if cur_response.is_some() {
+			break;
+		}
+	}
+	let remote_state_response = cur_response
+		.ok_or_else(|| err!(BadServerResponse("Could not fetch state from any server")))?;
+
+	for pdu in remote_state_response.pdus.clone() {
+		match services.rooms.event_handler.parse_incoming_pdu(&pdu).await {
+			| Ok(t) => t,
+			| Err(e) => {
+				warn!("Could not parse PDU, ignoring: {e}");
+				continue;
+			},
+		};
+	}
+
+	info!("Going through room_state response PDUs");
+	for result in remote_state_response.pdus.iter().map(|pdu| {
+		services
+			.server_keys
+			.validate_and_add_event_id(pdu, &room_version)
+	}) {
+		let Ok((evt_id, value)) = result.await else {
+			continue;
+		};
+
+		let pdu = PduEvent::from_id_val(&evt_id, value.clone()).map_err(|e| {
+			debug_error!("Invalid PDU in fetching remote room state PDUs response: {value:#?}");
+			err!(BadServerResponse(debug_error!("Invalid PDU in send_join response: {e:?}")))
+		})?;
+
+		services.rooms.outlier.add_pdu_outlier(&evt_id, &value);
+
+		if let Some(state_key) = &pdu.state_key {
+			let shortstatekey = services
+				.rooms
+				.short
+				.get_or_create_shortstatekey(&pdu.kind.to_string().into(), state_key)
+				.await;
+
+			state.insert(shortstatekey, pdu.event_id.clone());
+		}
+	}
+
+	info!("Going through auth_chain response");
+	for result in remote_state_response.auth_chain.iter().map(|pdu| {
+		services
+			.server_keys
+			.validate_and_add_event_id(pdu, &room_version)
+	}) {
+		let Ok((event_id, value)) = result.await else {
+			continue;
+		};
+
+		services.rooms.outlier.add_pdu_outlier(&event_id, &value);
+	}
+
+	let new_room_state = services
+		.rooms
+		.event_handler
+		.resolve_state(&room_id, &room_version, state)
+		.await?;
+
+	info!("Forcing new room state");
+	let state_lock = services.rooms.state.mutex.lock(&room_id).await;
+	let HashSetCompressStateEvent {
+		shortstatehash: short_state_hash,
+		added,
+		removed,
+	} = services
+		.rooms
+		.state_compressor
+		.save_state(room_id.as_ref(), new_room_state)
+		.await?;
+
+	services
+		.rooms
+		.state
+		.force_state(room_id.as_ref(), short_state_hash, added, removed, &state_lock)
+		.await?;
+
+	info!(
+		"Updating joined counts for room just in case (e.g. we may have found a difference in \
+		 the room's m.room.member state"
+	);
+	services
+		.rooms
+		.state_cache
+		.update_joined_count(&room_id)
+		.await;
+	drop(state_lock);
 	Ok(())
 }
 

src/api/client/membership/leave.rs

@@ -215,7 +215,7 @@ pub async fn leave_room(
 	Ok(())
 }
 
-async fn remote_leave_room(
+pub async fn remote_leave_room(
 	services: &Services,
 	user_id: &UserId,
 	room_id: &RoomId,

src/api/client/membership/mod.rs

@@ -29,7 +29,7 @@ pub(crate) use self::{
 };
 pub use self::{
 	join::join_room_by_id_helper,
-	leave::{leave_all_rooms, leave_room},
+	leave::{leave_all_rooms, leave_room, remote_leave_room},
 };
 use crate::{Ruma, client::full_user_deactivate};
 

src/api/client/mod.rs

@@ -54,7 +54,7 @@ pub(super) use keys::*;
 pub(super) use media::*;
 pub(super) use media_legacy::*;
 pub(super) use membership::*;
-pub use membership::{join_room_by_id_helper, leave_all_rooms, leave_room};
+pub use membership::{join_room_by_id_helper, leave_all_rooms, leave_room, remote_leave_room};
 pub(super) use message::*;
 pub(super) use openid::*;
 pub(super) use presence::*;