From 084facf474f222b04992d541477822273e8ffced Mon Sep 17 00:00:00 2001 From: Jason Volk Date: Fri, 31 Oct 2025 20:22:20 +0000 Subject: [PATCH] Move user profile related functions to profile unit. Signed-off-by: Jason Volk --- src/service/users/mod.rs | 180 +------------------------ src/service/users/profile.rs | 254 ++++++++++++++++++++++++++++++----- 2 files changed, 224 insertions(+), 210 deletions(-) diff --git a/src/service/users/mod.rs b/src/service/users/mod.rs index 5e193dad..8e7e8bcc 100644 --- a/src/service/users/mod.rs +++ b/src/service/users/mod.rs @@ -6,21 +6,17 @@ mod profile; use std::sync::Arc; -use futures::{FutureExt, Stream, StreamExt, TryFutureExt, TryStreamExt, future::join3}; +use futures::{Stream, StreamExt, TryFutureExt}; use ruma::{ - OwnedMxcUri, OwnedRoomId, OwnedUserId, UserId, + OwnedRoomId, OwnedUserId, UserId, api::client::filter::FilterDefinition, - events::{ - GlobalAccountDataEventType, - ignored_user_list::IgnoredUserListEvent, - room::member::{MembershipState, RoomMemberEventContent}, - }, + events::{GlobalAccountDataEventType, ignored_user_list::IgnoredUserListEvent}, }; use tuwunel_core::{ Err, Result, debug_warn, err, is_equal_to, pdu::PduBuilder, trace, - utils::{self, IterStream, ReadyExt, TryFutureExtExt, stream::TryIgnore}, + utils::{self, ReadyExt, stream::TryIgnore}, warn, }; use tuwunel_database::{Deserialized, Json, Map}; @@ -260,68 +256,6 @@ impl Service { Ok(()) } - /// Returns the displayname of a user on this homeserver. - pub async fn displayname(&self, user_id: &UserId) -> Result { - 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) { - if let Some(displayname) = displayname { - self.db - .userid_displayname - .insert(user_id, displayname); - } else { - self.db.userid_displayname.remove(user_id); - } - } - - /// Get the `avatar_url` of a user. - pub async fn avatar_url(&self, user_id: &UserId) -> Result { - 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) { - match avatar_url { - | Some(avatar_url) => { - self.db - .userid_avatarurl - .insert(user_id, &avatar_url); - }, - | _ => { - self.db.userid_avatarurl.remove(user_id); - }, - } - } - - /// Get the blurhash of a user. - pub async fn blurhash(&self, user_id: &UserId) -> Result { - self.db - .userid_blurhash - .get(user_id) - .await - .deserialized() - } - - /// Sets a new avatar_url or removes it if avatar_url is None. - pub fn set_blurhash(&self, user_id: &UserId, blurhash: Option) { - if let Some(blurhash) = blurhash { - self.db.userid_blurhash.insert(user_id, blurhash); - } else { - self.db.userid_blurhash.remove(user_id); - } - } - /// Creates a new sync filter. Returns the filter id. #[must_use] pub fn create_filter(&self, user_id: &UserId, filter: &FilterDefinition) -> String { @@ -450,112 +384,6 @@ impl Service { Err!(FeatureDisabled("ldap")) } - pub async fn update_displayname( - &self, - user_id: &UserId, - displayname: Option, - rooms: &[OwnedRoomId], - ) { - let (current_avatar_url, current_blurhash, current_displayname) = join3( - self.services.users.avatar_url(user_id).ok(), - self.services.users.blurhash(user_id).ok(), - self.services.users.displayname(user_id).ok(), - ) - .await; - - if displayname == current_displayname { - return; - } - - self.services - .users - .set_displayname(user_id, displayname.clone()); - - // Send a new join membership event into rooms - let avatar_url = ¤t_avatar_url; - let blurhash = ¤t_blurhash; - let displayname = &displayname; - let rooms: Vec<_> = rooms - .iter() - .try_stream() - .and_then(async |room_id: &OwnedRoomId| { - let pdu = PduBuilder::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, - }); - - Ok((pdu, room_id)) - }) - .ignore_err() - .collect() - .await; - - self.update_all_rooms(user_id, rooms) - .boxed() - .await; - } - - pub async fn update_avatar_url( - &self, - user_id: &UserId, - avatar_url: Option, - blurhash: Option, - rooms: &[OwnedRoomId], - ) { - let (current_avatar_url, current_blurhash, current_displayname) = join3( - self.services.users.avatar_url(user_id).ok(), - self.services.users.blurhash(user_id).ok(), - self.services.users.displayname(user_id).ok(), - ) - .await; - - if current_avatar_url == avatar_url && current_blurhash == blurhash { - return; - } - - self.services - .users - .set_avatar_url(user_id, avatar_url.clone()); - self.services - .users - .set_blurhash(user_id, blurhash.clone()); - - // Send a new join membership event into rooms - let avatar_url = &avatar_url; - let blurhash = &blurhash; - let displayname = ¤t_displayname; - let rooms: Vec<_> = rooms - .iter() - .try_stream() - .and_then(async |room_id: &OwnedRoomId| { - let pdu = PduBuilder::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, - }); - - Ok((pdu, room_id)) - }) - .ignore_err() - .collect() - .await; - - self.update_all_rooms(user_id, rooms) - .boxed() - .await; - } - async fn update_all_rooms(&self, user_id: &UserId, rooms: Vec<(PduBuilder, &OwnedRoomId)>) { for (pdu_builder, room_id) in rooms { let state_lock = self.services.state.mutex.lock(room_id).await; diff --git a/src/service/users/profile.rs b/src/service/users/profile.rs index e0057a85..f35725bb 100644 --- a/src/service/users/profile.rs +++ b/src/service/users/profile.rs @@ -1,19 +1,222 @@ -use futures::{Stream, StreamExt, TryFutureExt}; -use ruma::UserId; -use tuwunel_core::{Result, implement, utils::stream::TryIgnore}; +use futures::{FutureExt, Stream, StreamExt, TryFutureExt, TryStreamExt, future::join3}; +use ruma::{ + OwnedMxcUri, OwnedRoomId, UserId, + events::room::member::{MembershipState, RoomMemberEventContent}, +}; +use tuwunel_core::{ + Result, implement, + matrix::PduBuilder, + utils::{ + future::TryExtExt, + stream::{IterStream, TryIgnore}, + }, +}; use tuwunel_database::{Deserialized, Ignore, Interfix, Json}; -/// Gets a specific user profile key #[implement(super::Service)] -pub async fn profile_key( +pub async fn update_displayname( &self, user_id: &UserId, - profile_key: &str, -) -> Result { - let key = (user_id, profile_key); + displayname: Option, + rooms: &[OwnedRoomId], +) { + let (current_avatar_url, current_blurhash, current_displayname) = join3( + self.services.users.avatar_url(user_id).ok(), + self.services.users.blurhash(user_id).ok(), + self.services.users.displayname(user_id).ok(), + ) + .await; + + if displayname == current_displayname { + return; + } + + self.services + .users + .set_displayname(user_id, displayname.clone()); + + // Send a new join membership event into rooms + let avatar_url = ¤t_avatar_url; + let blurhash = ¤t_blurhash; + let displayname = &displayname; + let rooms: Vec<_> = rooms + .iter() + .try_stream() + .and_then(async |room_id: &OwnedRoomId| { + let pdu = PduBuilder::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, + }); + + Ok((pdu, room_id)) + }) + .ignore_err() + .collect() + .await; + + self.update_all_rooms(user_id, rooms) + .boxed() + .await; +} + +/// Sets a new displayname or removes it if displayname is None. You still +/// need to notify all rooms of this change. +#[implement(super::Service)] +pub fn set_displayname(&self, user_id: &UserId, displayname: Option) { + if let Some(displayname) = displayname { + self.db + .userid_displayname + .insert(user_id, displayname); + } else { + self.db.userid_displayname.remove(user_id); + } +} + +/// Returns the displayname of a user on this homeserver. +#[implement(super::Service)] +pub async fn displayname(&self, user_id: &UserId) -> Result { + self.db + .userid_displayname + .get(user_id) + .await + .deserialized() +} + +#[implement(super::Service)] +pub async fn update_avatar_url( + &self, + user_id: &UserId, + avatar_url: Option, + blurhash: Option, + rooms: &[OwnedRoomId], +) { + let (current_avatar_url, current_blurhash, current_displayname) = join3( + self.services.users.avatar_url(user_id).ok(), + self.services.users.blurhash(user_id).ok(), + self.services.users.displayname(user_id).ok(), + ) + .await; + + if current_avatar_url == avatar_url && current_blurhash == blurhash { + return; + } + + self.services + .users + .set_avatar_url(user_id, avatar_url.clone()); + self.services + .users + .set_blurhash(user_id, blurhash.clone()); + + // Send a new join membership event into rooms + let avatar_url = &avatar_url; + let blurhash = &blurhash; + let displayname = ¤t_displayname; + let rooms: Vec<_> = rooms + .iter() + .try_stream() + .and_then(async |room_id: &OwnedRoomId| { + let pdu = PduBuilder::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, + }); + + Ok((pdu, room_id)) + }) + .ignore_err() + .collect() + .await; + + self.update_all_rooms(user_id, rooms) + .boxed() + .await; +} + +/// Sets a new avatar_url or removes it if avatar_url is None. +#[implement(super::Service)] +pub fn set_avatar_url(&self, user_id: &UserId, avatar_url: Option) { + match avatar_url { + | Some(avatar_url) => { + self.db + .userid_avatarurl + .insert(user_id, &avatar_url); + }, + | _ => { + self.db.userid_avatarurl.remove(user_id); + }, + } +} + +/// Get the `avatar_url` of a user. +#[implement(super::Service)] +pub async fn avatar_url(&self, user_id: &UserId) -> Result { + self.db + .userid_avatarurl + .get(user_id) + .await + .deserialized() +} + +/// Sets a new avatar_url or removes it if avatar_url is None. +#[implement(super::Service)] +pub fn set_blurhash(&self, user_id: &UserId, blurhash: Option) { + if let Some(blurhash) = blurhash { + self.db.userid_blurhash.insert(user_id, blurhash); + } else { + self.db.userid_blurhash.remove(user_id); + } +} + +/// Get the blurhash of a user. +#[implement(super::Service)] +pub async fn blurhash(&self, user_id: &UserId) -> Result { + self.db + .userid_blurhash + .get(user_id) + .await + .deserialized() +} + +/// Sets a new timezone or removes it if timezone is None. +#[implement(super::Service)] +pub fn set_timezone(&self, user_id: &UserId, timezone: Option) { + // TODO: insert to the stable MSC4175 key when it's stable + let key = (user_id, "us.cloke.msc4175.tz"); + + if let Some(timezone) = timezone { + self.db + .useridprofilekey_value + .put_raw(key, &timezone); + } else { + self.db.useridprofilekey_value.del(key); + } +} + +/// Get the timezone of a user. +#[implement(super::Service)] +pub async fn timezone(&self, user_id: &UserId) -> Result { + // TODO: transparently migrate unstable key usage to the stable key once MSC4133 + // and MSC4175 are stable, likely a remove/insert in this block. + + // first check the unstable prefix then check the stable prefix + let unstable_key = (user_id, "us.cloke.msc4175.tz"); + let stable_key = (user_id, "m.tz"); self.db .useridprofilekey_value - .qry(&key) + .qry(&unstable_key) + .or_else(|_| self.db.useridprofilekey_value.qry(&stable_key)) .await .deserialized() } @@ -54,34 +257,17 @@ pub fn set_profile_key( } } -/// Get the timezone of a user. +/// Gets a specific user profile key #[implement(super::Service)] -pub async fn timezone(&self, user_id: &UserId) -> Result { - // TODO: transparently migrate unstable key usage to the stable key once MSC4133 - // and MSC4175 are stable, likely a remove/insert in this block. - - // first check the unstable prefix then check the stable prefix - let unstable_key = (user_id, "us.cloke.msc4175.tz"); - let stable_key = (user_id, "m.tz"); +pub async fn profile_key( + &self, + user_id: &UserId, + profile_key: &str, +) -> Result { + let key = (user_id, profile_key); self.db .useridprofilekey_value - .qry(&unstable_key) - .or_else(|_| self.db.useridprofilekey_value.qry(&stable_key)) + .qry(&key) .await .deserialized() } - -/// Sets a new timezone or removes it if timezone is None. -#[implement(super::Service)] -pub fn set_timezone(&self, user_id: &UserId, timezone: Option) { - // TODO: insert to the stable MSC4175 key when it's stable - let key = (user_id, "us.cloke.msc4175.tz"); - - if let Some(timezone) = timezone { - self.db - .useridprofilekey_value - .put_raw(key, &timezone); - } else { - self.db.useridprofilekey_value.del(key); - } -}