refactor: Fix errors in api/client/profile.rs and api/client/unstable.rs

src/api/client/mod.rs

@@ -15,6 +15,7 @@
 pub(super) mod media_legacy;
 pub(super) mod membership;
 pub(super) mod message;
+pub(super) mod mutual_rooms;
 pub(super) mod openid;
 pub(super) mod presence;
 pub(super) mod profile;
@@ -35,7 +36,6 @@
 pub(super) mod threads;
 pub(super) mod to_device;
 pub(super) mod typing;
-pub(super) mod unstable;
 pub(super) mod unversioned;
 pub(super) mod user_directory;
 pub(super) mod voip;
@@ -60,10 +60,10 @@
 pub(super) use membership::*;
 pub use membership::{join_room_by_id_helper, leave_all_rooms, leave_room, remote_leave_room};
 pub(super) use message::*;
+pub(super) use mutual_rooms::*;
 pub(super) use openid::*;
 pub(super) use presence::*;
 pub(super) use profile::*;
-pub use profile::{update_all_rooms, update_avatar_url, update_displayname};
 pub use push::recreate_push_rules_and_return;
 pub(super) use push::*;
 pub(super) use read_marker::*;
@@ -82,7 +82,6 @@
 pub(super) use threads::*;
 pub(super) use to_device::*;
 pub(super) use typing::*;
-pub(super) use unstable::*;
 pub(super) use unversioned::*;
 pub(super) use user_directory::*;
 pub(super) use voip::*;

src/api/client/mutual_rooms.rs

@@ -0,0 +1,36 @@
+use axum::extract::State;
+use conduwuit::{Err, Result};
+use futures::StreamExt;
+use ruma::api::client::membership::mutual_rooms;
+
+use crate::Ruma;
+
+/// # `GET /_matrix/client/unstable/uk.half-shot.msc2666/user/mutual_rooms`
+///
+/// Gets all the rooms the sender shares with the specified user.
+///
+/// An implementation of [MSC2666](https://github.com/matrix-org/matrix-spec-proposals/pull/2666)
+#[tracing::instrument(skip_all, name = "mutual_rooms", level = "info")]
+pub(crate) async fn get_mutual_rooms_route(
+	State(services): State<crate::State>,
+	body: Ruma<mutual_rooms::unstable::Request>,
+) -> Result<mutual_rooms::unstable::Response> {
+	let sender_user = body.sender_user();
+
+	if sender_user == body.user_id {
+		return Err!(Request(Unknown("You cannot request rooms in common with yourself.")));
+	}
+
+	if !services.users.exists(&body.user_id).await {
+		return Ok(mutual_rooms::unstable::Response::new(vec![]));
+	}
+
+	let mutual_rooms = services
+		.rooms
+		.state_cache
+		.get_shared_rooms(sender_user, &body.user_id)
+		.collect()
+		.await;
+
+	Ok(mutual_rooms::unstable::Response::new(mutual_rooms))
+}

src/api/client/profile.rs

@@ -1,229 +1,26 @@
+use std::collections::BTreeMap;
+
 use axum::extract::State;
-use conduwuit::{
-	Err, Result,
-	matrix::pdu::PartialPdu,
-	utils::{IterStream, future::TryExtExt, stream::TryIgnore},
-	warn,
-};
+use conduwuit::{Err, Result, matrix::pdu::PartialPdu};
 use conduwuit_service::Services;
-use futures::{
-	FutureExt, StreamExt, TryStreamExt,
-	future::{join, join3, join4},
-};
+use futures::StreamExt;
 use ruma::{
-	OwnedMxcUri, OwnedRoomId, UserId,
+	UserId,
 	api::{
 		client::profile::{
-			get_avatar_url, get_display_name, get_profile, set_avatar_url, set_display_name,
+			delete_profile_field, get_profile, get_profile_field, set_profile_field,
 		},
 		federation,
 	},
+	assign,
 	events::room::member::{MembershipState, RoomMemberEventContent},
 	presence::PresenceState,
+	profile::{ProfileFieldName, ProfileFieldValue},
 };
+use serde_json::{Value, to_value};
 
 use crate::Ruma;
 
-/// # `PUT /_matrix/client/r0/profile/{userId}/displayname`
-///
-/// Updates the displayname.
-///
-/// - Also makes sure other users receive the update using presence EDUs
-pub(crate) async fn set_displayname_route(
-	State(services): State<crate::State>,
-	body: Ruma<set_display_name::v3::Request>,
-) -> Result<set_display_name::v3::Response> {
-	let sender_user = body.sender_user();
-	if services.users.is_suspended(sender_user).await? {
-		return Err!(Request(UserSuspended("You cannot perform this action while suspended.")));
-	}
-
-	if *sender_user != body.user_id && body.appservice_info.is_none() {
-		return Err!(Request(Forbidden("You cannot update the profile of another user")));
-	}
-
-	let all_joined_rooms: Vec<OwnedRoomId> = services
-		.rooms
-		.state_cache
-		.rooms_joined(&body.user_id)
-		.map(ToOwned::to_owned)
-		.collect()
-		.await;
-
-	update_displayname(&services, &body.user_id, body.displayname.clone(), &all_joined_rooms)
-		.boxed()
-		.await;
-
-	if services.config.allow_local_presence {
-		// Presence update
-		services
-			.presence
-			.ping_presence(&body.user_id, &PresenceState::Online)
-			.await?;
-	}
-
-	Ok(set_display_name::v3::Response {})
-}
-
-/// # `GET /_matrix/client/v3/profile/{userId}/displayname`
-///
-/// Returns the displayname of the user.
-///
-/// - If user is on another server and we do not have a local copy already fetch
-///   displayname over federation
-pub(crate) async fn get_displayname_route(
-	State(services): State<crate::State>,
-	body: Ruma<get_display_name::v3::Request>,
-) -> Result<get_display_name::v3::Response> {
-	if !services.globals.user_is_local(&body.user_id) {
-		// Create and update our local copy of the user
-		if let Ok(response) = services
-			.sending
-			.send_federation_request(
-				body.user_id.server_name(),
-				federation::query::get_profile_information::v1::Request {
-					user_id: body.user_id.clone(),
-					field: None, // we want the full user's profile to update locally too
-				},
-			)
-			.await
-		{
-			if !services.users.exists(&body.user_id).await {
-				services.users.create(&body.user_id, None, None).await?;
-			}
-
-			services
-				.users
-				.set_displayname(&body.user_id, response.displayname.clone());
-			services
-				.users
-				.set_avatar_url(&body.user_id, response.avatar_url.clone());
-			services
-				.users
-				.set_blurhash(&body.user_id, response.blurhash.clone());
-
-			return Ok(get_display_name::v3::Response { displayname: response.displayname });
-		}
-	}
-
-	if !services.users.exists(&body.user_id).await {
-		// Return 404 if this user doesn't exist and we couldn't fetch it over
-		// federation
-		return Err!(Request(NotFound("Profile was not found.")));
-	}
-
-	Ok(get_display_name::v3::Response {
-		displayname: services.users.displayname(&body.user_id).await.ok(),
-	})
-}
-
-/// # `PUT /_matrix/client/v3/profile/{userId}/avatar_url`
-///
-/// Updates the `avatar_url` and `blurhash`.
-///
-/// - Also makes sure other users receive the update using presence EDUs
-pub(crate) async fn set_avatar_url_route(
-	State(services): State<crate::State>,
-	body: Ruma<set_avatar_url::v3::Request>,
-) -> Result<set_avatar_url::v3::Response> {
-	let sender_user = body.sender_user();
-	if services.users.is_suspended(sender_user).await? {
-		return Err!(Request(UserSuspended("You cannot perform this action while suspended.")));
-	}
-
-	if *sender_user != body.user_id && body.appservice_info.is_none() {
-		return Err!(Request(Forbidden("You cannot update the profile of another user")));
-	}
-
-	let all_joined_rooms: Vec<OwnedRoomId> = services
-		.rooms
-		.state_cache
-		.rooms_joined(&body.user_id)
-		.map(ToOwned::to_owned)
-		.collect()
-		.await;
-
-	update_avatar_url(
-		&services,
-		&body.user_id,
-		body.avatar_url.clone(),
-		body.blurhash.clone(),
-		&all_joined_rooms,
-	)
-	.boxed()
-	.await;
-
-	if services.config.allow_local_presence {
-		// Presence update
-		services
-			.presence
-			.ping_presence(&body.user_id, &PresenceState::Online)
-			.await
-			.ok();
-	}
-
-	Ok(set_avatar_url::v3::Response {})
-}
-
-/// # `GET /_matrix/client/v3/profile/{userId}/avatar_url`
-///
-/// Returns the `avatar_url` and `blurhash` of the user.
-///
-/// - If user is on another server and we do not have a local copy already fetch
-///   `avatar_url` and blurhash over federation
-pub(crate) async fn get_avatar_url_route(
-	State(services): State<crate::State>,
-	body: Ruma<get_avatar_url::v3::Request>,
-) -> Result<get_avatar_url::v3::Response> {
-	if !services.globals.user_is_local(&body.user_id) {
-		// Create and update our local copy of the user
-		if let Ok(response) = services
-			.sending
-			.send_federation_request(
-				body.user_id.server_name(),
-				federation::query::get_profile_information::v1::Request {
-					user_id: body.user_id.clone(),
-					field: None, // we want the full user's profile to update locally as well
-				},
-			)
-			.await
-		{
-			if !services.users.exists(&body.user_id).await {
-				services.users.create(&body.user_id, None, None).await?;
-			}
-
-			services
-				.users
-				.set_displayname(&body.user_id, response.displayname.clone());
-			services
-				.users
-				.set_avatar_url(&body.user_id, response.avatar_url.clone());
-			services
-				.users
-				.set_blurhash(&body.user_id, response.blurhash.clone());
-
-			return Ok(get_avatar_url::v3::Response {
-				avatar_url: response.avatar_url,
-				blurhash: response.blurhash,
-			});
-		}
-	}
-
-	if !services.users.exists(&body.user_id).await {
-		// Return 404 if this user doesn't exist and we couldn't fetch it over
-		// federation
-		return Err!(Request(NotFound("Profile was not found.")));
-	}
-
-	let (avatar_url, blurhash) = join(
-		services.users.avatar_url(&body.user_id).ok(),
-		services.users.blurhash(&body.user_id).ok(),
-	)
-	.await;
-
-	Ok(get_avatar_url::v3::Response { avatar_url, blurhash })
-}
-
 /// # `GET /_matrix/client/v3/profile/{userId}`
 ///
 /// Returns the displayname, avatar_url, blurhash, and custom profile fields of
@@ -235,188 +32,289 @@ pub(crate) async fn get_profile_route(
 	State(services): State<crate::State>,
 	body: Ruma<get_profile::v3::Request>,
 ) -> Result<get_profile::v3::Response> {
+	let Some(profile) = fetch_full_profile(&services, &body.user_id).await else {
+		return Err!(Request(NotFound("This user's profile could not be fetched.")));
+	};
+
+	Ok(get_profile::v3::Response::from_iter(profile.into_iter()))
+}
+
+pub(crate) async fn get_profile_field_route(
+	State(services): State<crate::State>,
+	body: Ruma<get_profile_field::v3::Request>,
+) -> Result<get_profile_field::v3::Response> {
+	let value = fetch_profile_field(&services, &body.user_id, body.field.clone()).await?;
+
+	Ok(assign!(get_profile_field::v3::Response::default(), { value }))
+}
+
+pub(crate) async fn set_profile_field_route(
+	State(services): State<crate::State>,
+	body: Ruma<set_profile_field::v3::Request>,
+) -> Result<set_profile_field::v3::Response> {
+	if body.user_id != body.sender_user()
+		&& !(body.appservice_info.is_some()
+			|| services.admin.user_is_admin(body.sender_user()).await)
+	{
+		return Err!(Request(Forbidden("You may not change other users' profile data.")));
+	}
+
 	if !services.globals.user_is_local(&body.user_id) {
-		// Create and update our local copy of the user
-		if let Ok(response) = services
-			.sending
-			.send_federation_request(
-				body.user_id.server_name(),
-				federation::query::get_profile_information::v1::Request {
-					user_id: body.user_id.clone(),
-					field: None,
-				},
-			)
-			.await
-		{
-			if !services.users.exists(&body.user_id).await {
-				services.users.create(&body.user_id, None, None).await?;
+		return Err!(Request(InvalidParam("You may not change a remote user's profile data.")));
+	}
+
+	set_profile_field(&services, &body.user_id, ProfileFieldChange::Set(body.value.clone()))
+		.await;
+
+	Ok(set_profile_field::v3::Response::new())
+}
+
+pub(crate) async fn delete_profile_field_route(
+	State(services): State<crate::State>,
+	body: Ruma<delete_profile_field::v3::Request>,
+) -> Result<delete_profile_field::v3::Response> {
+	if body.user_id != body.sender_user()
+		&& !(body.appservice_info.is_some()
+			|| services.admin.user_is_admin(body.sender_user()).await)
+	{
+		return Err!(Request(Forbidden("You may not change other users' profile data.")));
+	}
+
+	if !services.globals.user_is_local(&body.user_id) {
+		return Err!(Request(InvalidParam("You may not change a remote user's profile data.")));
+	}
+
+	set_profile_field(&services, &body.user_id, ProfileFieldChange::Delete(body.field.clone()))
+		.await;
+
+	Ok(delete_profile_field::v3::Response::new())
+}
+
+async fn fetch_full_profile(
+	services: &Services,
+	user_id: &UserId,
+) -> Option<BTreeMap<String, Value>> {
+	// If the user exists locally, fetch their local profile
+	if services.users.exists(user_id).await {
+		let mut profile = BTreeMap::new();
+
+		// Get displayname and avatar_url independently because `all_profile_keys`
+		// doesn't include them
+		for field in [ProfileFieldName::AvatarUrl, ProfileFieldName::DisplayName] {
+			let key = field.as_str().to_owned();
+
+			if let Some(value) = get_local_profile_field(services, user_id, field).await {
+				profile.insert(key, value.value().into_owned());
 			}
+		}
 
-			services
-				.users
-				.set_displayname(&body.user_id, response.displayname.clone());
-			services
-				.users
-				.set_avatar_url(&body.user_id, response.avatar_url.clone());
-			services
-				.users
-				.set_blurhash(&body.user_id, response.blurhash.clone());
+		// Insert all other profile fields
+		let mut all_fields = services.users.all_profile_keys(user_id);
 
-			for (profile_key, profile_key_value) in &response.custom_profile_fields {
+		while let Some((key, value)) = all_fields.next().await {
+			profile.insert(key, value);
+		}
+
+		return Some(profile);
+	}
+
+	// Otherwise ask their homeserver
+	let Ok(response) = services
+		.sending
+		.send_federation_request(
+			user_id.server_name(),
+			federation::query::get_profile_information::v1::Request::new(user_id.to_owned()),
+		)
+		.await
+	else {
+		return None;
+	};
+
+	// Update our local copies of their profile fields
+	services.users.clear_profile(user_id).await;
+
+	for (field, value) in response.iter() {
+		let Ok(value) = ProfileFieldValue::new(&field, value.to_owned()) else {
+			// Skip malformed fields
+			continue;
+		};
+
+		set_profile_field(services, user_id, ProfileFieldChange::Set(value)).await;
+	}
+
+	Some(BTreeMap::from_iter(response.into_iter()))
+}
+
+async fn fetch_profile_field(
+	services: &Services,
+	user_id: &UserId,
+	field: ProfileFieldName,
+) -> Result<Option<ProfileFieldValue>> {
+	// If the user exists locally, fetch their local profile field
+	if services.globals.user_is_local(user_id) {
+		return Ok(get_local_profile_field(services, user_id, field).await);
+	}
+
+	// Otherwise ask their homeserver
+	let Ok(response) = services
+		.sending
+		.send_federation_request(
+			user_id.server_name(),
+			assign!(federation::query::get_profile_information::v1::Request::new(user_id.to_owned()), {
+				field: Some(field.clone())
+			}),
+		)
+		.await
+	else {
+		return Err!(Request(NotFound(
+			"User's homeserver could not provide this profile field."
+		)));
+	};
+
+	if let Some(value) = response.get(field.as_str()).map(ToOwned::to_owned) {
+		if let Ok(value) = ProfileFieldValue::new(field.as_str(), value) {
+			set_profile_field(services, user_id, ProfileFieldChange::Set(value.clone())).await;
+
+			Ok(Some(value))
+		} else {
+			Err!(Request(Unknown(
+				"User's homeserver returned malformed data for this profile field."
+			)))
+		}
+	} else {
+		set_profile_field(services, user_id, ProfileFieldChange::Delete(field)).await;
+
+		Ok(None)
+	}
+}
+
+async fn get_local_profile_field(
+	services: &Services,
+	user_id: &UserId,
+	field: ProfileFieldName,
+) -> Option<ProfileFieldValue> {
+	let value = match field.clone() {
+		| ProfileFieldName::AvatarUrl => services
+			.users
+			.avatar_url(user_id)
+			.await
+			.ok()
+			.map(to_value)
+			.transpose()
+			.expect("converting avatar url to value should succeed"),
+		| ProfileFieldName::DisplayName => services
+			.users
+			.displayname(user_id)
+			.await
+			.ok()
+			.map(to_value)
+			.transpose()
+			.expect("converting displayname to value should succeed"),
+		| other => services
+			.users
+			.profile_key(user_id, other.as_str())
+			.await
+			.ok(),
+	}?;
+
+	Some(
+		ProfileFieldValue::new(field.as_str(), value)
+			.expect("local profile field should be valid"),
+	)
+}
+
+enum ProfileFieldChange {
+	Set(ProfileFieldValue),
+	Delete(ProfileFieldName),
+}
+
+impl ProfileFieldChange {
+	fn field_name(&self) -> ProfileFieldName {
+		match self {
+			| &ProfileFieldChange::Delete(ref name) => name.clone(),
+			| &ProfileFieldChange::Set(ref value) => value.field_name(),
+		}
+	}
+
+	fn value(&self) -> Option<Value> {
+		if let &ProfileFieldChange::Set(ref value) = self {
+			Some(value.value().into_owned())
+		} else {
+			None
+		}
+	}
+}
+
+async fn set_profile_field(services: &Services, user_id: &UserId, change: ProfileFieldChange) {
+	let field_name = change.field_name();
+
+	match change {
+		| ProfileFieldChange::Set(ProfileFieldValue::DisplayName(displayname)) => {
+			services
+				.users
+				.set_displayname(user_id, Some(displayname).filter(|dn| !dn.is_empty()));
+		},
+		| ProfileFieldChange::Set(ProfileFieldValue::AvatarUrl(avatar_url)) => {
+			services
+				.users
+				.set_avatar_url(user_id, Some(avatar_url).filter(|av| av.is_valid()));
+		},
+		| ProfileFieldChange::Delete(ProfileFieldName::DisplayName) => {
+			services.users.set_displayname(user_id, None);
+		},
+		| ProfileFieldChange::Delete(ProfileFieldName::AvatarUrl) => {
+			services.users.set_avatar_url(user_id, None);
+		},
+		| other =>
+			if other.field_name().as_str() == "blurhash" {
+				if let Some(Value::String(blurhash)) = other.value() {
+					services.users.set_blurhash(user_id, Some(blurhash));
+				} else {
+					services.users.set_blurhash(user_id, None);
+				}
+			} else {
 				services.users.set_profile_key(
-					&body.user_id,
-					profile_key,
-					Some(profile_key_value.clone()),
+					user_id,
+					other.field_name().as_str(),
+					other.value(),
 				);
-			}
+			},
+	}
 
-			return Ok(get_profile::v3::Response {
-				displayname: response.displayname,
-				avatar_url: response.avatar_url,
-				blurhash: response.blurhash,
-				custom_profile_fields: response.custom_profile_fields,
-			});
+	// If the user is local and changed their displayname or avatar_url, update it
+	// in all their joined rooms
+	if matches!(field_name, ProfileFieldName::AvatarUrl | ProfileFieldName::DisplayName)
+		&& services.users.is_active_local(user_id).await
+	{
+		let displayname = services.users.displayname(user_id).await.ok();
+		let avatar_url = services.users.avatar_url(user_id).await.ok();
+		let membership_content = assign!(
+			RoomMemberEventContent::new(MembershipState::Join), { displayname, avatar_url }
+		);
+
+		let mut all_joined_rooms = services.rooms.state_cache.rooms_joined(user_id);
+
+		while let Some(room_id) = all_joined_rooms.next().await {
+			let state_lock = services.rooms.state.mutex.lock(room_id.as_str()).await;
+
+			let _ = services
+				.rooms
+				.timeline
+				.build_and_append_pdu(
+					PartialPdu::state(user_id.to_string(), &membership_content),
+					user_id,
+					Some(&room_id),
+					&state_lock,
+				)
+				.await;
 		}
-	}
 
-	if !services.users.exists(&body.user_id).await {
-		// Return 404 if this user doesn't exist and we couldn't fetch it over
-		// federation
-		return Err!(Request(NotFound("Profile was not found.")));
-	}
-
-	let (avatar_url, blurhash, displayname, custom_profile_fields) = join4(
-		services.users.avatar_url(&body.user_id).ok(),
-		services.users.blurhash(&body.user_id).ok(),
-		services.users.displayname(&body.user_id).ok(),
-		services.users.all_profile_keys(&body.user_id).collect(),
-	)
-	.await;
-
-	Ok(get_profile::v3::Response {
-		avatar_url,
-		blurhash,
-		displayname,
-		custom_profile_fields,
-	})
-}
-
-pub async fn update_displayname(
-	services: &Services,
-	user_id: &UserId,
-	displayname: Option<String>,
-	all_joined_rooms: &[OwnedRoomId],
-) {
-	let (current_avatar_url, current_blurhash, current_displayname) = join3(
-		services.users.avatar_url(user_id).ok(),
-		services.users.blurhash(user_id).ok(),
-		services.users.displayname(user_id).ok(),
-	)
-	.await;
-
-	if displayname == current_displayname {
-		return;
-	}
-
-	services.users.set_displayname(user_id, displayname.clone());
-
-	// Send a new join membership event into all joined rooms
-	let avatar_url = &current_avatar_url;
-	let blurhash = &current_blurhash;
-	let displayname = &displayname;
-	let all_joined_rooms: Vec<_> = all_joined_rooms
-		.iter()
-		.try_stream()
-		.and_then(|room_id: &OwnedRoomId| async move {
-			let pdu = PartialPdu::state(user_id.to_string(), &RoomMemberEventContent {
-				displayname: displayname.clone(),
-				membership: MembershipState::Join,
-				avatar_url: avatar_url.clone(),
-				blurhash: blurhash.clone(),
-				join_authorized_via_users_server: None,
-				reason: None,
-				is_direct: None,
-				third_party_invite: None,
-				redact_events: None,
-			});
-
-			Ok((pdu, room_id))
-		})
-		.ignore_err()
-		.collect()
-		.await;
-
-	update_all_rooms(services, all_joined_rooms, user_id)
-		.boxed()
-		.await;
-}
-
-pub async fn update_avatar_url(
-	services: &Services,
-	user_id: &UserId,
-	avatar_url: Option<OwnedMxcUri>,
-	blurhash: Option<String>,
-	all_joined_rooms: &[OwnedRoomId],
-) {
-	let (current_avatar_url, current_blurhash, current_displayname) = join3(
-		services.users.avatar_url(user_id).ok(),
-		services.users.blurhash(user_id).ok(),
-		services.users.displayname(user_id).ok(),
-	)
-	.await;
-
-	if current_avatar_url == avatar_url && current_blurhash == blurhash {
-		return;
-	}
-
-	services.users.set_avatar_url(user_id, avatar_url.clone());
-	services.users.set_blurhash(user_id, blurhash.clone());
-
-	// Send a new join membership event into all joined rooms
-	let avatar_url = &avatar_url;
-	let blurhash = &blurhash;
-	let displayname = &current_displayname;
-	let all_joined_rooms: Vec<_> = all_joined_rooms
-		.iter()
-		.try_stream()
-		.and_then(|room_id: &OwnedRoomId| async move {
-			let pdu = PartialPdu::state(user_id.to_string(), &RoomMemberEventContent {
-				avatar_url: avatar_url.clone(),
-				blurhash: blurhash.clone(),
-				membership: MembershipState::Join,
-				displayname: displayname.clone(),
-				join_authorized_via_users_server: None,
-				reason: None,
-				is_direct: None,
-				third_party_invite: None,
-				redact_events: None,
-			});
-
-			Ok((pdu, room_id))
-		})
-		.ignore_err()
-		.collect()
-		.await;
-
-	update_all_rooms(services, all_joined_rooms, user_id)
-		.boxed()
-		.await;
-}
-
-pub async fn update_all_rooms(
-	services: &Services,
-	all_joined_rooms: Vec<(PartialPdu, &OwnedRoomId)>,
-	user_id: &UserId,
-) {
-	for (partial_pdu, room_id) in all_joined_rooms {
-		let state_lock = services.rooms.state.mutex.lock(room_id).await;
-		if let Err(e) = services
-			.rooms
-			.timeline
-			.build_and_append_pdu(partial_pdu, user_id, Some(room_id), &state_lock)
-			.await
-		{
-			warn!(%user_id, %room_id, "Failed to update/send new profile join membership update in room: {e}");
+		if services.config.allow_local_presence {
+			// Send a presence EDU to indicate the profile changed
+			let _ = services
+				.presence
+				.ping_presence(user_id, &PresenceState::Online)
+				.await;
 		}
 	}
 }

src/api/client/unstable.rs

@@ -1,303 +0,0 @@
-use std::collections::BTreeMap;
-
-use axum::extract::State;
-use axum_client_ip::InsecureClientIp;
-use conduwuit::{Err, Result};
-use futures::{FutureExt, StreamExt};
-use ruma::{
-	OwnedRoomId,
-	api::{
-		client::{
-			membership::mutual_rooms,
-			profile::{delete_profile_key, get_profile_key, set_profile_key},
-		},
-		federation,
-	},
-	presence::PresenceState,
-};
-
-use super::{update_avatar_url, update_displayname};
-use crate::Ruma;
-
-/// # `GET /_matrix/client/unstable/uk.half-shot.msc2666/user/mutual_rooms`
-///
-/// Gets all the rooms the sender shares with the specified user.
-///
-/// TODO: Implement pagination, currently this just returns everything
-///
-/// An implementation of [MSC2666](https://github.com/matrix-org/matrix-spec-proposals/pull/2666)
-#[tracing::instrument(skip_all, fields(%client), name = "mutual_rooms", level = "info")]
-pub(crate) async fn get_mutual_rooms_route(
-	State(services): State<crate::State>,
-	InsecureClientIp(client): InsecureClientIp,
-	body: Ruma<mutual_rooms::unstable::Request>,
-) -> Result<mutual_rooms::unstable::Response> {
-	let sender_user = body.sender_user();
-
-	if sender_user == body.user_id {
-		return Err!(Request(Unknown("You cannot request rooms in common with yourself.")));
-	}
-
-	if !services.users.exists(&body.user_id).await {
-		return Ok(mutual_rooms::unstable::Response { joined: vec![], next_batch_token: None });
-	}
-
-	let mutual_rooms: Vec<OwnedRoomId> = services
-		.rooms
-		.state_cache
-		.get_shared_rooms(sender_user, &body.user_id)
-		.map(ToOwned::to_owned)
-		.collect()
-		.await;
-
-	Ok(mutual_rooms::unstable::Response {
-		joined: mutual_rooms,
-		next_batch_token: None,
-	})
-}
-
-/// # `PUT /_matrix/client/unstable/uk.tcpip.msc4133/profile/{user_id}/{field}`
-///
-/// Updates the profile key-value field of a user, as per MSC4133.
-///
-/// This also handles the avatar_url and displayname being updated.
-pub(crate) async fn set_profile_key_route(
-	State(services): State<crate::State>,
-	body: Ruma<set_profile_key::unstable::Request>,
-) -> Result<set_profile_key::unstable::Response> {
-	let sender_user = body.sender_user();
-
-	if *sender_user != body.user_id && body.appservice_info.is_none() {
-		return Err!(Request(Forbidden("You cannot update the profile of another user")));
-	}
-
-	if body.kv_pair.is_empty() {
-		return Err!(Request(BadJson(
-			"The key-value pair JSON body is empty. Use DELETE to delete a key"
-		)));
-	}
-
-	if body.kv_pair.len() > 1 {
-		// TODO: support PATCH or "recursively" adding keys in some sort
-		return Err!(Request(BadJson(
-			"This endpoint can only take one key-value pair at a time"
-		)));
-	}
-
-	let Some(profile_key_value) = body.kv_pair.get(&body.key_name) else {
-		return Err!(Request(BadJson(
-			"The key does not match the URL field key, or JSON body is empty (use DELETE)"
-		)));
-	};
-
-	if body.kv_pair.keys().any(|key| key.len() > 128) {
-		return Err!(Request(BadJson("Key names cannot be longer than 128 bytes")));
-	}
-
-	if body.key_name == "displayname" {
-		let Some(display_name) = profile_key_value.as_str() else {
-			return Err!(Request(BadJson("displayname must be a string")));
-		};
-		let all_joined_rooms: Vec<OwnedRoomId> = services
-			.rooms
-			.state_cache
-			.rooms_joined(&body.user_id)
-			.map(Into::into)
-			.collect()
-			.await;
-
-		update_displayname(
-			&services,
-			&body.user_id,
-			Some(display_name.to_owned()),
-			&all_joined_rooms,
-		)
-		.boxed()
-		.await;
-	} else if body.key_name == "avatar_url" {
-		let Some(avatar_url) = profile_key_value.as_str() else {
-			return Err!(Request(BadJson("avatar_url must be a string")));
-		};
-		let mxc = ruma::OwnedMxcUri::from(avatar_url);
-
-		let all_joined_rooms: Vec<OwnedRoomId> = services
-			.rooms
-			.state_cache
-			.rooms_joined(&body.user_id)
-			.map(Into::into)
-			.collect()
-			.await;
-
-		update_avatar_url(&services, &body.user_id, Some(mxc), None, &all_joined_rooms)
-			.boxed()
-			.await;
-	} else {
-		services.users.set_profile_key(
-			&body.user_id,
-			&body.key_name,
-			Some(profile_key_value.clone()),
-		);
-	}
-
-	if services.config.allow_local_presence {
-		// Presence update
-		services
-			.presence
-			.ping_presence(&body.user_id, &PresenceState::Online)
-			.await?;
-	}
-
-	Ok(set_profile_key::unstable::Response {})
-}
-
-/// # `DELETE /_matrix/client/unstable/uk.tcpip.msc4133/profile/{user_id}/{field}`
-///
-/// Deletes the profile key-value field of a user, as per MSC4133.
-///
-/// This also handles the avatar_url and displayname being updated.
-pub(crate) async fn delete_profile_key_route(
-	State(services): State<crate::State>,
-	body: Ruma<delete_profile_key::unstable::Request>,
-) -> Result<delete_profile_key::unstable::Response> {
-	let sender_user = body.sender_user();
-
-	if *sender_user != body.user_id && body.appservice_info.is_none() {
-		return Err!(Request(Forbidden("You cannot update the profile of another user")));
-	}
-
-	if body.kv_pair.len() > 1 {
-		// TODO: support PATCH or "recursively" adding keys in some sort
-		return Err!(Request(BadJson(
-			"This endpoint can only take one key-value pair at a time"
-		)));
-	}
-
-	if body.key_name == "displayname" {
-		let all_joined_rooms: Vec<OwnedRoomId> = services
-			.rooms
-			.state_cache
-			.rooms_joined(&body.user_id)
-			.map(Into::into)
-			.collect()
-			.await;
-
-		update_displayname(&services, &body.user_id, None, &all_joined_rooms)
-			.boxed()
-			.await;
-	} else if body.key_name == "avatar_url" {
-		let all_joined_rooms: Vec<OwnedRoomId> = services
-			.rooms
-			.state_cache
-			.rooms_joined(&body.user_id)
-			.map(Into::into)
-			.collect()
-			.await;
-
-		update_avatar_url(&services, &body.user_id, None, None, &all_joined_rooms)
-			.boxed()
-			.await;
-	} else {
-		services
-			.users
-			.set_profile_key(&body.user_id, &body.key_name, None);
-	}
-
-	if services.config.allow_local_presence {
-		// Presence update
-		services
-			.presence
-			.ping_presence(&body.user_id, &PresenceState::Online)
-			.await?;
-	}
-
-	Ok(delete_profile_key::unstable::Response {})
-}
-
-/// # `GET /_matrix/client/unstable/uk.tcpip.msc4133/profile/{userId}/{field}}`
-///
-/// Gets the profile key-value field of a user, as per MSC4133.
-///
-/// - If user is on another server and we do not have a local copy already fetch
-///   the value over federation
-pub(crate) async fn get_profile_key_route(
-	State(services): State<crate::State>,
-	body: Ruma<get_profile_key::unstable::Request>,
-) -> Result<get_profile_key::unstable::Response> {
-	let mut profile_key_value: BTreeMap<String, serde_json::Value> = BTreeMap::new();
-
-	if !services.globals.user_is_local(&body.user_id) {
-		// Create and update our local copy of the user
-		if let Ok(response) = services
-			.sending
-			.send_federation_request(
-				body.user_id.server_name(),
-				federation::query::get_profile_information::v1::Request {
-					user_id: body.user_id.clone(),
-					field: None, // we want the full user's profile to update locally as well
-				},
-			)
-			.await
-		{
-			if !services.users.exists(&body.user_id).await {
-				services.users.create(&body.user_id, None, None).await?;
-			}
-
-			services
-				.users
-				.set_displayname(&body.user_id, response.displayname.clone());
-
-			services
-				.users
-				.set_avatar_url(&body.user_id, response.avatar_url.clone());
-
-			services
-				.users
-				.set_blurhash(&body.user_id, response.blurhash.clone());
-
-			match response.custom_profile_fields.get(&body.key_name) {
-				| Some(value) => {
-					profile_key_value.insert(body.key_name.clone(), value.clone());
-					services.users.set_profile_key(
-						&body.user_id,
-						&body.key_name,
-						Some(value.clone()),
-					);
-				},
-				| _ => {
-					return Err!(Request(NotFound("The requested profile key does not exist.")));
-				},
-			}
-
-			if profile_key_value.is_empty() {
-				return Err!(Request(NotFound("The requested profile key does not exist.")));
-			}
-
-			return Ok(get_profile_key::unstable::Response { value: profile_key_value });
-		}
-	}
-
-	if !services.users.exists(&body.user_id).await {
-		// Return 404 if this user doesn't exist and we couldn't fetch it over
-		// federation
-		return Err!(Request(NotFound("Profile was not found.")));
-	}
-
-	match services
-		.users
-		.profile_key(&body.user_id, &body.key_name)
-		.await
-	{
-		| Ok(value) => {
-			profile_key_value.insert(body.key_name.clone(), value);
-		},
-		| _ => {
-			return Err!(Request(NotFound("The requested profile key does not exist.")));
-		},
-	}
-
-	if profile_key_value.is_empty() {
-		return Err!(Request(NotFound("The requested profile key does not exist.")));
-	}
-
-	Ok(get_profile_key::unstable::Response { value: profile_key_value })
-}

src/api/router.rs

@@ -22,9 +22,6 @@
 pub fn build(router: Router<State>, server: &Server) -> Router<State> {
 	let config = &server.config;
 	let mut router = router
-        .ruma_route(&client::get_profile_key_route)
-        .ruma_route(&client::set_profile_key_route)
-        .ruma_route(&client::delete_profile_key_route)
         .ruma_route(&client::appservice_ping)
 		.ruma_route(&client::get_supported_versions_route)
 		.ruma_route(&client::get_register_available_route)
@@ -64,10 +61,9 @@ pub fn build(router: Router<State>, server: &Server) -> Router<State> {
 		.ruma_route(&client::set_room_account_data_route)
 		.ruma_route(&client::get_global_account_data_route)
 		.ruma_route(&client::get_room_account_data_route)
-		.ruma_route(&client::set_displayname_route)
-		.ruma_route(&client::get_displayname_route)
-		.ruma_route(&client::set_avatar_url_route)
-		.ruma_route(&client::get_avatar_url_route)
+		.ruma_route(&client::get_profile_field_route)
+        .ruma_route(&client::set_profile_field_route)
+        .ruma_route(&client::delete_profile_field_route)
 		.ruma_route(&client::get_profile_route)
 		.ruma_route(&client::set_presence_route)
 		.ruma_route(&client::get_presence_route)

src/service/rooms/timeline/create.rs

@@ -13,11 +13,9 @@
 };
 use futures::{StreamExt, TryStreamExt, future, future::ready};
 use ruma::{
-	CanonicalJsonObject, CanonicalJsonValue, OwnedEventId, RoomId, UserId,
-	events::{StateEventType, TimelineEventType, room::create::RoomCreateEventContent},
-	uint,
+	CanonicalJsonObject, CanonicalJsonValue, OwnedEventId, OwnedRoomId, RoomId, RoomVersionId, UserId, events::{StateEventType, TimelineEventType, room::create::RoomCreateEventContent}, room_version_rules::RoomVersionRules, uint
 };
-use serde_json::value::to_raw_value;
+use serde_json::value::{RawValue, to_raw_value};
 
 use super::RoomMutexGuard;
 
@@ -80,7 +78,7 @@ pub async fn create_event(
 	sender: &UserId,
 	room_id: Option<&RoomId>,
 	_mutex_lock: &RoomMutexGuard,
-) -> Result<(PduEvent, RoomVersionId)> {
+) -> Result<(PduEvent, RoomVersionRules)> {
 	let PartialPdu {
 		event_type,
 		content,
@@ -260,13 +258,13 @@ pub async fn create_event(
 		pdu.event_id,
 		pdu.room_id.as_ref().map_or("None", |id| id.as_str())
 	);
-	Ok((pdu, room_version))
+	Ok((pdu, room_version_rules))
 }
 
 #[implement(super::Service)]
 pub async fn create_hash_and_sign_event(
 	&self,
-	pdu_builder: PduBuilder,
+	partial_pdu: PartialPdu,
 	sender: &UserId,
 	room_id: Option<&RoomId>,
 	mutex_lock: &RoomMutexGuard, /* Take mutex guard to make sure users get the room
@@ -275,8 +273,8 @@ pub async fn create_hash_and_sign_event(
 	if !self.services.globals.user_is_local(sender) {
 		return Err!(Request(Forbidden("Sender must be a local user")));
 	}
-	let (mut pdu, room_version) = self
-		.create_event(pdu_builder, sender, room_id, mutex_lock)
+	let (mut pdu, room_version_rules) = self
+		.create_event(partial_pdu, sender, room_id, mutex_lock)
 		.await?;
 	// Hash and sign
 	let mut pdu_json = utils::to_canonical_object(&pdu).map_err(|e| {

src/service/users/mod.rs

@@ -1274,7 +1274,6 @@ pub fn set_profile_key(
 		profile_key: &str,
 		profile_key_value: Option<serde_json::Value>,
 	) {
-		// TODO: insert to the stable MSC4175 key when it's stable
 		let key = (user_id, profile_key);
 
 		if let Some(value) = profile_key_value {
@@ -1284,6 +1283,17 @@ pub fn set_profile_key(
 		}
 	}
 
+	/// Clears all profile data for a user, including display name and avatar
+	/// url.
+	pub async fn clear_profile(&self, user_id: &UserId) {
+		self.set_displayname(user_id, None);
+		self.set_avatar_url(user_id, None);
+		self.set_blurhash(user_id, None);
+		self.all_profile_keys(user_id)
+			.ready_for_each(|(key, _)| self.set_profile_key(user_id, &key, None))
+			.await;
+	}
+
 	#[cfg(not(feature = "ldap"))]
 	pub async fn search_ldap(&self, _user_id: &UserId) -> Result<Vec<(String, Option<bool>)>> {
 		Err!(FeatureDisabled("ldap"))