refactor: Split the users service into several modules

Also resolvers several lints, including needless async on create_user

src/admin/user/commands.rs

@@ -69,8 +69,7 @@ pub(super) async fn create_user(&self, username: String, password: Option<String
 		// Create user
 		self.services
 			.users
-			.create(&user_id, Some(HashedPassword::new(&password)?))
-			.await?;
+			.create(&user_id, Some(HashedPassword::new(&password)?))?;
 
 		// Default to pretty displayname
 		let mut displayname = user_id.localpart().to_owned();

src/api/client/account/register.rs

@@ -134,7 +134,7 @@ pub(crate) async fn register_route(
 	};
 
 	// Create user
-	services.users.create(&user_id, password).await?;
+	services.users.create(&user_id, password)?;
 
 	// Set an initial display name
 	let mut displayname = user_id.localpart().to_owned();

src/service/admin/create.rs

@@ -48,7 +48,7 @@ pub async fn create_admin_room(services: &Services) -> Result {
 
 	// Create a user for the server
 	let server_user = services.globals.server_user.as_ref();
-	services.users.create(server_user, None).await?;
+	services.users.create(server_user, None)?;
 
 	let mut create_content = if room_version_rules.authorization.use_room_create_sender {
 		RoomCreateEventContent::new_v1(server_user.into())

src/service/appservice/mod.rs

@@ -109,10 +109,7 @@ async fn start_appservice(&self, id: String, registration: Registration) -> Resu
 		)?;
 
 		if !self.services.users.exists(&appservice_user_id).await {
-			self.services
-				.users
-				.create(&appservice_user_id, None)
-				.await?;
+			self.services.users.create(&appservice_user_id, None)?;
 		} else if self
 			.services
 			.users

src/service/rooms/state_cache/update.rs

@@ -38,7 +38,7 @@ pub async fn update_membership(
 		if !self.services.globals.user_is_local(user_id)
 			&& !self.services.users.exists(user_id).await
 		{
-			self.services.users.create(user_id, None).await?;
+			self.services.users.create(user_id, None)?;
 		}
 
 		match &membership.membership {

src/service/users/account.rs

@@ -0,0 +1,393 @@
+use conduwuit::{
+	Err, debug_error, debug_warn, err, trace, utils,
+	utils::{ReadyExt, stream::TryIgnore},
+};
+use database::{Deserialized, Json};
+use futures::{Stream, StreamExt, TryFutureExt};
+use ruma::{
+	MilliSecondsSinceUnixEpoch, OwnedDeviceId, OwnedUserId, UserId,
+	events::{GlobalAccountDataEventType, ignored_user_list::IgnoredUserListEvent},
+};
+use ruminuwuity::invite_permission_config::{FilterLevel, InvitePermissionConfigEvent};
+
+use crate::users::{HashedPassword, UserSuspension};
+
+impl super::Service {
+	/// Returns true/false based on whether the recipient/receiving user has
+	/// ignored the sender.
+	pub async fn user_is_ignored(&self, sender_user: &UserId, recipient_user: &UserId) -> bool {
+		self.services
+			.account_data
+			.get_global(recipient_user, GlobalAccountDataEventType::IgnoredUserList)
+			.await
+			.is_ok_and(|ignored: IgnoredUserListEvent| {
+				ignored
+					.content
+					.ignored_users
+					.keys()
+					.any(|blocked_user| blocked_user == sender_user)
+			})
+	}
+
+	/// Returns the recipient's filter level for an invite from the sender.
+	///
+	/// If the sender is ignored by the recipient, `Ignore` is returned.
+	/// Otherwise, the resulting value depends on their invite blocking or
+	/// invite filtering configuration.
+	pub async fn invite_filter_level(
+		&self,
+		sender_user: &UserId,
+		recipient_user: &UserId,
+	) -> FilterLevel {
+		if self.user_is_ignored(sender_user, recipient_user).await {
+			FilterLevel::Ignore
+		} else {
+			let (stable, unstable) = tokio::join!(
+				self.services
+					.account_data
+					.get_global::<InvitePermissionConfigEvent>(
+						recipient_user,
+						GlobalAccountDataEventType::InvitePermissionConfig
+					),
+				self.services
+					.account_data
+					.get_global::<InvitePermissionConfigEvent>(
+						recipient_user,
+						"org.matrix.msc4155.invite_permission_config".into()
+					) // TODO: MSC4155 probably needs upstreaming to ruma at some point
+			);
+			if stable.is_err() && unstable.is_err() {
+				return FilterLevel::Allow;
+			}
+			stable
+				.unwrap_or_else(|_| unstable.unwrap())
+				.content
+				.user_filter_level(sender_user)
+		}
+	}
+
+	/// Check if a user is an admin
+	#[inline]
+	pub async fn is_admin(&self, user_id: &UserId) -> bool {
+		self.services.admin.user_is_admin(user_id).await
+	}
+
+	/// Create a new user account on this homeserver. Set the password to `None`
+	/// to create a non-local user. Non-local users with a password will return
+	/// an error.
+	#[inline]
+	pub fn create(
+		&self,
+		user_id: &UserId,
+		password: Option<HashedPassword>,
+	) -> conduwuit::Result<()> {
+		if !self.services.globals.user_is_local(user_id) && password.is_some() {
+			return Err!("Cannot create a nonlocal user with a set password");
+		}
+
+		self.set_password(user_id, password);
+
+		Ok(())
+	}
+
+	/// Deactivates an account, removing all of their device IDs and unsetting
+	/// their password.
+	pub async fn deactivate_account(&self, user_id: &UserId) -> conduwuit::Result<()> {
+		// Remove all associated devices
+		self.all_device_ids(user_id)
+			.for_each(async |device_id| self.remove_device(user_id, &device_id).await)
+			.await;
+
+		// Set the password to "" to indicate a deactivated account. Hashes will never
+		// result in an empty string, so the user will not be able to log in again.
+		// Systems like changing the password without logging in should check if the
+		// account is deactivated.
+		self.set_password(user_id, None);
+
+		// TODO: Unhook 3PID
+		Ok(())
+	}
+
+	/// Suspend account, placing it in a read-only state
+	pub async fn suspend_account(&self, user_id: &UserId, suspending_user: &UserId) {
+		self.db.userid_suspension.raw_put(
+			user_id,
+			Json(UserSuspension {
+				suspended: true,
+				suspended_at: MilliSecondsSinceUnixEpoch::now().get().into(),
+				suspended_by: suspending_user.to_string(),
+			}),
+		);
+	}
+
+	/// Unsuspend account, placing it in a read-write state
+	pub async fn unsuspend_account(&self, user_id: &UserId) {
+		self.db.userid_suspension.remove(user_id);
+	}
+
+	/// Locks an account, preventing it being used until it is unlocked.
+	pub async fn lock_account(&self, user_id: &UserId, locking_user: &UserId) {
+		// NOTE: Locking is basically just suspension with a more severe effect,
+		// so we'll just re-use the suspension data structure to store the lock state.
+		let suspension = self
+			.db
+			.userid_lock
+			.get(user_id)
+			.await
+			.deserialized::<UserSuspension>()
+			.unwrap_or_else(|_| UserSuspension {
+				suspended: true,
+				suspended_at: MilliSecondsSinceUnixEpoch::now().get().into(),
+				suspended_by: locking_user.to_string(),
+			});
+
+		self.db.userid_lock.raw_put(user_id, Json(suspension));
+	}
+
+	/// Unlocks an account, allowing the user to log in and use it again.
+	pub async fn unlock_account(&self, user_id: &UserId) { self.db.userid_lock.remove(user_id); }
+
+	/// Check if the provided user ID belongs to an existing (possibly
+	/// deactivated) account on this homeserver.
+	#[inline]
+	pub async fn exists(&self, user_id: &UserId) -> bool {
+		self.services.globals.user_is_local(user_id)
+			&& self.db.userid_password.get(user_id).await.is_ok()
+	}
+
+	/// Check if account is deactivated (has an empty password). Returns a
+	/// NotFound error if the user does not exist.
+	pub async fn is_deactivated(&self, user_id: &UserId) -> conduwuit::Result<bool> {
+		self.db
+			.userid_password
+			.get(user_id)
+			.map_ok(|val| val.is_empty())
+			.map_err(|_| err!(Request(NotFound("User does not exist."))))
+			.await
+	}
+
+	/// Check if account is suspended. Returns false if the user does not exist.
+	pub async fn is_suspended(&self, user_id: &UserId) -> conduwuit::Result<bool> {
+		match self
+			.db
+			.userid_suspension
+			.get(user_id)
+			.await
+			.deserialized::<UserSuspension>()
+		{
+			| Ok(s) => Ok(s.suspended),
+			| Err(e) =>
+				if e.is_not_found() {
+					Ok(false)
+				} else {
+					Err(e)
+				},
+		}
+	}
+
+	/// Returns true if the user is locked. Returns false if the user does not
+	/// exist or is not locked.
+	pub async fn is_locked(&self, user_id: &UserId) -> conduwuit::Result<bool> {
+		match self
+			.db
+			.userid_lock
+			.get(user_id)
+			.await
+			.deserialized::<UserSuspension>()
+		{
+			| Ok(s) => Ok(s.suspended),
+			| Err(e) =>
+				if e.is_not_found() {
+					Ok(false)
+				} else {
+					Err(e)
+				},
+		}
+	}
+
+	/// Disables login for a user, preventing them from creating new devices,
+	/// but allows them to continue using their existing sessions unimpeded.
+	pub fn disable_login(&self, user_id: &UserId) {
+		self.db.userid_logindisabled.insert(user_id, "");
+	}
+
+	/// Re-enables login for a user, allowing them to create new devices again.
+	pub fn enable_login(&self, user_id: &UserId) { self.db.userid_logindisabled.remove(user_id); }
+
+	/// Returns true if the target user's login is disabled.
+	pub async fn is_login_disabled(&self, user_id: &UserId) -> bool {
+		self.db
+			.userid_logindisabled
+			.exists(user_id.as_str())
+			.await
+			.is_ok()
+	}
+
+	/// Check if account is active (not deactivated)
+	pub async fn is_active(&self, user_id: &UserId) -> bool {
+		!self.is_deactivated(user_id).await.unwrap_or(true)
+	}
+
+	/// Check if account is a local user, and is active (not deactivated)
+	pub async fn is_active_local(&self, user_id: &UserId) -> bool {
+		self.services.globals.user_is_local(user_id) && self.is_active(user_id).await
+	}
+
+	/// Returns the number of users registered on this server, including
+	/// deactivated users.
+	#[inline]
+	pub async fn count(&self) -> usize { self.db.userid_password.count().await }
+
+	/// Find out which user an access token belongs to. Will panic if the access
+	/// token is empty.
+	pub async fn find_from_token(
+		&self,
+		token: &str,
+	) -> conduwuit::Result<(OwnedUserId, OwnedDeviceId)> {
+		assert!(!token.is_empty(), "Empty access token");
+		self.db.token_userdeviceid.get(token).await.deserialized()
+	}
+
+	/// Returns an iterator over all users on this homeserver.
+	pub fn stream(&self) -> impl Stream<Item = OwnedUserId> + Send {
+		self.db.userid_password.keys().ignore_err()
+	}
+
+	/// Returns a list of active local users.
+	///
+	/// A user account is considered `local` if the associated password is not
+	/// empty.
+	pub fn list_local_users(&self) -> impl Stream<Item = OwnedUserId> + Send + '_ {
+		self.db
+			.userid_password
+			.stream()
+			.ignore_err()
+			.ready_filter_map(|(u, p): (OwnedUserId, &[u8])| (!p.is_empty()).then_some(u))
+	}
+
+	/// Set a user's password.
+	pub fn set_password(&self, user_id: &UserId, password: Option<HashedPassword>) {
+		if let Some(hash) = password {
+			self.db.userid_password.insert(user_id, hash.0);
+		} else {
+			self.db.userid_password.insert(user_id, b"");
+		}
+	}
+
+	/// Check a user's password.
+	pub async fn check_password(
+		&self,
+		user_id: &UserId,
+		password: &str,
+	) -> conduwuit::Result<OwnedUserId> {
+		let (hash, user_id): (String, OwnedUserId) =
+			if let Ok(hash) = self.db.userid_password.get(user_id).await.deserialized() {
+				(hash, user_id.to_owned())
+			} else {
+				// We also check the lowercased version of the user ID to handle legacy user IDs
+				// better
+				let lowercase_user_id = UserId::parse(user_id.as_str().to_lowercase()).unwrap();
+
+				if let Ok(hash) = self
+					.db
+					.userid_password
+					.get(lowercase_user_id.as_str())
+					.await
+					.deserialized()
+				{
+					(hash, lowercase_user_id)
+				} else {
+					return Err!(Request(Forbidden("This user cannot log in with a password.")));
+				}
+			};
+
+		if hash.is_empty() {
+			return Err!(Request(UserDeactivated("This user is deactivated")));
+		}
+
+		utils::hash::verify_password(password, &hash)
+			.inspect_err(|e| debug_error!("{e}"))
+			.map_err(|_| err!(Request(Forbidden("Invalid identifier or password."))))?;
+
+		Ok(user_id)
+	}
+
+	/// Creates an OpenID token, which can be used to prove that a user has
+	/// access to an account (primarily for integrations)
+	pub fn create_openid_token(&self, user_id: &UserId, token: &str) -> conduwuit::Result<u64> {
+		use std::num::Saturating as Sat;
+
+		let expires_in = self.services.server.config.openid_token_ttl;
+		let expires_at = Sat(utils::millis_since_unix_epoch()) + Sat(expires_in) * Sat(1000);
+
+		let mut value = expires_at.0.to_be_bytes().to_vec();
+		value.extend_from_slice(user_id.as_bytes());
+
+		self.db
+			.openidtoken_expiresatuserid
+			.insert(token.as_bytes(), value.as_slice());
+
+		Ok(expires_in)
+	}
+
+	/// Find out which user an OpenID access token belongs to.
+	pub async fn find_from_openid_token(&self, token: &str) -> conduwuit::Result<OwnedUserId> {
+		let Ok(value) = self.db.openidtoken_expiresatuserid.get(token).await else {
+			return Err!(Request(Unauthorized("OpenID token is unrecognised")));
+		};
+
+		let (expires_at_bytes, user_bytes) = value.split_at(0_u64.to_be_bytes().len());
+		let expires_at =
+			u64::from_be_bytes(expires_at_bytes.try_into().map_err(|e| {
+				err!(Database("expires_at in openid_userid is invalid u64. {e}"))
+			})?);
+
+		if expires_at < utils::millis_since_unix_epoch() {
+			debug_warn!("OpenID token is expired, removing");
+			self.db.openidtoken_expiresatuserid.remove(token.as_bytes());
+
+			return Err!(Request(Unauthorized("OpenID token is expired")));
+		}
+
+		let user_string = utils::string_from_bytes(user_bytes)
+			.map_err(|e| err!(Database("User ID in openid_userid is invalid unicode. {e}")))?;
+
+		OwnedUserId::try_from(user_string)
+			.map_err(|e| err!(Database("User ID in openid_userid is invalid. {e}")))
+	}
+
+	/// Creates a short-lived login token, which can be used to log in using the
+	/// `m.login.token` mechanism.
+	pub fn create_login_token(&self, user_id: &UserId, token: &str) -> u64 {
+		use std::num::Saturating as Sat;
+
+		let expires_in = self.services.server.config.login_token_ttl;
+		let expires_at = Sat(utils::millis_since_unix_epoch()) + Sat(expires_in);
+
+		let value = (expires_at.0, user_id);
+		self.db.logintoken_expiresatuserid.raw_put(token, value);
+
+		expires_in
+	}
+
+	/// Find out which user a login token belongs to.
+	/// Removes the token to prevent double-use attacks.
+	pub async fn find_from_login_token(&self, token: &str) -> conduwuit::Result<OwnedUserId> {
+		let Ok(value) = self.db.logintoken_expiresatuserid.get(token).await else {
+			return Err!(Request(Forbidden("Login token is unrecognised")));
+		};
+		let (expires_at, user_id): (u64, OwnedUserId) = value.deserialized()?;
+
+		if expires_at < utils::millis_since_unix_epoch() {
+			trace!(%user_id, ?token, "Removing expired login token");
+
+			self.db.logintoken_expiresatuserid.remove(token);
+
+			return Err!(Request(Forbidden("Login token is expired")));
+		}
+
+		self.db.logintoken_expiresatuserid.remove(token);
+
+		Ok(user_id)
+	}
+}

src/service/users/device.rs

@@ -0,0 +1,321 @@
+use std::net::IpAddr;
+
+use conduwuit::{
+	Err, utils,
+	utils::{ReadyExt, stream::TryIgnore},
+};
+use database::{Deserialized, Ignore, Interfix, Json};
+use futures::{Stream, StreamExt};
+use ruma::{
+	DeviceId, MilliSecondsSinceUnixEpoch, OwnedDeviceId, OwnedUserId, UserId,
+	api::client::device::Device, events::AnyToDeviceEvent, serde::Raw, uint,
+};
+use serde_json::json;
+
+use crate::users::increment;
+
+impl super::Service {
+	/// Adds a new device to a user. The user must exist, otherwise InvalidParam
+	/// is returned.
+	pub async fn create_device(
+		&self,
+		user_id: &UserId,
+		device_id: &DeviceId,
+		token: &str,
+		initial_device_display_name: Option<String>,
+		client_ip: Option<String>,
+	) -> conduwuit::Result<()> {
+		if !self.exists(user_id).await {
+			return Err!(Request(InvalidParam(error!(
+				"Called create_device for non-existent user {user_id}"
+			))));
+		}
+
+		let key = (user_id, device_id);
+		let mut device = Device::new(device_id.into());
+		device.display_name = initial_device_display_name;
+		device.last_seen_ip = client_ip;
+		device.last_seen_ts = Some(MilliSecondsSinceUnixEpoch::now());
+
+		increment(&self.db.userid_devicelistversion, user_id.as_bytes());
+		self.db.userdeviceid_metadata.put(key, Json(device));
+		self.set_token(user_id, device_id, token).await
+	}
+
+	/// Removes a device from a user.
+	pub async fn remove_device(&self, user_id: &UserId, device_id: &DeviceId) {
+		// Remove dehydrated device if this is the dehydrated device
+		let _: conduwuit::Result<_> = self
+			.remove_dehydrated_device(user_id, Some(device_id))
+			.await;
+
+		let userdeviceid = (user_id, device_id);
+
+		// Remove tokens
+		if let Ok(old_token) = self.db.userdeviceid_token.qry(&userdeviceid).await {
+			self.db.userdeviceid_token.del(userdeviceid);
+			self.db.token_userdeviceid.remove(&old_token);
+		}
+
+		// Remove todevice events
+		let prefix = (user_id, device_id, Interfix);
+		self.db
+			.todeviceid_events
+			.keys_prefix_raw(&prefix)
+			.ignore_err()
+			.ready_for_each(|key| self.db.todeviceid_events.remove(key))
+			.await;
+
+		// TODO: Remove onetimekeys
+
+		increment(&self.db.userid_devicelistversion, user_id.as_bytes());
+
+		self.db.userdeviceid_metadata.del(userdeviceid);
+		self.mark_device_key_update(user_id).await;
+	}
+
+	/// Returns an iterator over all device ids of this user.
+	pub fn all_device_ids<'a>(
+		&'a self,
+		user_id: &'a UserId,
+	) -> impl Stream<Item = OwnedDeviceId> + Send + 'a {
+		let prefix = (user_id, Interfix);
+		self.db
+			.userdeviceid_metadata
+			.keys_prefix(&prefix)
+			.ignore_err()
+			.map(|(_, device_id): (Ignore, OwnedDeviceId)| device_id)
+	}
+
+	/// Gets the access token associated with a device.
+	pub async fn get_token(
+		&self,
+		user_id: &UserId,
+		device_id: &DeviceId,
+	) -> conduwuit::Result<String> {
+		let key = (user_id, device_id);
+		self.db.userdeviceid_token.qry(&key).await.deserialized()
+	}
+
+	/// Generate a unique access token that doesn't collide with existing tokens
+	pub async fn generate_unique_token(&self) -> String {
+		loop {
+			let token = utils::random_string(32);
+
+			// Check for collision with existing appservice and user tokens
+			let (appservice, usr) = tokio::join!(
+				self.services.appservice.find_from_token(&token),
+				self.db.token_userdeviceid.get(&token)
+			);
+			if appservice.is_ok() || usr.is_ok() {
+				continue;
+			}
+
+			return token;
+		}
+	}
+
+	/// Replaces the access token of one device.
+	pub async fn set_token(
+		&self,
+		user_id: &UserId,
+		device_id: &DeviceId,
+		token: &str,
+	) -> conduwuit::Result<()> {
+		let key = (user_id, device_id);
+		if self.db.userdeviceid_metadata.qry(&key).await.is_err() {
+			return Err!(Database(error!(
+				%user_id,
+				%device_id,
+				"User does not exist or device has no metadata."
+			)));
+		}
+
+		// Check for token collision with appservices
+		if self
+			.services
+			.appservice
+			.find_from_token(token)
+			.await
+			.is_ok()
+		{
+			return Err!(Request(InvalidParam(
+				"Token conflicts with an existing appservice token"
+			)));
+		}
+
+		// Remove old token
+		if let Ok(old_token) = self.db.userdeviceid_token.qry(&key).await {
+			self.db.token_userdeviceid.remove(&old_token);
+			// It will be removed from userdeviceid_token by the insert later
+		}
+
+		// Assign token to user device combination
+		self.db.userdeviceid_token.put_raw(key, token);
+		self.db.token_userdeviceid.raw_put(token, key);
+
+		Ok(())
+	}
+
+	/// Pushes a new to-device event into a device's inbox.
+	pub async fn add_to_device_event(
+		&self,
+		sender: &UserId,
+		target_user_id: &UserId,
+		target_device_id: &DeviceId,
+		event_type: &str,
+		content: serde_json::Value,
+	) {
+		let count = self.services.globals.next_count().unwrap();
+
+		let key = (target_user_id, target_device_id, count);
+		self.db.todeviceid_events.put(
+			key,
+			Json(json!({
+				"type": event_type,
+				"sender": sender,
+				"content": content,
+			})),
+		);
+	}
+
+	/// Gets all to-device events between the two counts.
+	pub fn get_to_device_events<'a>(
+		&'a self,
+		user_id: &'a UserId,
+		device_id: &'a DeviceId,
+		since: Option<u64>,
+		to: Option<u64>,
+	) -> impl Stream<Item = (u64, Raw<AnyToDeviceEvent>)> + Send + 'a {
+		type Key = (OwnedUserId, OwnedDeviceId, u64);
+
+		let from = (user_id, device_id, since.map_or(0, |since| since.saturating_add(1)));
+
+		self.db
+			.todeviceid_events
+			.stream_from(&from)
+			.ignore_err()
+			.ready_take_while(move |((user_id_, device_id_, count), _): &(Key, _)| {
+				user_id == *user_id_
+					&& device_id == *device_id_
+					&& to.is_none_or(|to| *count <= to)
+			})
+			.map(|((_, _, count), event)| (count, event))
+	}
+
+	/// Removes to-device events from the target device's inbox, until the given
+	/// count.
+	pub async fn remove_to_device_events<Until>(
+		&self,
+		user_id: &UserId,
+		device_id: &DeviceId,
+		until: Until,
+	) where
+		Until: Into<Option<u64>> + Send,
+	{
+		type Key = (OwnedUserId, OwnedDeviceId, u64);
+
+		let until = until.into().unwrap_or(u64::MAX);
+		let from = (user_id, device_id, until);
+		self.db
+			.todeviceid_events
+			.rev_keys_from(&from)
+			.ignore_err()
+			.ready_take_while(move |(user_id_, device_id_, _): &Key| {
+				user_id == *user_id_ && device_id == *device_id_
+			})
+			.ready_for_each(|key: Key| {
+				self.db.todeviceid_events.del(key);
+			})
+			.await;
+	}
+
+	/// Updates device metadata and increments the device list version.
+	pub async fn update_device_metadata(
+		&self,
+		user_id: &UserId,
+		device_id: &DeviceId,
+		device: &Device,
+	) -> conduwuit::Result<()> {
+		increment(&self.db.userid_devicelistversion, user_id.as_bytes());
+		self.update_device_metadata_no_increment(user_id, device_id, device)
+	}
+
+	/// Updates device metadata without incrementing the device list version.
+	/// This is namely used for updating the last_seen_ip and last_seen_ts
+	/// values, as those do not need a device list version bump due to them not
+	/// being relevant to other consumers.
+	fn update_device_metadata_no_increment(
+		&self,
+		user_id: &UserId,
+		device_id: &DeviceId,
+		device: &Device,
+	) -> conduwuit::Result<()> {
+		let key = (user_id, device_id);
+		self.db.userdeviceid_metadata.put(key, Json(device));
+
+		Ok(())
+	}
+
+	/// Updates the last seen timestamp for a device. Silently does nothing if
+	/// the last update was less than 10 seconds ago, or the device does not
+	/// exist.
+	pub async fn update_device_last_seen(
+		&self,
+		user_id: &UserId,
+		device_id: Option<&DeviceId>,
+		ip: IpAddr,
+	) {
+		let now = MilliSecondsSinceUnixEpoch::now();
+		if let Some(device_id) = device_id {
+			if let Ok(mut device) = self.get_device_metadata(user_id, device_id).await {
+				device.last_seen_ip = Some(ip.to_string());
+				// If the last update was less than 10 seconds ago, don't update the timestamp
+				if let Some(prev) = device.last_seen_ts {
+					if now.get().saturating_sub(prev.get()) < uint!(10_000) {
+						return;
+					}
+				}
+				device.last_seen_ts = Some(now);
+
+				self.update_device_metadata_no_increment(user_id, device_id, &device)
+					.ok();
+			}
+		}
+	}
+
+	/// Get device metadata.
+	pub async fn get_device_metadata(
+		&self,
+		user_id: &UserId,
+		device_id: &DeviceId,
+	) -> conduwuit::Result<Device> {
+		self.db
+			.userdeviceid_metadata
+			.qry(&(user_id, device_id))
+			.await
+			.deserialized()
+	}
+
+	/// Gets the most recent device list version for a user.
+	pub async fn get_devicelist_version(&self, user_id: &UserId) -> conduwuit::Result<u64> {
+		self.db
+			.userid_devicelistversion
+			.get(user_id)
+			.await
+			.deserialized()
+	}
+
+	/// Gets metadata for all devices belonging to the target user.
+	pub fn all_devices_metadata<'a>(
+		&'a self,
+		user_id: &'a UserId,
+	) -> impl Stream<Item = Device> + Send + 'a {
+		let key = (user_id, Interfix);
+		self.db
+			.userdeviceid_metadata
+			.stream_prefix(&key)
+			.ignore_err()
+			.map(|(_, val): (Ignore, Device)| val)
+	}
+}

src/service/users/filters.rs

@@ -0,0 +1,26 @@
+use conduwuit::utils;
+use database::{Deserialized, Json};
+use ruma::{UserId, api::client::filter::FilterDefinition};
+
+impl super::Service {
+	/// Creates a new sync filter. Returns the filter id.
+	pub fn create_filter(&self, user_id: &UserId, filter: &FilterDefinition) -> String {
+		let filter_id = utils::random_string(4);
+
+		// TODO: filters should be de-duplicated and also not per-user
+		let key = (user_id, &filter_id);
+		self.db.userfilterid_filter.put(key, Json(filter));
+
+		filter_id
+	}
+
+	/// Fetches a filter from a filter ID belonging to a user.
+	pub async fn get_filter(
+		&self,
+		user_id: &UserId,
+		filter_id: &str,
+	) -> conduwuit::Result<FilterDefinition> {
+		let key = (user_id, filter_id);
+		self.db.userfilterid_filter.qry(&key).await.deserialized()
+	}
+}

src/service/users/keys.rs

@@ -0,0 +1,532 @@
+use std::collections::BTreeMap;
+
+use conduwuit::{
+	Err, Error, Result, err,
+	utils::{ReadyExt, stream::TryIgnore, string::Unquoted},
+};
+use database::{Deserialized, Ignore, Json};
+use futures::{Stream, StreamExt, TryFutureExt};
+use ruma::{
+	DeviceId, OneTimeKeyAlgorithm, OneTimeKeyId, OneTimeKeyName, OwnedKeyId, OwnedOneTimeKeyId,
+	OwnedUserId, RoomId, UInt, UserId,
+	api::error::ErrorKind,
+	encryption::{CrossSigningKey, DeviceKeys, OneTimeKey},
+	serde::Raw,
+};
+
+use crate::users::{clean_signatures, parse_master_key, parse_user_signing_key};
+
+impl super::Service {
+	/// Adds a single one-time key to a device.
+	pub async fn add_one_time_key(
+		&self,
+		user_id: &UserId,
+		device_id: &DeviceId,
+		one_time_key_key: &OneTimeKeyId,
+		one_time_key_value: &Raw<OneTimeKey>,
+	) -> Result {
+		// All devices have metadata
+		// Only existing devices should be able to call this, but we shouldn't assert
+		// either...
+		let key = (user_id, device_id);
+		if self.db.userdeviceid_metadata.qry(&key).await.is_err() {
+			return Err!(Database(error!(
+				%user_id,
+				%device_id,
+				"User does not exist or device has no metadata."
+			)));
+		}
+
+		let mut key = user_id.as_bytes().to_vec();
+		key.push(0xFF);
+		key.extend_from_slice(device_id.as_bytes());
+		key.push(0xFF);
+		// TODO: Use DeviceKeyId::to_string when it's available (and update everything,
+		// because there are no wrapping quotation marks anymore)
+		key.extend_from_slice(
+			serde_json::to_string(one_time_key_key)
+				.expect("DeviceKeyId::to_string always works")
+				.as_bytes(),
+		);
+
+		self.db
+			.onetimekeyid_onetimekeys
+			.raw_put(key, Json(one_time_key_value));
+
+		let count = self.services.globals.next_count().unwrap();
+		self.db.userid_lastonetimekeyupdate.raw_put(user_id, count);
+
+		Ok(())
+	}
+
+	/// Save a fallback key for the given user, device, and algorithm
+	/// This key will replace an existing fallback key.
+	pub async fn add_fallback_key(
+		&self,
+		user_id: &UserId,
+		device_id: &DeviceId,
+		fallback_key_id: &OneTimeKeyId,
+		fallback_key: &Raw<OneTimeKey>,
+		used: bool,
+	) -> Result {
+		// All devices have metadata
+		// Only existing devices should be able to call this, but we shouldn't assert
+		// either...
+		let key = (user_id, device_id);
+		if self.db.userdeviceid_metadata.qry(&key).await.is_err() {
+			return Err!(Database(error!(
+				%user_id,
+				%device_id,
+				"User does not exist or device has no metadata."
+			)));
+		}
+
+		// There is one fallback key slot per user, per device, per algorithm
+		// Therefore we use this as the DB key for this column
+		let db_key = (user_id, device_id, fallback_key_id.algorithm());
+
+		self.db
+			.fallbackkeyid_fallbackkey
+			.put(db_key, (used, fallback_key_id.as_str(), Json(fallback_key)));
+
+		Ok(())
+	}
+
+	/// Returns the timestamp at when the last OTK update for the user was, or 0
+	/// if the keys have never been updated.
+	pub async fn last_one_time_keys_update(&self, user_id: &UserId) -> u64 {
+		self.db
+			.userid_lastonetimekeyupdate
+			.get(user_id)
+			.await
+			.deserialized()
+			.unwrap_or(0)
+	}
+
+	/// Consumes a one-time key belonging to the device of a given algorithm. If
+	/// no one-time keys are available, a fallback key is returned instead (if
+	/// available). If neither an OTK nor fallback key are available, NotFound
+	/// is returned.
+	pub async fn take_one_time_key(
+		&self,
+		user_id: &UserId,
+		device_id: &DeviceId,
+		key_algorithm: &OneTimeKeyAlgorithm,
+	) -> Result<(OwnedKeyId<OneTimeKeyAlgorithm, OneTimeKeyName>, Raw<OneTimeKey>)> {
+		let count = self.services.globals.next_count()?.to_be_bytes();
+		self.db.userid_lastonetimekeyupdate.insert(user_id, count);
+
+		let mut prefix = user_id.as_bytes().to_vec();
+		prefix.push(0xFF);
+		prefix.extend_from_slice(device_id.as_bytes());
+		prefix.push(0xFF);
+		prefix.push(b'"'); // Annoying quotation mark
+		prefix.extend_from_slice(key_algorithm.as_ref().as_bytes());
+		prefix.push(b':');
+
+		let one_time_key = self
+			.db
+			.onetimekeyid_onetimekeys
+			.raw_stream_prefix(&prefix)
+			.ignore_err()
+			.next()
+			.await
+			.map(|(key, val)| {
+				self.db.onetimekeyid_onetimekeys.remove(key);
+
+				let key = key
+					.rsplit(|&b| b == 0xFF)
+					.next()
+					.ok_or_else(|| err!(Database("OneTimeKeyId in db is invalid.")))
+					.unwrap();
+
+				let key = serde_json::from_slice(key)
+					.map_err(|e| err!(Database("OneTimeKeyId in db is invalid. {e}")))
+					.unwrap();
+
+				let val = serde_json::from_slice(val)
+					.map_err(|e| err!(Database("OneTimeKeys in db are invalid. {e}")))
+					.unwrap();
+
+				(key, val)
+			});
+
+		if let Some(result) = one_time_key {
+			return Ok(result);
+		}
+
+		// No one-time key has been found. Look for a fallback key.
+
+		let db_key = (user_id, device_id, key_algorithm);
+
+		let fallback_key = self
+			.db
+			.fallbackkeyid_fallbackkey
+			.qry(&db_key)
+			.await
+			.ok()
+			.and_then(|handle| {
+				handle
+					.deserialized::<(bool, OwnedOneTimeKeyId, Raw<OneTimeKey>)>()
+					.ok()
+			});
+
+		if let Some((used, fallback_key_id, fallback_key_value)) = fallback_key {
+			if !used {
+				// write the key to the database again to mark it as used
+				self.add_fallback_key(
+					user_id,
+					device_id,
+					&fallback_key_id,
+					&fallback_key_value,
+					true,
+				)
+				.await?;
+			}
+			return Ok((fallback_key_id, fallback_key_value));
+		}
+
+		Err(err!(Request(NotFound("No one-time key or fallback key found"))))
+	}
+
+	/// Returns the number of one-time keys the given device has. Does not count
+	/// fallback keys.
+	pub async fn count_one_time_keys(
+		&self,
+		user_id: &UserId,
+		device_id: &DeviceId,
+	) -> BTreeMap<OneTimeKeyAlgorithm, UInt> {
+		type KeyVal<'a> = ((Ignore, Ignore, &'a Unquoted), Ignore);
+
+		let mut algorithm_counts = BTreeMap::<OneTimeKeyAlgorithm, _>::new();
+		let query = (user_id, device_id);
+		self.db
+			.onetimekeyid_onetimekeys
+			.stream_prefix(&query)
+			.ignore_err()
+			.ready_for_each(|((_, _, device_key_id), _): KeyVal<'_>| {
+				let one_time_key_id: &OneTimeKeyId = device_key_id
+					.as_str()
+					.try_into()
+					.expect("Invalid DeviceKeyID in database");
+
+				let count: &mut UInt = algorithm_counts
+					.entry(one_time_key_id.algorithm())
+					.or_default();
+
+				*count = count.saturating_add(1_u32.into());
+			})
+			.await;
+
+		algorithm_counts
+	}
+
+	/// Returns a list of *unused* fallback key types.
+	pub async fn list_unused_fallback_key_types(
+		&self,
+		user_id: &UserId,
+		device_id: &DeviceId,
+	) -> Vec<OneTimeKeyAlgorithm> {
+		type KeyVal = ((String, String, OneTimeKeyAlgorithm), (bool, String, Ignore));
+
+		let mut query = user_id.as_bytes().to_vec();
+		query.push(0xFF);
+		query.extend_from_slice(device_id.as_bytes());
+		query.push(0xFF);
+
+		let mut unused_algorithms = Vec::new();
+
+		self.db
+			.fallbackkeyid_fallbackkey
+			.stream_prefix(&query)
+			.ignore_err()
+			.ready_for_each(|((_, _, fallback_key_algorithm), (used, ..)): KeyVal| {
+				if !used {
+					unused_algorithms.push(fallback_key_algorithm);
+				}
+			})
+			.await;
+
+		unused_algorithms
+	}
+
+	/// Adds device identity keys to a device, overwriting existing ones if they
+	/// exist. Dispatches a device key update.
+	pub async fn add_device_keys(
+		&self,
+		user_id: &UserId,
+		device_id: &DeviceId,
+		device_keys: &Raw<DeviceKeys>,
+	) {
+		let key = (user_id, device_id);
+
+		self.db.keyid_key.put(key, Json(device_keys));
+		self.mark_device_key_update(user_id).await;
+	}
+
+	/// Adds cross-signing keys for the given user. If notify is true, a device
+	/// key update is dispatched.
+	pub async fn add_cross_signing_keys(
+		&self,
+		user_id: &UserId,
+		master_key: &Option<Raw<CrossSigningKey>>,
+		self_signing_key: &Option<Raw<CrossSigningKey>>,
+		user_signing_key: &Option<Raw<CrossSigningKey>>,
+		notify: bool,
+	) -> Result<()> {
+		let mut prefix = user_id.as_bytes().to_vec();
+		prefix.push(0xFF);
+
+		if let Some(master_key) = master_key {
+			let (master_key_key, _) = parse_master_key(user_id, master_key)?;
+
+			self.db
+				.keyid_key
+				.insert(&master_key_key, master_key.json().get().as_bytes());
+
+			self.db
+				.userid_masterkeyid
+				.insert(user_id.as_bytes(), &master_key_key);
+		}
+
+		// Self-signing key
+		if let Some(self_signing_key) = self_signing_key {
+			let mut self_signing_key_ids = self_signing_key
+				.deserialize()
+				.map_err(|e| err!(Request(InvalidParam("Invalid self signing key: {e:?}"))))?
+				.keys
+				.into_values();
+
+			let self_signing_key_id = self_signing_key_ids.next().ok_or(Error::BadRequest(
+				ErrorKind::InvalidParam,
+				"Self signing key contained no key.",
+			))?;
+
+			if self_signing_key_ids.next().is_some() {
+				return Err(Error::BadRequest(
+					ErrorKind::InvalidParam,
+					"Self signing key contained more than one key.",
+				));
+			}
+
+			let mut self_signing_key_key = prefix.clone();
+			self_signing_key_key.extend_from_slice(self_signing_key_id.as_bytes());
+
+			self.db
+				.keyid_key
+				.insert(&self_signing_key_key, self_signing_key.json().get().as_bytes());
+
+			self.db
+				.userid_selfsigningkeyid
+				.insert(user_id.as_bytes(), &self_signing_key_key);
+		}
+
+		// User-signing key
+		if let Some(user_signing_key) = user_signing_key {
+			let user_signing_key_id = parse_user_signing_key(user_signing_key)?;
+
+			let user_signing_key_key = (user_id, &user_signing_key_id);
+			self.db
+				.keyid_key
+				.put_raw(user_signing_key_key, user_signing_key.json().get().as_bytes());
+
+			self.db
+				.userid_usersigningkeyid
+				.raw_put(user_id, user_signing_key_key);
+		}
+
+		if notify {
+			self.mark_device_key_update(user_id).await;
+		}
+
+		Ok(())
+	}
+
+	/// Appends a new signature to the given key ID. Dispatches a new device key
+	/// update.
+	pub async fn sign_key(
+		&self,
+		target_id: &UserId,
+		key_id: &str,
+		signature: (String, String),
+		sender_id: &UserId,
+	) -> Result {
+		let key = (target_id, key_id);
+
+		let mut cross_signing_key: serde_json::Value = self
+			.db
+			.keyid_key
+			.qry(&key)
+			.await
+			.map_err(|_| err!(Request(InvalidParam("Tried to sign nonexistent key"))))?
+			.deserialized()
+			.map_err(|e| err!(Database(debug_warn!("key in keyid_key is invalid: {e:?}"))))?;
+
+		let signatures = cross_signing_key
+			.get_mut("signatures")
+			.ok_or_else(|| {
+				err!(Database(debug_warn!("key in keyid_key has no signatures field")))
+			})?
+			.as_object_mut()
+			.ok_or_else(|| {
+				err!(Database(debug_warn!("key in keyid_key has invalid signatures field.")))
+			})?
+			.entry(sender_id.to_string())
+			.or_insert_with(|| serde_json::Map::new().into());
+
+		signatures
+			.as_object_mut()
+			.ok_or_else(|| {
+				err!(Database(debug_warn!("signatures in keyid_key for a user is invalid.")))
+			})?
+			.insert(signature.0, signature.1.into());
+
+		let key = (target_id, key_id);
+		self.db.keyid_key.put(key, Json(cross_signing_key));
+
+		self.mark_device_key_update(target_id).await;
+
+		Ok(())
+	}
+
+	/// Returns a stream of changed keys between the two counts.
+	#[inline]
+	pub fn keys_changed<'a>(
+		&'a self,
+		user_id: &'a UserId,
+		from: Option<u64>,
+		to: Option<u64>,
+	) -> impl Stream<Item = OwnedUserId> + Send + 'a {
+		self.keys_changed_user_or_room(user_id.as_str(), from, to)
+			.map(|(user_id, ..)| user_id)
+	}
+
+	/// Returns a stream of changed keys between the two counts in the given
+	/// room specifically.
+	#[inline]
+	pub fn room_keys_changed<'a>(
+		&'a self,
+		room_id: &'a RoomId,
+		from: Option<u64>,
+		to: Option<u64>,
+	) -> impl Stream<Item = (OwnedUserId, u64)> + Send + 'a {
+		self.keys_changed_user_or_room(room_id.as_str(), from, to)
+	}
+
+	fn keys_changed_user_or_room<'a>(
+		&'a self,
+		user_or_room_id: &'a str,
+		from: Option<u64>,
+		to: Option<u64>,
+	) -> impl Stream<Item = (OwnedUserId, u64)> + Send + 'a {
+		type KeyVal<'a> = ((&'a str, u64), OwnedUserId);
+
+		let from = from.unwrap_or(0);
+		let to = to.unwrap_or(u64::MAX);
+		let start = (user_or_room_id, from.saturating_add(1));
+		self.db
+			.keychangeid_userid
+			.stream_from(&start)
+			.ignore_err()
+			.ready_take_while(move |((prefix, count), _): &KeyVal<'_>| {
+				*prefix == user_or_room_id && *count <= to
+			})
+			.map(|((_, count), user_id): KeyVal<'_>| (user_id, count))
+	}
+
+	/// Marks that a user's device keys have been updated, so that other users
+	/// can be notified of the change.
+	pub async fn mark_device_key_update(&self, user_id: &UserId) {
+		let count = self.services.globals.next_count().unwrap();
+
+		self.services
+			.state_cache
+			.rooms_joined(user_id)
+			// Don't send key updates to unencrypted rooms
+			.filter_map(async |room_id| {
+				if self.services.state_accessor.is_encrypted_room(&room_id).await {
+					Some(room_id)
+				} else {
+					None
+				}
+			})
+			.ready_for_each(|room_id| {
+				let key = (room_id, count);
+				self.db.keychangeid_userid.put_raw(key, user_id);
+			})
+			.await;
+
+		let key = (user_id, count);
+		self.db.keychangeid_userid.put_raw(key, user_id);
+	}
+
+	/// Returns the device identity keys for a given device.
+	pub async fn get_device_keys<'a>(
+		&'a self,
+		user_id: &'a UserId,
+		device_id: &DeviceId,
+	) -> Result<Raw<DeviceKeys>> {
+		let key_id = (user_id, device_id);
+		self.db.keyid_key.qry(&key_id).await.deserialized()
+	}
+
+	/// Gets a specific cross-signing key.
+	pub async fn get_key<F>(
+		&self,
+		key_id: &[u8],
+		sender_user: Option<&UserId>,
+		user_id: &UserId,
+		allowed_signatures: &F,
+	) -> Result<Raw<CrossSigningKey>>
+	where
+		F: Fn(&UserId) -> bool + Send + Sync,
+	{
+		let key: serde_json::Value = self.db.keyid_key.get(key_id).await.deserialized()?;
+
+		let cleaned = clean_signatures(key, sender_user, user_id, allowed_signatures)?;
+		let raw_value = serde_json::value::to_raw_value(&cleaned)?;
+		Ok(Raw::from_json(raw_value))
+	}
+
+	/// Gets a user's master cross-signing key.
+	pub async fn get_master_key<F>(
+		&self,
+		sender_user: Option<&UserId>,
+		user_id: &UserId,
+		allowed_signatures: &F,
+	) -> Result<Raw<CrossSigningKey>>
+	where
+		F: Fn(&UserId) -> bool + Send + Sync,
+	{
+		let key_id = self.db.userid_masterkeyid.get(user_id).await?;
+
+		self.get_key(&key_id, sender_user, user_id, allowed_signatures)
+			.await
+	}
+
+	/// Gets a user's self-signing cross-signing key.
+	pub async fn get_self_signing_key<F>(
+		&self,
+		sender_user: Option<&UserId>,
+		user_id: &UserId,
+		allowed_signatures: &F,
+	) -> Result<Raw<CrossSigningKey>>
+	where
+		F: Fn(&UserId) -> bool + Send + Sync,
+	{
+		let key_id = self.db.userid_selfsigningkeyid.get(user_id).await?;
+
+		self.get_key(&key_id, sender_user, user_id, allowed_signatures)
+			.await
+	}
+
+	/// Gets a user's user-signing cross-signing key.
+	pub async fn get_user_signing_key(&self, user_id: &UserId) -> Result<Raw<CrossSigningKey>> {
+		self.db
+			.userid_usersigningkeyid
+			.get(user_id)
+			.and_then(|key_id| self.db.keyid_key.get(&*key_id))
+			.await
+			.deserialized()
+	}
+}

src/service/users/profile.rs

@@ -0,0 +1,94 @@
+use conduwuit::utils::{ReadyExt, stream::TryIgnore};
+use database::{Deserialized, Ignore, Interfix, Json};
+use futures::{Stream, StreamExt};
+use ruma::{OwnedMxcUri, UserId};
+
+impl super::Service {
+	/// Returns the displayname of a user on this homeserver.
+	pub async fn displayname(&self, user_id: &UserId) -> conduwuit::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>) {
+		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) -> conduwuit::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>) {
+		match avatar_url {
+			| Some(avatar_url) => {
+				self.db.userid_avatarurl.insert(user_id, &avatar_url);
+			},
+			| _ => {
+				self.db.userid_avatarurl.remove(user_id);
+			},
+		}
+	}
+
+	/// Gets a specific user profile key
+	pub async fn profile_key(
+		&self,
+		user_id: &UserId,
+		profile_key: &str,
+	) -> conduwuit::Result<serde_json::Value> {
+		let key = (user_id, profile_key);
+		self.db
+			.useridprofilekey_value
+			.qry(&key)
+			.await
+			.and_then(|handle| serde_json::from_slice(&handle).map_err(Into::into))
+	}
+
+	/// Gets all the user's profile keys and values in an iterator
+	pub fn all_profile_keys<'a>(
+		&'a self,
+		user_id: &'a UserId,
+	) -> impl Stream<Item = (String, serde_json::Value)> + 'a + Send {
+		type KeyVal<'a> = ((Ignore, String), &'a [u8]);
+
+		let prefix = (user_id, Interfix);
+		self.db
+			.useridprofilekey_value
+			.stream_prefix(&prefix)
+			.ignore_err()
+			.map(|((_, key), value): KeyVal<'_>| Ok((key, serde_json::from_slice(value)?)))
+			.ignore_err()
+	}
+
+	/// Sets a new profile key value, removes the key if value is None
+	pub fn set_profile_key(
+		&self,
+		user_id: &UserId,
+		profile_key: &str,
+		profile_key_value: Option<serde_json::Value>,
+	) {
+		let key = (user_id, profile_key);
+
+		if let Some(value) = profile_key_value {
+			self.db.useridprofilekey_value.put(key, Json(value));
+		} else {
+			self.db.useridprofilekey_value.del(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.all_profile_keys(user_id)
+			.ready_for_each(|(key, _)| self.set_profile_key(user_id, &key, None))
+			.await;
+	}
+}