slightly dedupe pfp/displayname updates, update blurhash in PDUs too

Signed-off-by: strawberry <strawberry@puppygock.gay>

src/api/client/mod.rs

@@ -51,6 +51,7 @@ pub use membership::{join_room_by_id_helper, leave_all_rooms, leave_room, valida
 pub(super) use message::*;
 pub(super) use presence::*;
 pub(super) use profile::*;
+pub use profile::{update_all_rooms, update_avatar_url, update_displayname};
 pub(super) use push::*;
 pub(super) use read_marker::*;
 pub(super) use redact::*;

src/api/client/profile.rs

@@ -10,6 +10,7 @@ use ruma::{
 	},
 	events::{room::member::RoomMemberEventContent, StateEventType, TimelineEventType},
 	presence::PresenceState,
+	OwnedMxcUri, OwnedRoomId, OwnedUserId,
 };
 use serde_json::value::to_raw_value;
 use tracing::warn;
@@ -28,70 +29,14 @@ pub(crate) async fn set_displayname_route(
 	body: Ruma<set_display_name::v3::Request>,
 ) -> Result<set_display_name::v3::Response> {
 	let sender_user = body.sender_user.as_ref().expect("user is authenticated");
-
+	let all_joined_rooms: Vec<OwnedRoomId> = services()
-	services()
-		.users
-		.set_displayname(sender_user, body.displayname.clone())
-		.await?;
-
-	// Send a new membership event and presence update into all joined rooms
-	let all_rooms_joined: Vec<_> = services()
 		.rooms
 		.state_cache
 		.rooms_joined(sender_user)
 		.filter_map(Result::ok)
-		.map(|room_id| {
-			Ok::<_, Error>((
-				PduBuilder {
-					event_type: TimelineEventType::RoomMember,
-					content: to_raw_value(&RoomMemberEventContent {
-						displayname: body.displayname.clone(),
-						join_authorized_via_users_server: None,
-						..serde_json::from_str(
-							services()
-								.rooms
-								.state_accessor
-								.room_state_get(&room_id, &StateEventType::RoomMember, sender_user.as_str())?
-								.ok_or_else(|| {
-									Error::bad_database("Tried to send displayname update for user not in the room.")
-								})?
-								.content
-								.get(),
-						)
-						.map_err(|_| Error::bad_database("Database contains invalid PDU."))?
-					})
-					.expect("event is valid, we just created it"),
-					unsigned: None,
-					state_key: Some(sender_user.to_string()),
-					redacts: None,
-				},
-				room_id,
-			))
-		})
-		.filter_map(Result::ok)
 		.collect();
 
-	for (pdu_builder, room_id) in all_rooms_joined {
+	update_displayname(sender_user.clone(), body.displayname.clone(), all_joined_rooms).await?;
-		let mutex_state = Arc::clone(
-			services()
-				.globals
-				.roomid_mutex_state
-				.write()
-				.await
-				.entry(room_id.clone())
-				.or_default(),
-		);
-		let state_lock = mutex_state.lock().await;
-
-		if let Err(e) = services()
-			.rooms
-			.timeline
-			.build_and_append_pdu(pdu_builder, sender_user, &room_id, &state_lock)
-			.await
-		{
-			warn!(%e, "Failed to update/send new display name in room");
-		}
-	}
 
 	if services().globals.allow_local_presence() {
 		// Presence update
@@ -168,75 +113,20 @@ pub(crate) async fn set_avatar_url_route(
 	body: Ruma<set_avatar_url::v3::Request>,
 ) -> Result<set_avatar_url::v3::Response> {
 	let sender_user = body.sender_user.as_ref().expect("user is authenticated");
-
+	let all_joined_rooms: Vec<OwnedRoomId> = services()
-	services()
-		.users
-		.set_avatar_url(sender_user, body.avatar_url.clone())
-		.await?;
-
-	services()
-		.users
-		.set_blurhash(sender_user, body.blurhash.clone())
-		.await?;
-
-	// Send a new membership event and presence update into all joined rooms
-	let all_joined_rooms: Vec<_> = services()
 		.rooms
 		.state_cache
 		.rooms_joined(sender_user)
 		.filter_map(Result::ok)
-		.map(|room_id| {
-			Ok::<_, Error>((
-				PduBuilder {
-					event_type: TimelineEventType::RoomMember,
-					content: to_raw_value(&RoomMemberEventContent {
-						avatar_url: body.avatar_url.clone(),
-						join_authorized_via_users_server: None,
-						..serde_json::from_str(
-							services()
-								.rooms
-								.state_accessor
-								.room_state_get(&room_id, &StateEventType::RoomMember, sender_user.as_str())?
-								.ok_or_else(|| {
-									Error::bad_database("Tried to send displayname update for user not in the room.")
-								})?
-								.content
-								.get(),
-						)
-						.map_err(|_| Error::bad_database("Database contains invalid PDU."))?
-					})
-					.expect("event is valid, we just created it"),
-					unsigned: None,
-					state_key: Some(sender_user.to_string()),
-					redacts: None,
-				},
-				room_id,
-			))
-		})
-		.filter_map(Result::ok)
 		.collect();
 
-	for (pdu_builder, room_id) in all_joined_rooms {
+	update_avatar_url(
-		let mutex_state = Arc::clone(
+		sender_user.clone(),
-			services()
+		body.avatar_url.clone(),
-				.globals
+		body.blurhash.clone(),
-				.roomid_mutex_state
+		all_joined_rooms,
-				.write()
+	)
-				.await
+	.await?;
-				.entry(room_id.clone())
-				.or_default(),
-		);
-		let state_lock = mutex_state.lock().await;
-
-		if let Err(e) = services()
-			.rooms
-			.timeline
-			.build_and_append_pdu(pdu_builder, sender_user, &room_id, &state_lock)
-			.await
-		{
-			warn!(%e, "Failed to set/update room with new avatar URL / pfp");
-		}
-	}
 
 	if services().globals.allow_local_presence() {
 		// Presence update
@@ -363,3 +253,126 @@ pub(crate) async fn get_profile_route(body: Ruma<get_profile::v3::Request>) -> R
 		displayname: services().users.displayname(&body.user_id)?,
 	})
 }
+
+pub async fn update_displayname(
+	user_id: OwnedUserId, displayname: Option<String>, all_joined_rooms: Vec<OwnedRoomId>,
+) -> Result<()> {
+	services()
+		.users
+		.set_displayname(&user_id, displayname.clone())
+		.await?;
+
+	// Send a new join membership event into all joined rooms
+	let all_joined_rooms: Vec<_> = all_joined_rooms
+		.iter()
+		.map(|room_id| {
+			Ok::<_, Error>((
+				PduBuilder {
+					event_type: TimelineEventType::RoomMember,
+					content: to_raw_value(&RoomMemberEventContent {
+						displayname: displayname.clone(),
+						join_authorized_via_users_server: None,
+						..serde_json::from_str(
+							services()
+								.rooms
+								.state_accessor
+								.room_state_get(room_id, &StateEventType::RoomMember, user_id.as_str())?
+								.ok_or_else(|| {
+									Error::bad_database("Tried to send display name update for user not in the room.")
+								})?
+								.content
+								.get(),
+						)
+						.map_err(|_| Error::bad_database("Database contains invalid PDU."))?
+					})
+					.expect("event is valid, we just created it"),
+					unsigned: None,
+					state_key: Some(user_id.to_string()),
+					redacts: None,
+				},
+				room_id,
+			))
+		})
+		.filter_map(Result::ok)
+		.collect();
+
+	update_all_rooms(all_joined_rooms, user_id).await;
+
+	Ok(())
+}
+
+pub async fn update_avatar_url(
+	user_id: OwnedUserId, avatar_url: Option<OwnedMxcUri>, blurhash: Option<String>, all_joined_rooms: Vec<OwnedRoomId>,
+) -> Result<()> {
+	services()
+		.users
+		.set_avatar_url(&user_id, avatar_url.clone())
+		.await?;
+	services()
+		.users
+		.set_blurhash(&user_id, blurhash.clone())
+		.await?;
+
+	// Send a new join membership event into all joined rooms
+	let all_joined_rooms: Vec<_> = all_joined_rooms
+		.iter()
+		.map(|room_id| {
+			Ok::<_, Error>((
+				PduBuilder {
+					event_type: TimelineEventType::RoomMember,
+					content: to_raw_value(&RoomMemberEventContent {
+						avatar_url: avatar_url.clone(),
+						blurhash: blurhash.clone(),
+						join_authorized_via_users_server: None,
+						..serde_json::from_str(
+							services()
+								.rooms
+								.state_accessor
+								.room_state_get(room_id, &StateEventType::RoomMember, user_id.as_str())?
+								.ok_or_else(|| {
+									Error::bad_database("Tried to send avatar URL update for user not in the room.")
+								})?
+								.content
+								.get(),
+						)
+						.map_err(|_| Error::bad_database("Database contains invalid PDU."))?
+					})
+					.expect("event is valid, we just created it"),
+					unsigned: None,
+					state_key: Some(user_id.to_string()),
+					redacts: None,
+				},
+				room_id,
+			))
+		})
+		.filter_map(Result::ok)
+		.collect();
+
+	update_all_rooms(all_joined_rooms, user_id).await;
+
+	Ok(())
+}
+
+pub async fn update_all_rooms(all_joined_rooms: Vec<(PduBuilder, &OwnedRoomId)>, user_id: OwnedUserId) {
+	for (pdu_builder, room_id) in all_joined_rooms {
+		let mutex_state = Arc::clone(
+			services()
+				.globals
+				.roomid_mutex_state
+				.write()
+				.await
+				.entry(room_id.clone())
+				.or_default(),
+		);
+		let state_lock = mutex_state.lock().await;
+
+		if let Err(e) = services()
+			.rooms
+			.timeline
+			.build_and_append_pdu(pdu_builder, &user_id, room_id, &state_lock)
+			.await
+		{
+			warn!(%user_id, %room_id, %e, "Failed to update/send new profile join membership update in room");
+		}
+	}
+}

src/service/users/data.rs

@@ -307,7 +307,7 @@ impl Data for KeyValueDatabase {
 			.transpose()
 	}
 
-	/// Sets a new avatar_url or removes it if avatar_url is None.
+	/// Sets a new blurhash or removes it if blurhash is None.
 	fn set_blurhash(&self, user_id: &UserId, blurhash: Option<String>) -> Result<()> {
 		if let Some(blurhash) = blurhash {
 			self.userid_blurhash

src/service/users/mod.rs

@@ -299,7 +299,7 @@ impl Service {
 	/// Get the blurhash of a user.
 	pub fn blurhash(&self, user_id: &UserId) -> Result<Option<String>> { self.db.blurhash(user_id) }
 
-	/// Sets a new avatar_url or removes it if avatar_url is None.
+	/// Sets a new blurhash or removes it if blurhash is None.
 	pub async fn set_blurhash(&self, user_id: &UserId, blurhash: Option<String>) -> Result<()> {
 		self.db.set_blurhash(user_id, blurhash)
 	}