Implement command to deactivate user from admin channel

Use `leave_room` in `leave_all_rooms`

WIP: Add command to delete a list of users
also implements a flag to prevent the user from being removed from their joined rooms.

Report user deactivation failure reason

Don't send leave events by default when mass deactivating user accounts

Don't stop leaving rooms if an error was encountered

WIP: Rename command, make flags consistent, don't deactivate admin accounts.
Accounts should be deactivated as fast as possible and removing users from joined groups is completed afterwards.

Fix admin safety logic, improve command output

Continue leaving rooms if a room_id is invalid

Ignore errors from leave_room

Add notice to the list-local-users command
Output form list-local-users can be used directly without modification with the deactivate-all command

Only get mutex lock for admin room when sending message

src/client_server/account.rs

@@ -4,7 +4,7 @@ use super::{DEVICE_ID_LENGTH, SESSION_ID_LENGTH, TOKEN_LENGTH};
 use crate::{
     database::{admin::make_user_admin, DatabaseGuard},
     pdu::PduBuilder,
-    utils, Error, Result, Ruma,
+    utils, Database, Error, Result, Ruma,
 };
 use ruma::{
     api::client::{
@@ -398,55 +398,8 @@ pub async fn deactivate_route(
         return Err(Error::BadRequest(ErrorKind::NotJson, "Not json."));
     }
 
-    // Leave all joined rooms and reject all invitations
+    // Make the user leave all rooms before deactivation
-    // TODO: work over federation invites
+    db.rooms.leave_all_rooms(&sender_user, &db).await?;
-    let all_rooms = db
-        .rooms
-        .rooms_joined(sender_user)
-        .chain(
-            db.rooms
-                .rooms_invited(sender_user)
-                .map(|t| t.map(|(r, _)| r)),
-        )
-        .collect::<Vec<_>>();
-
-    for room_id in all_rooms {
-        let room_id = room_id?;
-        let event = RoomMemberEventContent {
-            membership: MembershipState::Leave,
-            displayname: None,
-            avatar_url: None,
-            is_direct: None,
-            third_party_invite: None,
-            blurhash: None,
-            reason: None,
-            join_authorized_via_users_server: None,
-        };
-
-        let mutex_state = Arc::clone(
-            db.globals
-                .roomid_mutex_state
-                .write()
-                .unwrap()
-                .entry(room_id.clone())
-                .or_default(),
-        );
-        let state_lock = mutex_state.lock().await;
-
-        db.rooms.build_and_append_pdu(
-            PduBuilder {
-                event_type: RoomEventType::RoomMember,
-                content: to_raw_value(&event).expect("event is valid, we just created it"),
-                unsigned: None,
-                state_key: Some(sender_user.to_string()),
-                redacts: None,
-            },
-            sender_user,
-            &room_id,
-            &db,
-            &state_lock,
-        )?;
-    }
 
     // Remove devices and mark account as deactivated
     db.users.deactivate_account(sender_user)?;

src/database/admin.rs

@@ -101,6 +101,12 @@ impl Admin {
                 tokio::select! {
                     Some(event) = receiver.recv() => {
                         let guard = db.read().await;
+
+                        let message_content = match event {
+                            AdminRoomEvent::SendMessage(content) => content,
+                            AdminRoomEvent::ProcessMessage(room_message) => process_admin_message(&*guard, room_message).await
+                        };
+
                         let mutex_state = Arc::clone(
                             guard.globals
                                 .roomid_mutex_state
@@ -109,18 +115,10 @@ impl Admin {
                                 .entry(conduit_room.clone())
                                 .or_default(),
                         );
+
                         let state_lock = mutex_state.lock().await;
 
-                        match event {
+                        send_message(message_content, guard, &state_lock);
-                            AdminRoomEvent::SendMessage(content) => {
-                                send_message(content, guard, &state_lock);
-                            }
-                            AdminRoomEvent::ProcessMessage(room_message) => {
-                                let reply_message = process_admin_message(&*guard, room_message).await;
-
-                                send_message(reply_message, guard, &state_lock);
-                            }
-                        }
 
                         drop(state_lock);
                     }
@@ -240,6 +238,39 @@ enum AdminCommand {
     /// List all rooms we are currently handling an incoming pdu from
     IncomingFederation,
 
+    /// Deactivate a user
+    ///
+    /// User will be removed from all rooms by default.
+    /// This behaviour can be overridden with the --no-leave-rooms flag.
+    DeactivateUser {
+        #[clap(short, long)]
+        leave_rooms: bool,
+        user_id: Box<UserId>,
+    },
+
+    #[clap(verbatim_doc_comment)]
+    /// Deactivate a list of users
+    ///
+    /// Recommended to use in conjunction with list-local-users.
+    ///
+    /// Users will not be removed from joined rooms by default.
+    /// Can be overridden with --leave-rooms flag.
+    /// Removing a mass amount of users from a room may cause a significant amount of leave events.
+    /// The time to leave rooms may depend significantly on joined rooms and servers.
+    ///
+    /// [commandbody]
+    /// # ```
+    /// # User list here
+    /// # ```
+    DeactivateAll {
+        #[clap(short, long)]
+        /// Remove users from their joined rooms
+        leave_rooms: bool,
+        #[clap(short, long)]
+        /// Also deactivate admin accounts
+        force: bool,
+    },
+
     /// Get the auth_chain of a PDU
     GetAuthChain {
         /// An event ID (the $ character followed by the base64 reference hash)
@@ -603,6 +634,97 @@ async fn process_admin_command(
             db.rooms.disabledroomids.remove(room_id.as_bytes())?;
             RoomMessageEventContent::text_plain("Room enabled.")
         }
+        AdminCommand::DeactivateUser {
+            leave_rooms,
+            user_id,
+        } => {
+            let user_id = Arc::<UserId>::from(user_id);
+            if db.users.exists(&user_id)? {
+                RoomMessageEventContent::text_plain(format!(
+                    "Making {} leave all rooms before deactivation...",
+                    user_id
+                ));
+
+                db.users.deactivate_account(&user_id)?;
+
+                if leave_rooms {
+                    db.rooms.leave_all_rooms(&user_id, &db).await?;
+                }
+
+                RoomMessageEventContent::text_plain(format!(
+                    "User {} has been deactivated",
+                    user_id
+                ))
+            } else {
+                RoomMessageEventContent::text_plain(format!(
+                    "User {} doesn't exist on this server",
+                    user_id
+                ))
+            }
+        }
+        AdminCommand::DeactivateAll { leave_rooms, force } => {
+            if body.len() > 2 && body[0].trim() == "```" && body.last().unwrap().trim() == "```" {
+                let usernames = body.clone().drain(1..body.len() - 1).collect::<Vec<_>>();
+
+                let mut user_ids: Vec<&UserId> = Vec::new();
+
+                for &username in &usernames {
+                    match <&UserId>::try_from(username) {
+                        Ok(user_id) => user_ids.push(user_id),
+                        Err(_) => {
+                            return Ok(RoomMessageEventContent::text_plain(format!(
+                                "{} is not a valid username",
+                                username
+                            )))
+                        }
+                    }
+                }
+
+                let mut deactivation_count = 0;
+                let mut admins = Vec::new();
+
+                if !force {
+                    user_ids.retain(|&user_id| {
+                        match db.users.is_admin(user_id, &db.rooms, &db.globals) {
+                            Ok(is_admin) => match is_admin {
+                                true => {
+                                    admins.push(user_id.localpart());
+                                    false
+                                }
+                                false => true,
+                            },
+                            Err(_) => false,
+                        }
+                    })
+                }
+
+                for &user_id in &user_ids {
+                    match db.users.deactivate_account(user_id) {
+                        Ok(_) => deactivation_count += 1,
+                        Err(_) => {}
+                    }
+                }
+
+                if leave_rooms {
+                    for &user_id in &user_ids {
+                        let _ = db.rooms.leave_all_rooms(user_id, &db).await;
+                    }
+                }
+
+                if admins.is_empty() {
+                    RoomMessageEventContent::text_plain(format!(
+                        "Deactivated {} accounts.",
+                        deactivation_count
+                    ))
+                } else {
+                    RoomMessageEventContent::text_plain(format!("Deactivated {} accounts.\nSkipped admin accounts: {:?}. Use --force to deactivate admin accounts", deactivation_count, admins.join(", ")))
+                }
+            } else {
+                RoomMessageEventContent::text_plain(
+                    "Expected code block in command body. Add --help for details.",
+                )
+            }
+        }
     };
 
     Ok(reply_message_content)

src/database/rooms.rs

@@ -2569,6 +2569,27 @@ impl Rooms {
         }
     }
 
+    // Make a user leave all their joined rooms
+    #[tracing::instrument(skip(self, db))]
+    pub async fn leave_all_rooms(&self, user_id: &UserId, db: &Database) -> Result<()> {
+        let all_rooms = db
+            .rooms
+            .rooms_joined(user_id)
+            .chain(db.rooms.rooms_invited(user_id).map(|t| t.map(|(r, _)| r)))
+            .collect::<Vec<_>>();
+
+        for room_id in all_rooms {
+            let room_id = match room_id {
+                Ok(room_id) => room_id,
+                Err(_) => continue,
+            };
+
+            let _ = self.leave_room(user_id, &room_id, db).await;
+        }
+
+        Ok(())
+    }
+
     #[tracing::instrument(skip(self, db))]
     pub async fn leave_room(
         &self,