refactor: Move profile field setting logic into users service

src/service/users/account.rs

@@ -10,17 +10,19 @@ use futures::{FutureExt, Stream, StreamExt};
 use lettre::Address;
 use ruma::{
 	MilliSecondsSinceUnixEpoch, OwnedDeviceId, OwnedUserId, UserId,
+	api::client::profile::PropagateTo,
 	events::{
 		GlobalAccountDataEventType, ignored_user_list::IgnoredUserListEvent,
 		push_rules::PushRulesEvent, room::message::RoomMessageEventContent,
 	},
+	profile::ProfileFieldValue,
 	push::Ruleset,
 };
 use ruminuwuity::invite_permission_config::{FilterLevel, InvitePermissionConfigEvent};
 
 use crate::{
 	appservice::RegistrationInfo,
-	users::{HashedPassword, UserSuspension},
+	users::{HashedPassword, UserSuspension, profile::ProfileFieldChange},
 };
 
 /// The status of an access token.
@@ -151,7 +153,13 @@ impl super::Service {
 				displayname.push_str(suffix);
 			}
 
-			self.set_displayname(user_id, Some(displayname));
+			self.set_profile_field(
+				user_id,
+				ProfileFieldChange::Set(ProfileFieldValue::DisplayName(displayname)),
+				PropagateTo::None,
+			)
+			.await
+			.expect("should be able to set display name");
 		};
 
 		// Set default push rules

src/service/users/mod.rs

@@ -14,11 +14,12 @@ use conduwuit::{
 	utils::{self},
 };
 use database::Map;
+pub use profile::ProfileFieldChange;
 use ruma::{UserId, api::error::ErrorKind, encryption::CrossSigningKey, serde::Raw};
 use serde::{Deserialize, Serialize};
 
 use crate::{
-	Dep, account_data, admin, appservice, config, firstrun, globals, oauth,
+	Dep, account_data, admin, appservice, config, firstrun, globals, oauth, presence,
 	rooms::{self, alias, membership},
 	threepid,
 };
@@ -62,9 +63,12 @@ struct Services {
 	globals: Dep<globals::Service>,
 	membership: Dep<membership::Service>,
 	oauth: Dep<oauth::Service>,
+	presence: Dep<presence::Service>,
+	state: Dep<rooms::state::Service>,
 	state_accessor: Dep<rooms::state_accessor::Service>,
 	state_cache: Dep<rooms::state_cache::Service>,
 	threepid: Dep<threepid::Service>,
+	timeline: Dep<rooms::timeline::Service>,
 }
 
 struct Data {
@@ -110,10 +114,13 @@ impl crate::Service for Service {
 				globals: args.depend::<globals::Service>("globals"),
 				membership: args.depend::<membership::Service>("membership"),
 				oauth: args.depend::<oauth::Service>("oauth"),
+				presence: args.depend::<presence::Service>("presence"),
+				state: args.depend::<rooms::state::Service>("rooms::state"),
 				state_accessor: args
 					.depend::<rooms::state_accessor::Service>("rooms::state_accessor"),
 				state_cache: args.depend::<rooms::state_cache::Service>("rooms::state_cache"),
 				threepid: args.depend::<threepid::Service>("threepid"),
+				timeline: args.depend::<rooms::timeline::Service>("rooms::timeline"),
 			},
 			db: Data {
 				keychangeid_userid: args.db["keychangeid_userid"].clone(),

src/service/users/profile.rs

@@ -1,17 +1,263 @@
-use conduwuit::utils::{ReadyExt, stream::TryIgnore};
+use std::collections::BTreeMap;
+
+use conduwuit::{
+	Err, Result,
+	pdu::PartialPdu,
+	utils::{ReadyExt, stream::TryIgnore, to_canonical_object},
+};
 use database::{Deserialized, Ignore, Interfix, Json};
 use futures::{Stream, StreamExt};
-use ruma::{OwnedMxcUri, UserId};
+use ruma::{
+	OwnedMxcUri, UserId,
+	api::client::profile::PropagateTo,
+	events::room::member::MembershipState,
+	presence::PresenceState,
+	profile::{ProfileFieldName, ProfileFieldValue},
+};
+use serde_json::{Value, to_value};
+
+pub enum ProfileFieldChange {
+	Set(ProfileFieldValue),
+	Delete(ProfileFieldName),
+}
+
+impl ProfileFieldChange {
+	fn field_name(&self) -> ProfileFieldName {
+		match self {
+			| &Self::Delete(ref name) => name.clone(),
+			| &Self::Set(ref value) => value.field_name(),
+		}
+	}
+
+	fn value(&self) -> Option<Value> {
+		if let Self::Set(value) = self {
+			Some(value.value().into_owned())
+		} else {
+			None
+		}
+	}
+}
 
 impl super::Service {
+	pub async fn set_profile_field(
+		&self,
+		user_id: &UserId,
+		change: ProfileFieldChange,
+		propagate_to: PropagateTo,
+	) -> Result<()> {
+		const MAX_KEY_LENGTH_BYTES: usize = 255;
+		const MAX_PROFILE_LENGTH_BYTES: usize = 65536;
+
+		let field_name = change.field_name();
+
+		// TODO: The spec mentions special error codes (M_PROFILE_TOO_LARGE,
+		// M_KEY_TOO_LARGE) for profile field size limits, but they're not in its list
+		// of error codes and Ruma doesn't have them. Should we return those, or is
+		// M_TOO_LARGE okay?
+		if field_name.as_str().len() > MAX_KEY_LENGTH_BYTES {
+			return Err!(Request(TooLarge(
+				"Individual profile keys must not exceed {MAX_KEY_LENGTH_BYTES} bytes in length."
+			)));
+		}
+
+		// Serialize the entire profile as canonical JSON, including the new change,
+		// to check if it exceeds 64 KiB
+		{
+			let mut full_profile = self.get_local_profile(user_id).await;
+
+			match &change {
+				| ProfileFieldChange::Set(value) => {
+					full_profile.insert(
+						value.field_name().as_str().to_owned(),
+						value.value().clone().into_owned(),
+					);
+				},
+				| ProfileFieldChange::Delete(key) => {
+					full_profile.remove(key.as_str());
+				},
+			}
+
+			if let Ok(canonical_profile) = to_canonical_object(full_profile) {
+				if serde_json::to_string(&canonical_profile)
+					.expect("should be able to serialize to string")
+					.len() > MAX_PROFILE_LENGTH_BYTES
+				{
+					return Err!(
+						"Profile data must not exceed {MAX_PROFILE_LENGTH_BYTES} bytes in \
+						 length."
+					);
+				}
+			} else {
+				return Err!(Request(BadJson("Failed to canonicalize profile.")));
+			}
+		}
+
+		// If the user is local and changed their displayname or avatar_url, update it
+		// in all their joined rooms. This is done before updating their profile data
+		// so we can check the old value of the field if `propagate_to` is `unchanged`.
+		if matches!(field_name, ProfileFieldName::AvatarUrl | ProfileFieldName::DisplayName)
+			&& matches!(propagate_to, PropagateTo::All | PropagateTo::Unchanged)
+			&& self.services.globals.user_is_local(user_id)
+		{
+			let current_displayname = self.displayname(user_id).await.ok();
+			let current_avatar_url = self.avatar_url(user_id).await.ok();
+
+			let mut all_joined_rooms = self.services.state_cache.rooms_joined(user_id);
+
+			while let Some(room_id) = all_joined_rooms.next().await {
+				// TODO: this clobbers any custom fields on the event content
+				let mut current_membership = self
+					.services
+					.state_accessor
+					.get_member(&room_id, user_id)
+					.await
+					.expect("should be able to fetch membership event for joined room");
+
+				assert_eq!(
+					current_membership.membership,
+					MembershipState::Join,
+					"user should be joined"
+				);
+
+				// If `propagate_to` is `unchanged`, and the current value of the field we're
+				// updating was changed from its global value in this room, skip it.
+				if matches!(propagate_to, PropagateTo::Unchanged) {
+					let field_changed_from_global = match field_name {
+						| ProfileFieldName::AvatarUrl =>
+							current_membership.avatar_url.as_ref() != current_avatar_url.as_ref(),
+						| ProfileFieldName::DisplayName =>
+							current_membership.displayname.as_ref()
+								!= current_displayname.as_ref(),
+						| _ => unreachable!(),
+					};
+
+					if field_changed_from_global {
+						continue;
+					}
+				}
+
+				let state_lock = self.services.state.mutex.lock(room_id.as_str()).await;
+
+				// Preserve keys in accordance with the key copying rules
+				current_membership.reason = None;
+				current_membership.join_authorized_via_users_server = None;
+				match &change {
+					| ProfileFieldChange::Set(ProfileFieldValue::AvatarUrl(avatar_url)) => {
+						current_membership.avatar_url = Some(avatar_url.clone());
+					},
+					| ProfileFieldChange::Set(ProfileFieldValue::DisplayName(displayname)) => {
+						current_membership.displayname = Some(displayname.clone());
+					},
+					| ProfileFieldChange::Delete(ProfileFieldName::AvatarUrl) => {
+						current_membership.avatar_url = None;
+					},
+					| ProfileFieldChange::Delete(ProfileFieldName::DisplayName) => {
+						current_membership.displayname = None;
+					},
+					| _ => unreachable!(),
+				}
+
+				let _ = self
+					.services
+					.timeline
+					.build_and_append_pdu(
+						PartialPdu::state(user_id.to_string(), &current_membership),
+						user_id,
+						Some(&room_id),
+						&state_lock,
+					)
+					.await;
+			}
+
+			if self.services.config.allow_local_presence {
+				// Send a presence EDU to indicate the profile changed
+				let _ = self
+					.services
+					.presence
+					.ping_presence(user_id, &PresenceState::Online)
+					.await;
+			}
+		}
+
+		match change {
+			| ProfileFieldChange::Set(ProfileFieldValue::DisplayName(displayname)) => {
+				self.set_displayname(user_id, Some(displayname).filter(|dn| !dn.is_empty()));
+			},
+			| ProfileFieldChange::Set(ProfileFieldValue::AvatarUrl(avatar_url)) => {
+				self.set_avatar_url(user_id, Some(avatar_url).filter(|av| av.is_valid()));
+			},
+			| ProfileFieldChange::Delete(ProfileFieldName::DisplayName) => {
+				self.set_displayname(user_id, None);
+			},
+			| ProfileFieldChange::Delete(ProfileFieldName::AvatarUrl) => {
+				self.set_avatar_url(user_id, None);
+			},
+			| other => self.set_profile_key(user_id, other.field_name().as_str(), other.value()),
+		}
+
+		Ok(())
+	}
+
+	pub async fn get_local_profile(&self, user_id: &UserId) -> BTreeMap<String, Value> {
+		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) = self.get_local_profile_field(user_id, field).await {
+				profile.insert(key, value.value().into_owned());
+			}
+		}
+
+		// Insert all other profile fields
+		let mut all_fields = self.all_profile_keys(user_id);
+
+		while let Some((key, value)) = all_fields.next().await {
+			profile.insert(key, value);
+		}
+
+		profile
+	}
+
+	pub async fn get_local_profile_field(
+		&self,
+		user_id: &UserId,
+		field: ProfileFieldName,
+	) -> Option<ProfileFieldValue> {
+		let value = match field.clone() {
+			| ProfileFieldName::AvatarUrl => self
+				.avatar_url(user_id)
+				.await
+				.ok()
+				.map(to_value)
+				.transpose()
+				.expect("converting avatar url to value should succeed"),
+			| ProfileFieldName::DisplayName => self
+				.displayname(user_id)
+				.await
+				.ok()
+				.map(to_value)
+				.transpose()
+				.expect("converting displayname to value should succeed"),
+			| other => self.profile_key(user_id, other.as_str()).await.ok(),
+		}?;
+
+		Some(
+			ProfileFieldValue::new(field.as_str(), value)
+				.expect("local profile field should be valid"),
+		)
+	}
+
 	/// Returns the displayname of a user on this homeserver.
-	pub async fn displayname(&self, user_id: &UserId) -> conduwuit::Result<String> {
+	pub async fn displayname(&self, user_id: &UserId) -> Result<String> {
 		self.db.userid_displayname.get(user_id).await.deserialized()
 	}
 
 	/// Sets a new displayname or removes it if displayname is None. You still
 	/// need to notify all rooms of this change.
-	pub fn set_displayname(&self, user_id: &UserId, displayname: Option<String>) {
+	fn set_displayname(&self, user_id: &UserId, displayname: Option<String>) {
 		if let Some(displayname) = displayname {
 			self.db.userid_displayname.insert(user_id, displayname);
 		} else {
@@ -20,12 +266,12 @@ impl super::Service {
 	}
 
 	/// Get the `avatar_url` of a user.
-	pub async fn avatar_url(&self, user_id: &UserId) -> conduwuit::Result<OwnedMxcUri> {
+	pub async fn avatar_url(&self, user_id: &UserId) -> Result<OwnedMxcUri> {
 		self.db.userid_avatarurl.get(user_id).await.deserialized()
 	}
 
 	/// Sets a new avatar_url or removes it if avatar_url is None.
-	pub fn set_avatar_url(&self, user_id: &UserId, avatar_url: Option<OwnedMxcUri>) {
+	fn set_avatar_url(&self, user_id: &UserId, avatar_url: Option<OwnedMxcUri>) {
 		match avatar_url {
 			| Some(avatar_url) => {
 				self.db.userid_avatarurl.insert(user_id, &avatar_url);
@@ -37,11 +283,7 @@ impl super::Service {
 	}
 
 	/// Gets a specific user profile key
-	pub async fn profile_key(
-		&self,
-		user_id: &UserId,
-		profile_key: &str,
-	) -> conduwuit::Result<serde_json::Value> {
+	pub async fn profile_key(&self, user_id: &UserId, profile_key: &str) -> Result<Value> {
 		let key = (user_id, profile_key);
 		self.db
 			.useridprofilekey_value
@@ -54,7 +296,7 @@ impl super::Service {
 	pub fn all_profile_keys<'a>(
 		&'a self,
 		user_id: &'a UserId,
-	) -> impl Stream<Item = (String, serde_json::Value)> + 'a + Send {
+	) -> impl Stream<Item = (String, Value)> + 'a + Send {
 		type KeyVal<'a> = ((Ignore, String), &'a [u8]);
 
 		let prefix = (user_id, Interfix);
@@ -67,11 +309,11 @@ impl super::Service {
 	}
 
 	/// Sets a new profile key value, removes the key if value is None
-	pub fn set_profile_key(
+	fn set_profile_key(
 		&self,
 		user_id: &UserId,
 		profile_key: &str,
-		profile_key_value: Option<serde_json::Value>,
+		profile_key_value: Option<Value>,
 	) {
 		let key = (user_id, profile_key);