From 53d6887b8648dfed89d0f4b6191b274029d2a9ea Mon Sep 17 00:00:00 2001 From: Ginger Date: Tue, 30 Jun 2026 16:02:14 -0400 Subject: [PATCH] refactor: Move profile field setting logic into users service --- src/api/client/profile.rs | 325 ++++------------------------------- src/api/server/query.rs | 15 +- src/service/users/account.rs | 12 +- src/service/users/mod.rs | 9 +- src/service/users/profile.rs | 270 +++++++++++++++++++++++++++-- 5 files changed, 319 insertions(+), 312 deletions(-) diff --git a/src/api/client/profile.rs b/src/api/client/profile.rs index b611e31de..a8dae57ff 100644 --- a/src/api/client/profile.rs +++ b/src/api/client/profile.rs @@ -1,9 +1,8 @@ use std::collections::BTreeMap; use axum::extract::State; -use conduwuit::{Err, Result, matrix::pdu::PartialPdu, utils::to_canonical_object}; +use conduwuit::{Err, Result}; use conduwuit_service::Services; -use futures::StreamExt; use ruma::{ UserId, api::{ @@ -13,11 +12,10 @@ use ruma::{ federation, }, assign, - events::room::member::MembershipState, - presence::PresenceState, profile::{ProfileFieldName, ProfileFieldValue}, }; -use serde_json::{Value, to_value}; +use serde_json::Value; +use service::users::ProfileFieldChange; use crate::Ruma; @@ -65,13 +63,14 @@ pub(crate) async fn set_profile_field_route( 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()), - body.propagate_to.clone(), - ) - .await?; + services + .users + .set_profile_field( + &body.user_id, + ProfileFieldChange::Set(body.value.clone()), + body.propagate_to.clone(), + ) + .await?; Ok(set_profile_field::v3::Response::new()) } @@ -94,13 +93,14 @@ pub(crate) async fn delete_profile_field_route( 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()), - body.propagate_to.clone(), - ) - .await?; + services + .users + .set_profile_field( + &body.user_id, + ProfileFieldChange::Delete(body.field.clone()), + body.propagate_to.clone(), + ) + .await?; Ok(delete_profile_field::v3::Response::new()) } @@ -111,7 +111,7 @@ async fn fetch_full_profile( ) -> Option> { // If the user exists locally, fetch their local profile if services.users.status(user_id).await.is_found() { - return Some(get_local_profile(services, user_id).await); + return Some(services.users.get_local_profile(user_id).await); } // Otherwise ask their homeserver @@ -135,13 +135,10 @@ async fn fetch_full_profile( continue; }; - let _ = set_profile_field( - services, - user_id, - ProfileFieldChange::Set(value), - PropagateTo::None, - ) - .await; + let _ = services + .users + .set_profile_field(user_id, ProfileFieldChange::Set(value), PropagateTo::None) + .await; } Some(BTreeMap::from_iter(response)) @@ -154,7 +151,7 @@ async fn fetch_profile_field( ) -> Result> { // 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); + return Ok(services.users.get_local_profile_field(user_id, field).await); } // Otherwise ask their homeserver @@ -175,13 +172,14 @@ async fn fetch_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) { - let _ = set_profile_field( - services, - user_id, - ProfileFieldChange::Set(value.clone()), - PropagateTo::None, - ) - .await; + let _ = services + .users + .set_profile_field( + user_id, + ProfileFieldChange::Set(value.clone()), + PropagateTo::None, + ) + .await; Ok(Some(value)) } else { @@ -190,260 +188,11 @@ async fn fetch_profile_field( ))) } } else { - let _ = set_profile_field( - services, - user_id, - ProfileFieldChange::Delete(field), - PropagateTo::None, - ) - .await; + let _ = services + .users + .set_profile_field(user_id, ProfileFieldChange::Delete(field), PropagateTo::None) + .await; Ok(None) } } - -pub(crate) async fn get_local_profile( - services: &Services, - user_id: &UserId, -) -> BTreeMap { - 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()); - } - } - - // Insert all other profile fields - let mut all_fields = services.users.all_profile_keys(user_id); - - while let Some((key, value)) = all_fields.next().await { - profile.insert(key, value); - } - - profile -} - -pub(crate) async fn get_local_profile_field( - services: &Services, - user_id: &UserId, - field: ProfileFieldName, -) -> Option { - 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 { - | &Self::Delete(ref name) => name.clone(), - | &Self::Set(ref value) => value.field_name(), - } - } - - fn value(&self) -> Option { - if let Self::Set(value) = self { - Some(value.value().into_owned()) - } else { - None - } - } -} - -async fn set_profile_field( - services: &Services, - 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 = get_local_profile(services, 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) - && services.globals.user_is_local(user_id) - { - let current_displayname = services.users.displayname(user_id).await.ok(); - let current_avatar_url = services.users.avatar_url(user_id).await.ok(); - - let mut all_joined_rooms = services.rooms.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 = services - .rooms - .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 = services.rooms.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 _ = services - .rooms - .timeline - .build_and_append_pdu( - PartialPdu::state(user_id.to_string(), ¤t_membership), - user_id, - Some(&room_id), - &state_lock, - ) - .await; - } - - 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; - } - } - - 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 => - services - .users - .set_profile_key(user_id, other.field_name().as_str(), other.value()), - } - - Ok(()) -} diff --git a/src/api/server/query.rs b/src/api/server/query.rs index ff19d8214..dccbce067 100644 --- a/src/api/server/query.rs +++ b/src/api/server/query.rs @@ -7,10 +7,7 @@ use ruma::{ api::federation::query::{get_profile_information, get_room_information}, }; -use crate::{ - Ruma, - client::{get_local_profile, get_local_profile_field}, -}; +use crate::Ruma; /// # `GET /_matrix/federation/v1/query/directory` /// @@ -75,15 +72,19 @@ pub(crate) async fn get_profile_information_route( let response = if let Some(field) = &body.field { let mut response = get_profile_information::v1::Response::new(); - if let Some(value) = - get_local_profile_field(&services, &body.user_id, field.to_owned()).await + if let Some(value) = services + .users + .get_local_profile_field(&body.user_id, field.to_owned()) + .await { response.set(value.field_name().as_str().to_owned(), value.value().into_owned()); } response } else { - get_local_profile(&services, &body.user_id) + services + .users + .get_local_profile(&body.user_id) .await .into_iter() .collect() diff --git a/src/service/users/account.rs b/src/service/users/account.rs index 816ab974b..a12387780 100644 --- a/src/service/users/account.rs +++ b/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 diff --git a/src/service/users/mod.rs b/src/service/users/mod.rs index 734651e5a..155cdeed4 100644 --- a/src/service/users/mod.rs +++ b/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, membership: Dep, oauth: Dep, + presence: Dep, + state: Dep, state_accessor: Dep, state_cache: Dep, threepid: Dep, + timeline: Dep, } struct Data { @@ -110,10 +114,13 @@ impl crate::Service for Service { globals: args.depend::("globals"), membership: args.depend::("membership"), oauth: args.depend::("oauth"), + presence: args.depend::("presence"), + state: args.depend::("rooms::state"), state_accessor: args .depend::("rooms::state_accessor"), state_cache: args.depend::("rooms::state_cache"), threepid: args.depend::("threepid"), + timeline: args.depend::("rooms::timeline"), }, db: Data { keychangeid_userid: args.db["keychangeid_userid"].clone(), diff --git a/src/service/users/profile.rs b/src/service/users/profile.rs index a4ed3ee3d..2c5444795 100644 --- a/src/service/users/profile.rs +++ b/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 { + 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(), ¤t_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 { + 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 { + 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 { + 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) { + fn set_displayname(&self, user_id: &UserId, displayname: Option) { 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 { + 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) { + 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); @@ -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 { + pub async fn profile_key(&self, user_id: &UserId, profile_key: &str) -> Result { 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 + 'a + Send { + ) -> impl Stream + '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, + profile_key_value: Option, ) { let key = (user_id, profile_key);