more funny settings (part 3 of 12)

scale rocksdb background jobs and subcompactions

change rocksdb default error level to info from error

delete unused num_threads function

fix warns from cargo

.github/FUNDING.yml

@@ -0,0 +1,5 @@
+github: [JadedBlueEyes]
+# Doesn't support an array, so we can only list nex
+ko_fi: nexy7574
+custom:
+  - https://ko-fi.com/JadedBlueEyes

Cargo.lock

@@ -3798,7 +3798,7 @@ dependencies = [
 [[package]]
 name = "ruma"
 version = "0.10.1"
-source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=d6870a7fb7f6cccff63f7fd0ff6c581bad80e983#d6870a7fb7f6cccff63f7fd0ff6c581bad80e983"
+source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=9b65f83981f6f53d185ce77da37aaef9dfd764a9#9b65f83981f6f53d185ce77da37aaef9dfd764a9"
 dependencies = [
  "assign",
  "js_int",
@@ -3818,7 +3818,7 @@ dependencies = [
 [[package]]
 name = "ruma-appservice-api"
 version = "0.10.0"
-source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=d6870a7fb7f6cccff63f7fd0ff6c581bad80e983#d6870a7fb7f6cccff63f7fd0ff6c581bad80e983"
+source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=9b65f83981f6f53d185ce77da37aaef9dfd764a9#9b65f83981f6f53d185ce77da37aaef9dfd764a9"
 dependencies = [
  "js_int",
  "ruma-common",
@@ -3830,7 +3830,7 @@ dependencies = [
 [[package]]
 name = "ruma-client-api"
 version = "0.18.0"
-source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=d6870a7fb7f6cccff63f7fd0ff6c581bad80e983#d6870a7fb7f6cccff63f7fd0ff6c581bad80e983"
+source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=9b65f83981f6f53d185ce77da37aaef9dfd764a9#9b65f83981f6f53d185ce77da37aaef9dfd764a9"
 dependencies = [
  "as_variant",
  "assign",
@@ -3853,7 +3853,7 @@ dependencies = [
 [[package]]
 name = "ruma-common"
 version = "0.13.0"
-source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=d6870a7fb7f6cccff63f7fd0ff6c581bad80e983#d6870a7fb7f6cccff63f7fd0ff6c581bad80e983"
+source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=9b65f83981f6f53d185ce77da37aaef9dfd764a9#9b65f83981f6f53d185ce77da37aaef9dfd764a9"
 dependencies = [
  "as_variant",
  "base64 0.22.1",
@@ -3885,7 +3885,7 @@ dependencies = [
 [[package]]
 name = "ruma-events"
 version = "0.28.1"
-source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=d6870a7fb7f6cccff63f7fd0ff6c581bad80e983#d6870a7fb7f6cccff63f7fd0ff6c581bad80e983"
+source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=9b65f83981f6f53d185ce77da37aaef9dfd764a9#9b65f83981f6f53d185ce77da37aaef9dfd764a9"
 dependencies = [
  "as_variant",
  "indexmap 2.9.0",
@@ -3910,7 +3910,7 @@ dependencies = [
 [[package]]
 name = "ruma-federation-api"
 version = "0.9.0"
-source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=d6870a7fb7f6cccff63f7fd0ff6c581bad80e983#d6870a7fb7f6cccff63f7fd0ff6c581bad80e983"
+source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=9b65f83981f6f53d185ce77da37aaef9dfd764a9#9b65f83981f6f53d185ce77da37aaef9dfd764a9"
 dependencies = [
  "bytes",
  "headers",
@@ -3932,7 +3932,7 @@ dependencies = [
 [[package]]
 name = "ruma-identifiers-validation"
 version = "0.9.5"
-source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=d6870a7fb7f6cccff63f7fd0ff6c581bad80e983#d6870a7fb7f6cccff63f7fd0ff6c581bad80e983"
+source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=9b65f83981f6f53d185ce77da37aaef9dfd764a9#9b65f83981f6f53d185ce77da37aaef9dfd764a9"
 dependencies = [
  "js_int",
  "thiserror 2.0.12",
@@ -3941,7 +3941,7 @@ dependencies = [
 [[package]]
 name = "ruma-identity-service-api"
 version = "0.9.0"
-source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=d6870a7fb7f6cccff63f7fd0ff6c581bad80e983#d6870a7fb7f6cccff63f7fd0ff6c581bad80e983"
+source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=9b65f83981f6f53d185ce77da37aaef9dfd764a9#9b65f83981f6f53d185ce77da37aaef9dfd764a9"
 dependencies = [
  "js_int",
  "ruma-common",
@@ -3951,7 +3951,7 @@ dependencies = [
 [[package]]
 name = "ruma-macros"
 version = "0.13.0"
-source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=d6870a7fb7f6cccff63f7fd0ff6c581bad80e983#d6870a7fb7f6cccff63f7fd0ff6c581bad80e983"
+source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=9b65f83981f6f53d185ce77da37aaef9dfd764a9#9b65f83981f6f53d185ce77da37aaef9dfd764a9"
 dependencies = [
  "cfg-if",
  "proc-macro-crate",
@@ -3966,7 +3966,7 @@ dependencies = [
 [[package]]
 name = "ruma-push-gateway-api"
 version = "0.9.0"
-source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=d6870a7fb7f6cccff63f7fd0ff6c581bad80e983#d6870a7fb7f6cccff63f7fd0ff6c581bad80e983"
+source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=9b65f83981f6f53d185ce77da37aaef9dfd764a9#9b65f83981f6f53d185ce77da37aaef9dfd764a9"
 dependencies = [
  "js_int",
  "ruma-common",
@@ -3978,7 +3978,7 @@ dependencies = [
 [[package]]
 name = "ruma-signatures"
 version = "0.15.0"
-source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=d6870a7fb7f6cccff63f7fd0ff6c581bad80e983#d6870a7fb7f6cccff63f7fd0ff6c581bad80e983"
+source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=9b65f83981f6f53d185ce77da37aaef9dfd764a9#9b65f83981f6f53d185ce77da37aaef9dfd764a9"
 dependencies = [
  "base64 0.22.1",
  "ed25519-dalek",

Cargo.toml

@@ -350,7 +350,7 @@ version = "0.1.2"
 [workspace.dependencies.ruma]
 git = "https://forgejo.ellis.link/continuwuation/ruwuma"
 #branch = "conduwuit-changes"
-rev = "d6870a7fb7f6cccff63f7fd0ff6c581bad80e983"
+rev = "9b65f83981f6f53d185ce77da37aaef9dfd764a9"
 features = [
     "compat",
     "rand",

src/admin/context.rs

@@ -7,13 +7,14 @@ use futures::{
 	io::{AsyncWriteExt, BufWriter},
 	lock::Mutex,
 };
-use ruma::EventId;
+use ruma::{EventId, UserId};
 
 pub(crate) struct Context<'a> {
 	pub(crate) services: &'a Services,
 	pub(crate) body: &'a [&'a str],
 	pub(crate) timer: SystemTime,
 	pub(crate) reply_id: Option<&'a EventId>,
+	pub(crate) sender: Option<&'a UserId>,
 	pub(crate) output: Mutex<BufWriter<Vec<u8>>>,
 }
 
@@ -36,4 +37,10 @@ impl Context<'_> {
 			output.write_all(s.as_bytes()).map_err(Into::into).await
 		})
 	}
+
+	/// Get the sender as a string, or service user ID if not available
+	pub(crate) fn sender_or_service_user(&self) -> &UserId {
+		self.sender
+			.unwrap_or_else(|| self.services.globals.server_user.as_ref())
+	}
 }

src/admin/processor.rs

@@ -63,6 +63,7 @@ async fn process_command(services: Arc<Services>, input: &CommandInput) -> Proce
 		body: &body,
 		timer: SystemTime::now(),
 		reply_id: input.reply_id.as_deref(),
+		sender: input.sender.as_deref(),
 		output: BufWriter::new(Vec::new()).into(),
 	};
 

src/admin/user/commands.rs

@@ -225,6 +225,47 @@ pub(super) async fn deactivate(&self, no_leave_rooms: bool, user_id: String) ->
 		.await
 }
 
+#[admin_command]
+pub(super) async fn suspend(&self, user_id: String) -> Result {
+	let user_id = parse_local_user_id(self.services, &user_id)?;
+
+	if user_id == self.services.globals.server_user {
+		return Err!("Not allowed to suspend the server service account.",);
+	}
+
+	if !self.services.users.exists(&user_id).await {
+		return Err!("User {user_id} does not exist.");
+	}
+	if self.services.users.is_admin(&user_id).await {
+		return Err!("Admin users cannot be suspended.");
+	}
+	// TODO: Record the actual user that sent the suspension where possible
+	self.services
+		.users
+		.suspend_account(&user_id, self.sender_or_service_user())
+		.await;
+
+	self.write_str(&format!("User {user_id} has been suspended."))
+		.await
+}
+
+#[admin_command]
+pub(super) async fn unsuspend(&self, user_id: String) -> Result {
+	let user_id = parse_local_user_id(self.services, &user_id)?;
+
+	if user_id == self.services.globals.server_user {
+		return Err!("Not allowed to unsuspend the server service account.",);
+	}
+
+	if !self.services.users.exists(&user_id).await {
+		return Err!("User {user_id} does not exist.");
+	}
+	self.services.users.unsuspend_account(&user_id).await;
+
+	self.write_str(&format!("User {user_id} has been unsuspended."))
+		.await
+}
+
 #[admin_command]
 pub(super) async fn reset_password(&self, username: String, password: Option<String>) -> Result {
 	let user_id = parse_local_user_id(self.services, &username)?;

src/admin/user/mod.rs

@@ -59,6 +59,28 @@ pub(super) enum UserCommand {
 		force: bool,
 	},
 
+	/// - Suspend a user
+	///
+	/// Suspended users are able to log in, sync, and read messages, but are not
+	/// able to send events nor redact them, cannot change their profile, and
+	/// are unable to join, invite to, or knock on rooms.
+	///
+	/// Suspended users can still leave rooms and deactivate their account.
+	/// Suspending them effectively makes them read-only.
+	Suspend {
+		/// Username of the user to suspend
+		user_id: String,
+	},
+
+	/// - Unsuspend a user
+	///
+	/// Reverses the effects of the `suspend` command, allowing the user to send
+	/// messages, change their profile, create room invites, etc.
+	Unsuspend {
+		/// Username of the user to unsuspend
+		user_id: String,
+	},
+
 	/// - List local users in the database
 	#[clap(alias = "list")]
 	ListUsers,

src/api/client/alias.rs

@@ -18,6 +18,9 @@ pub(crate) async fn create_alias_route(
 	body: Ruma<create_alias::v3::Request>,
 ) -> Result<create_alias::v3::Response> {
 	let sender_user = body.sender_user.as_ref().expect("user is authenticated");
+	if services.users.is_suspended(sender_user).await? {
+		return Err!(Request(UserSuspended("You cannot perform this action while suspended.")));
+	}
 
 	services
 		.rooms
@@ -63,6 +66,9 @@ pub(crate) async fn delete_alias_route(
 	body: Ruma<delete_alias::v3::Request>,
 ) -> Result<delete_alias::v3::Response> {
 	let sender_user = body.sender_user.as_ref().expect("user is authenticated");
+	if services.users.is_suspended(sender_user).await? {
+		return Err!(Request(UserSuspended("You cannot perform this action while suspended.")));
+	}
 
 	services
 		.rooms

src/api/client/directory.rs

@@ -128,6 +128,9 @@ pub(crate) async fn set_room_visibility_route(
 		// Return 404 if the room doesn't exist
 		return Err!(Request(NotFound("Room not found")));
 	}
+	if services.users.is_suspended(sender_user).await? {
+		return Err!(Request(UserSuspended("You cannot perform this action while suspended.")));
+	}
 
 	if services
 		.users

src/api/client/media.rs

@@ -52,6 +52,9 @@ pub(crate) async fn create_content_route(
 	body: Ruma<create_content::v3::Request>,
 ) -> Result<create_content::v3::Response> {
 	let user = body.sender_user.as_ref().expect("user is authenticated");
+	if services.users.is_suspended(user).await? {
+		return Err!(Request(UserSuspended("You cannot perform this action while suspended.")));
+	}
 
 	let filename = body.filename.as_deref();
 	let content_type = body.content_type.as_deref();

src/api/client/membership.rs

@@ -182,6 +182,9 @@ pub(crate) async fn join_room_by_id_route(
 	body: Ruma<join_room_by_id::v3::Request>,
 ) -> Result<join_room_by_id::v3::Response> {
 	let sender_user = body.sender_user();
+	if services.users.is_suspended(sender_user).await? {
+		return Err!(Request(UserSuspended("You cannot perform this action while suspended.")));
+	}
 
 	banned_room_check(
 		&services,
@@ -253,6 +256,9 @@ pub(crate) async fn join_room_by_id_or_alias_route(
 	let sender_user = body.sender_user.as_deref().expect("user is authenticated");
 	let appservice_info = &body.appservice_info;
 	let body = body.body;
+	if services.users.is_suspended(sender_user).await? {
+		return Err!(Request(UserSuspended("You cannot perform this action while suspended.")));
+	}
 
 	let (servers, room_id) = match OwnedRoomId::try_from(body.room_id_or_alias) {
 		| Ok(room_id) => {
@@ -374,6 +380,9 @@ pub(crate) async fn knock_room_route(
 ) -> Result<knock_room::v3::Response> {
 	let sender_user = body.sender_user();
 	let body = &body.body;
+	if services.users.is_suspended(sender_user).await? {
+		return Err!(Request(UserSuspended("You cannot perform this action while suspended.")));
+	}
 
 	let (servers, room_id) = match OwnedRoomId::try_from(body.room_id_or_alias.clone()) {
 		| Ok(room_id) => {
@@ -498,6 +507,9 @@ pub(crate) async fn invite_user_route(
 	body: Ruma<invite_user::v3::Request>,
 ) -> Result<invite_user::v3::Response> {
 	let sender_user = body.sender_user();
+	if services.users.is_suspended(sender_user).await? {
+		return Err!(Request(UserSuspended("You cannot perform this action while suspended.")));
+	}
 
 	if !services.users.is_admin(sender_user).await && services.config.block_non_admin_invites {
 		debug_error!(
@@ -572,6 +584,10 @@ pub(crate) async fn kick_user_route(
 	State(services): State<crate::State>,
 	body: Ruma<kick_user::v3::Request>,
 ) -> Result<kick_user::v3::Response> {
+	let sender_user = body.sender_user();
+	if services.users.is_suspended(sender_user).await? {
+		return Err!(Request(UserSuspended("You cannot perform this action while suspended.")));
+	}
 	let state_lock = services.rooms.state.mutex.lock(&body.room_id).await;
 
 	let Ok(event) = services
@@ -607,7 +623,7 @@ pub(crate) async fn kick_user_route(
 				third_party_invite: None,
 				..event
 			}),
-			body.sender_user(),
+			sender_user,
 			&body.room_id,
 			&state_lock,
 		)
@@ -631,6 +647,10 @@ pub(crate) async fn ban_user_route(
 		return Err!(Request(Forbidden("You cannot ban yourself.")));
 	}
 
+	if services.users.is_suspended(sender_user).await? {
+		return Err!(Request(UserSuspended("You cannot perform this action while suspended.")));
+	}
+
 	let state_lock = services.rooms.state.mutex.lock(&body.room_id).await;
 
 	let current_member_content = services
@@ -652,6 +672,7 @@ pub(crate) async fn ban_user_route(
 				is_direct: None,
 				join_authorized_via_users_server: None,
 				third_party_invite: None,
+				redact_events: body.redact_events,
 				..current_member_content
 			}),
 			sender_user,
@@ -672,6 +693,10 @@ pub(crate) async fn unban_user_route(
 	State(services): State<crate::State>,
 	body: Ruma<unban_user::v3::Request>,
 ) -> Result<unban_user::v3::Response> {
+	let sender_user = body.sender_user();
+	if services.users.is_suspended(sender_user).await? {
+		return Err!(Request(UserSuspended("You cannot perform this action while suspended.")));
+	}
 	let state_lock = services.rooms.state.mutex.lock(&body.room_id).await;
 
 	let current_member_content = services
@@ -700,7 +725,7 @@ pub(crate) async fn unban_user_route(
 				is_direct: None,
 				..current_member_content
 			}),
-			body.sender_user(),
+			sender_user,
 			&body.room_id,
 			&state_lock,
 		)
@@ -931,24 +956,32 @@ pub async fn join_room_by_id_helper(
 		return Ok(join_room_by_id::v3::Response { room_id: room_id.into() });
 	}
 
-	if let Ok(membership) = services
-		.rooms
-		.state_accessor
-		.get_member(room_id, sender_user)
-		.await
-	{
-		if membership.membership == MembershipState::Ban {
-			debug_warn!("{sender_user} is banned from {room_id} but attempted to join");
-			return Err!(Request(Forbidden("You are banned from the room.")));
-		}
-	}
-
 	let server_in_room = services
 		.rooms
 		.state_cache
 		.server_in_room(services.globals.server_name(), room_id)
 		.await;
 
+	// Only check our known membership if we're already in the room.
+	// See: https://forgejo.ellis.link/continuwuation/continuwuity/issues/855
+	let membership = if server_in_room {
+		services
+			.rooms
+			.state_accessor
+			.get_member(room_id, sender_user)
+			.await
+	} else {
+		debug!("Ignoring local state for join {room_id}, we aren't in the room yet.");
+		Ok(RoomMemberEventContent::new(MembershipState::Leave))
+	};
+	if let Ok(m) = membership {
+		if m.membership == MembershipState::Ban {
+			debug_warn!("{sender_user} is banned from {room_id} but attempted to join");
+			// TODO: return reason
+			return Err!(Request(Forbidden("You are banned from the room.")));
+		}
+	}
+
 	let local_join = server_in_room
 		|| servers.is_empty()
 		|| (servers.len() == 1 && services.globals.server_is_ours(&servers[0]));
@@ -1825,6 +1858,7 @@ pub async fn leave_room(
 		displayname: None,
 		third_party_invite: None,
 		blurhash: None,
+		redact_events: None,
 	};
 
 	let is_banned = services.rooms.metadata.is_banned(room_id);

src/api/client/profile.rs

@@ -36,6 +36,9 @@ 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");
+	if services.users.is_suspended(sender_user).await? {
+		return Err!(Request(UserSuspended("You cannot perform this action while suspended.")));
+	}
 
 	if *sender_user != body.user_id && body.appservice_info.is_none() {
 		return Err!(Request(Forbidden("You cannot update the profile of another user")));
@@ -125,6 +128,9 @@ 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");
+	if services.users.is_suspended(sender_user).await? {
+		return Err!(Request(UserSuspended("You cannot perform this action while suspended.")));
+	}
 
 	if *sender_user != body.user_id && body.appservice_info.is_none() {
 		return Err!(Request(Forbidden("You cannot update the profile of another user")));
@@ -343,6 +349,7 @@ pub async fn update_displayname(
 				reason: None,
 				is_direct: None,
 				third_party_invite: None,
+				redact_events: None,
 			});
 
 			Ok((pdu, room_id))
@@ -396,6 +403,7 @@ pub async fn update_avatar_url(
 				reason: None,
 				is_direct: None,
 				third_party_invite: None,
+				redact_events: None,
 			});
 
 			Ok((pdu, room_id))

src/api/client/read_marker.rs

@@ -58,29 +58,34 @@ pub(crate) async fn set_read_marker_route(
 	}
 
 	if let Some(event) = &body.read_receipt {
-		let receipt_content = BTreeMap::from_iter([(
+		if !services.users.is_suspended(sender_user).await? {
-			event.to_owned(),
+			let receipt_content = BTreeMap::from_iter([(
-			BTreeMap::from_iter([(
+				event.to_owned(),
-				ReceiptType::Read,
+				BTreeMap::from_iter([(
-				BTreeMap::from_iter([(sender_user.to_owned(), ruma::events::receipt::Receipt {
+					ReceiptType::Read,
-					ts: Some(MilliSecondsSinceUnixEpoch::now()),
+					BTreeMap::from_iter([(
-					thread: ReceiptThread::Unthreaded,
+						sender_user.to_owned(),
-				})]),
+						ruma::events::receipt::Receipt {
-			)]),
+							ts: Some(MilliSecondsSinceUnixEpoch::now()),
-		)]);
+							thread: ReceiptThread::Unthreaded,
+						},
+					)]),
+				)]),
+			)]);
 
-		services
+			services
-			.rooms
+				.rooms
-			.read_receipt
+				.read_receipt
-			.readreceipt_update(
+				.readreceipt_update(
-				sender_user,
+					sender_user,
-				&body.room_id,
+					&body.room_id,
-				&ruma::events::receipt::ReceiptEvent {
+					&ruma::events::receipt::ReceiptEvent {
-					content: ruma::events::receipt::ReceiptEventContent(receipt_content),
+						content: ruma::events::receipt::ReceiptEventContent(receipt_content),
-					room_id: body.room_id.clone(),
+						room_id: body.room_id.clone(),
-				},
+					},
-			)
+				)
-			.await;
+				.await;
+		}
 	}
 
 	if let Some(event) = &body.private_read_receipt {

src/api/client/redact.rs

@@ -1,5 +1,5 @@
 use axum::extract::State;
-use conduwuit::{Result, matrix::pdu::PduBuilder};
+use conduwuit::{Err, Result, matrix::pdu::PduBuilder};
 use ruma::{
 	api::client::redact::redact_event, events::room::redaction::RoomRedactionEventContent,
 };
@@ -17,6 +17,10 @@ pub(crate) async fn redact_event_route(
 ) -> Result<redact_event::v3::Response> {
 	let sender_user = body.sender_user.as_ref().expect("user is authenticated");
 	let body = body.body;
+	if services.users.is_suspended(sender_user).await? {
+		// TODO: Users can redact their own messages while suspended
+		return Err!(Request(UserSuspended("You cannot perform this action while suspended.")));
+	}
 
 	let state_lock = services.rooms.state.mutex.lock(&body.room_id).await;
 

src/api/client/room/create.rs

@@ -70,6 +70,10 @@ pub(crate) async fn create_room_route(
 		));
 	}
 
+	if services.users.is_suspended(sender_user).await? {
+		return Err!(Request(UserSuspended("You cannot perform this action while suspended.")));
+	}
+
 	let room_id: OwnedRoomId = match &body.room_id {
 		| Some(custom_room_id) => custom_room_id_check(&services, custom_room_id)?,
 		| _ => RoomId::new(&services.server.name),

src/api/client/room/upgrade.rs

@@ -2,7 +2,7 @@ use std::cmp::max;
 
 use axum::extract::State;
 use conduwuit::{
-	Error, Result, err, info,
+	Err, Error, Result, err, info,
 	matrix::{StateKey, pdu::PduBuilder},
 };
 use futures::StreamExt;
@@ -63,6 +63,10 @@ pub(crate) async fn upgrade_room_route(
 		));
 	}
 
+	if services.users.is_suspended(sender_user).await? {
+		return Err!(Request(UserSuspended("You cannot perform this action while suspended.")));
+	}
+
 	// Create a replacement room
 	let replacement_room = RoomId::new(services.globals.server_name());
 
@@ -189,6 +193,7 @@ pub(crate) async fn upgrade_room_route(
 					blurhash: services.users.blurhash(sender_user).await.ok(),
 					reason: None,
 					join_authorized_via_users_server: None,
+					redact_events: None,
 				})
 				.expect("event is valid, we just created it"),
 				unsigned: None,

src/api/client/send.rs

@@ -23,6 +23,9 @@ pub(crate) async fn send_message_event_route(
 	let sender_user = body.sender_user();
 	let sender_device = body.sender_device.as_deref();
 	let appservice_info = body.appservice_info.as_ref();
+	if services.users.is_suspended(sender_user).await? {
+		return Err!(Request(UserSuspended("You cannot perform this action while suspended.")));
+	}
 
 	// Forbid m.room.encrypted if encryption is disabled
 	if MessageLikeEventType::RoomEncrypted == body.event_type && !services.config.allow_encryption

src/api/client/state.rs

@@ -34,6 +34,10 @@ pub(crate) async fn send_state_event_for_key_route(
 ) -> Result<send_state_event::v3::Response> {
 	let sender_user = body.sender_user();
 
+	if services.users.is_suspended(sender_user).await? {
+		return Err!(Request(UserSuspended("You cannot perform this action while suspended.")));
+	}
+
 	Ok(send_state_event::v3::Response {
 		event_id: send_state_event_for_key_helper(
 			&services,

src/api/client/typing.rs

@@ -26,41 +26,42 @@ pub(crate) async fn create_typing_event_route(
 	{
 		return Err!(Request(Forbidden("You are not in this room.")));
 	}
-
+	if !services.users.is_suspended(sender_user).await? {
-	match body.state {
+		match body.state {
-		| Typing::Yes(duration) => {
+			| Typing::Yes(duration) => {
-			let duration = utils::clamp(
+				let duration = utils::clamp(
-				duration.as_millis().try_into().unwrap_or(u64::MAX),
+					duration.as_millis().try_into().unwrap_or(u64::MAX),
+					services
+						.server
+						.config
+						.typing_client_timeout_min_s
+						.try_mul(1000)?,
+					services
+						.server
+						.config
+						.typing_client_timeout_max_s
+						.try_mul(1000)?,
+				);
 				services
-					.server
+					.rooms
-					.config
+					.typing
-					.typing_client_timeout_min_s
+					.typing_add(
-					.try_mul(1000)?,
+						sender_user,
+						&body.room_id,
+						utils::millis_since_unix_epoch()
+							.checked_add(duration)
+							.expect("user typing timeout should not get this high"),
+					)
+					.await?;
+			},
+			| _ => {
 				services
-					.server
+					.rooms
-					.config
+					.typing
-					.typing_client_timeout_max_s
+					.typing_remove(sender_user, &body.room_id)
-					.try_mul(1000)?,
+					.await?;
-			);
+			},
-			services
+		}
-				.rooms
-				.typing
-				.typing_add(
-					sender_user,
-					&body.room_id,
-					utils::millis_since_unix_epoch()
-						.checked_add(duration)
-						.expect("user typing timeout should not get this high"),
-				)
-				.await?;
-		},
-		| _ => {
-			services
-				.rooms
-				.typing
-				.typing_remove(sender_user, &body.room_id)
-				.await?;
-		},
 	}
 
 	// ping presence

src/database/maps.rs

@@ -378,6 +378,10 @@ pub(super) static MAPS: &[Descriptor] = &[
 		name: "userid_password",
 		..descriptor::RANDOM
 	},
+	Descriptor {
+		name: "userid_suspension",
+		..descriptor::RANDOM_SMALL
+	},
 	Descriptor {
 		name: "userid_presenceid",
 		..descriptor::RANDOM_SMALL

src/service/admin/mod.rs

@@ -44,11 +44,13 @@ struct Services {
 	services: StdRwLock<Option<Weak<crate::Services>>>,
 }
 
-/// Inputs to a command are a multi-line string and optional reply_id.
+/// Inputs to a command are a multi-line string, optional reply_id, and optional
+/// sender.
 #[derive(Debug)]
 pub struct CommandInput {
 	pub command: String,
 	pub reply_id: Option<OwnedEventId>,
+	pub sender: Option<Box<UserId>>,
 }
 
 /// Prototype of the tab-completer. The input is buffered text when tab
@@ -161,7 +163,22 @@ impl Service {
 	pub fn command(&self, command: String, reply_id: Option<OwnedEventId>) -> Result<()> {
 		self.channel
 			.0
-			.send(CommandInput { command, reply_id })
+			.send(CommandInput { command, reply_id, sender: None })
+			.map_err(|e| err!("Failed to enqueue admin command: {e:?}"))
+	}
+
+	/// Posts a command to the command processor queue with sender information
+	/// and returns. Processing will take place on the service worker's task
+	/// asynchronously. Errors if the queue is full.
+	pub fn command_with_sender(
+		&self,
+		command: String,
+		reply_id: Option<OwnedEventId>,
+		sender: Box<UserId>,
+	) -> Result<()> {
+		self.channel
+			.0
+			.send(CommandInput { command, reply_id, sender: Some(sender) })
 			.map_err(|e| err!("Failed to enqueue admin command: {e:?}"))
 	}
 
@@ -172,7 +189,7 @@ impl Service {
 		command: String,
 		reply_id: Option<OwnedEventId>,
 	) -> ProcessorResult {
-		self.process_command(CommandInput { command, reply_id })
+		self.process_command(CommandInput { command, reply_id, sender: None })
 			.await
 	}
 

src/service/rooms/timeline/mod.rs

@@ -532,9 +532,11 @@ impl Service {
 					self.services.search.index_pdu(shortroomid, &pdu_id, &body);
 
 					if self.services.admin.is_admin_command(pdu, &body).await {
-						self.services
+						self.services.admin.command_with_sender(
-							.admin
+							body,
-							.command(body, Some((*pdu.event_id).into()))?;
+							Some((*pdu.event_id).into()),
+							pdu.sender.clone().into(),
+						)?;
 					}
 				}
 			},

src/service/users/mod.rs

@@ -16,10 +16,21 @@ use ruma::{
 	},
 	serde::Raw,
 };
+use serde::{Deserialize, Serialize};
 use serde_json::json;
 
 use crate::{Dep, account_data, admin, globals, rooms};
 
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct UserSuspension {
+	/// Whether the user is currently suspended
+	pub suspended: bool,
+	/// When the user was suspended (Unix timestamp in milliseconds)
+	pub suspended_at: u64,
+	/// User ID of who suspended this user
+	pub suspended_by: String,
+}
+
 pub struct Service {
 	services: Services,
 	db: Data,
@@ -52,6 +63,7 @@ struct Data {
 	userid_lastonetimekeyupdate: Arc<Map>,
 	userid_masterkeyid: Arc<Map>,
 	userid_password: Arc<Map>,
+	userid_suspension: Arc<Map>,
 	userid_selfsigningkeyid: Arc<Map>,
 	userid_usersigningkeyid: Arc<Map>,
 	useridprofilekey_value: Arc<Map>,
@@ -87,6 +99,7 @@ impl crate::Service for Service {
 				userid_lastonetimekeyupdate: args.db["userid_lastonetimekeyupdate"].clone(),
 				userid_masterkeyid: args.db["userid_masterkeyid"].clone(),
 				userid_password: args.db["userid_password"].clone(),
+				userid_suspension: args.db["userid_suspension"].clone(),
 				userid_selfsigningkeyid: args.db["userid_selfsigningkeyid"].clone(),
 				userid_usersigningkeyid: args.db["userid_usersigningkeyid"].clone(),
 				useridprofilekey_value: args.db["useridprofilekey_value"].clone(),
@@ -143,6 +156,23 @@ impl Service {
 		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);
+	}
+
 	/// Check if a user has an account on this homeserver.
 	#[inline]
 	pub async fn exists(&self, user_id: &UserId) -> bool {
@@ -159,6 +189,25 @@ impl Service {
 			.await
 	}
 
+	/// Check if account is suspended
+	pub async fn is_suspended(&self, user_id: &UserId) -> 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)
+				},
+		}
+	}
+
 	/// Check if account is active, infallible
 	pub async fn is_active(&self, user_id: &UserId) -> bool {
 		!self.is_deactivated(user_id).await.unwrap_or(true)