Separate active state from lock state in admin API

- Allow the admin API to deactivate a user without locking it, and to
  unlock a user without reactivating it.
- Make unlock-and-reactivate flows unset the "deactivated_at" timestamp.
- Revert adding an "unlock" parameter on `ReactivateUserJob`, as the
  option is used only by the admin API which doesn't use a job.

Author: AndrewFerr

crates/cli/src/commands/manage.rs

@@ -542,7 +542,7 @@ impl Options {
 
                 warn!(%user.id, "User scheduling user reactivation");
                 repo.queue_job()
-                    .schedule_job(&mut rng, &clock, ReactivateUserJob::new(&user, true))
+                    .schedule_job(&mut rng, &clock, ReactivateUserJob::new(&user))
                     .await?;
 
                 repo.into_inner().commit().await?;

crates/handlers/src/admin/v1/users/deactivate.rs

@@ -12,6 +12,8 @@ use mas_storage::{
     BoxRng,
     queue::{DeactivateUserJob, QueueJobRepositoryExt as _},
 };
+use schemars::JsonSchema;
+use serde::Deserialize;
 use tracing::info;
 use ulid::Ulid;
 
@@ -49,7 +51,25 @@ impl IntoResponse for RouteError {
     }
 }
 
-pub fn doc(operation: TransformOperation) -> TransformOperation {
+/// # JSON payload for the `POST /api/admin/v1/users/:id/deactivate` endpoint
+#[derive(Default, Deserialize, JsonSchema)]
+#[serde(rename = "DeactivateUserRequest")]
+pub struct Request {
+    /// Whether to skip locking the user before deactivation.
+    #[serde(default)]
+    skip_lock: bool,
+}
+
+pub fn doc(mut operation: TransformOperation) -> TransformOperation {
+    operation
+        .inner_mut()
+        .request_body
+        .as_mut()
+        .unwrap()
+        .as_item_mut()
+        .unwrap()
+        .required = false;
+
     operation
         .id("deactivateUser")
         .summary("Deactivate a user")
@@ -76,15 +96,17 @@ pub async fn handler(
     }: CallContext,
     NoApi(mut rng): NoApi<BoxRng>,
     id: UlidPathParam,
+    body: Option<Json<Request>>,
 ) -> Result<Json<SingleResponse<User>>, RouteError> {
+    let Json(params) = body.unwrap_or_default();
     let id = *id;
     let mut user = repo
         .user()
         .lookup(id)
         .await?
         .ok_or(RouteError::NotFound(id))?;
 
-    if user.locked_at.is_none() {
+    if !params.skip_lock && user.locked_at.is_none() {
         user = repo.user().lock(&clock, user).await?;
     }
 
@@ -105,14 +127,13 @@ pub async fn handler(
 mod tests {
     use chrono::Duration;
     use hyper::{Request, StatusCode};
-    use insta::assert_json_snapshot;
+    use insta::{allow_duplicates, assert_json_snapshot};
     use mas_storage::{Clock, RepositoryAccess, user::UserRepository};
     use sqlx::PgPool;
 
     use crate::test_utils::{RequestBuilderExt, ResponseExt, TestState, setup};
 
-    #[sqlx::test(migrator = "mas_storage_pg::MIGRATOR")]
-    async fn test_deactivate_user(pool: PgPool) {
+    async fn test_deactivate_user_helper(pool: PgPool, skip_lock: Option<bool>) {
         setup();
         let mut state = TestState::from_pool(pool.clone()).await.unwrap();
         let token = state.token_with_scope("urn:mas:admin").await;
@@ -125,19 +146,27 @@ mod tests {
             .unwrap();
         repo.save().await.unwrap();
 
-        let request = Request::post(format!("/api/admin/v1/users/{}/deactivate", user.id))
-            .bearer(&token)
-            .empty();
+        let request =
+            Request::post(format!("/api/admin/v1/users/{}/deactivate", user.id)).bearer(&token);
+        let request = match skip_lock {
+            None => request.empty(),
+            Some(skip_lock) => request.json(serde_json::json!({
+                "skip_lock": skip_lock,
+            })),
+        };
         let response = state.request(request).await;
         response.assert_status(StatusCode::OK);
         let body: serde_json::Value = response.json();
 
-        // The locked_at timestamp should be the same as the current time
+        // The locked_at timestamp should be the same as the current time, or null if not locked
         assert_eq!(
             body["data"]["attributes"]["locked_at"],
-            serde_json::json!(state.clock.now())
+            if !skip_lock.unwrap_or(false) {
+                serde_json::json!(state.clock.now())
+            } else {
+                serde_json::Value::Null
+            }
         );
-        // TODO: have test coverage on deactivated_at timestamp
 
         // Make sure to run the jobs in the queue
         state.run_jobs_in_queue().await;
@@ -149,7 +178,7 @@ mod tests {
         response.assert_status(StatusCode::OK);
         let body: serde_json::Value = response.json();
 
-        assert_json_snapshot!(body, @r#"
+        allow_duplicates!(assert_json_snapshot!(body, @r#"
         {
           "data": {
             "type": "user",
@@ -169,7 +198,17 @@ mod tests {
             "self": "/api/admin/v1/users/01FSHN9AG0MZAA6S4AF7CTV32E"
           }
         }
-        "#);
+        "#));
+    }
+
+    #[sqlx::test(migrator = "mas_storage_pg::MIGRATOR")]
+    async fn test_deactivate_user(pool: PgPool) {
+        test_deactivate_user_helper(pool, Option::None).await;
+    }
+
+    #[sqlx::test(migrator = "mas_storage_pg::MIGRATOR")]
+    async fn test_deactivate_user_skip_lock(pool: PgPool) {
+        test_deactivate_user_helper(pool, Option::Some(true)).await;
     }
 
     #[sqlx::test(migrator = "mas_storage_pg::MIGRATOR")]
@@ -206,7 +245,6 @@ mod tests {
             body["data"]["attributes"]["locked_at"],
             serde_json::Value::Null
         );
-        // TODO: have test coverage on deactivated_at timestamp
 
         // Make sure to run the jobs in the queue
         state.run_jobs_in_queue().await;

crates/handlers/src/admin/v1/users/reactivate.rs

@@ -3,15 +3,13 @@
 // SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
 // Please see LICENSE files in the repository root for full details.
 
-use aide::{NoApi, OperationIo, transform::TransformOperation};
-use axum::{Json, response::IntoResponse};
+use std::sync::Arc;
+
+use aide::{OperationIo, transform::TransformOperation};
+use axum::{Json, extract::State, response::IntoResponse};
 use hyper::StatusCode;
 use mas_axum_utils::record_error;
-use mas_storage::{
-    BoxRng,
-    queue::{QueueJobRepositoryExt as _, ReactivateUserJob},
-};
-use tracing::info;
+use mas_matrix::HomeserverConnection;
 use ulid::Ulid;
 
 use crate::{
@@ -30,6 +28,9 @@ pub enum RouteError {
     #[error(transparent)]
     Internal(Box<dyn std::error::Error + Send + Sync + 'static>),
 
+    #[error(transparent)]
+    Homeserver(anyhow::Error),
+
     #[error("User ID {0} not found")]
     NotFound(Ulid),
 }
@@ -39,9 +40,9 @@ impl_from_error_for_route!(mas_storage::RepositoryError);
 impl IntoResponse for RouteError {
     fn into_response(self) -> axum::response::Response {
         let error = ErrorResponse::from_error(&self);
-        let sentry_event_id = record_error!(self, Self::Internal(_));
+        let sentry_event_id = record_error!(self, Self::Internal(_) | Self::Homeserver(_));
         let status = match self {
-            Self::Internal(_) => StatusCode::INTERNAL_SERVER_ERROR,
+            Self::Internal(_) | Self::Homeserver(_) => StatusCode::INTERNAL_SERVER_ERROR,
             Self::NotFound(_) => StatusCode::NOT_FOUND,
         };
         (status, sentry_event_id, Json(error)).into_response()
@@ -69,10 +70,8 @@ pub fn doc(operation: TransformOperation) -> TransformOperation {
 
 #[tracing::instrument(name = "handler.admin.v1.users.reactivate", skip_all)]
 pub async fn handler(
-    CallContext {
-        mut repo, clock, ..
-    }: CallContext,
-    NoApi(mut rng): NoApi<BoxRng>,
+    CallContext { mut repo, .. }: CallContext,
+    State(homeserver): State<Arc<dyn HomeserverConnection>>,
     id: UlidPathParam,
 ) -> Result<Json<SingleResponse<User>>, RouteError> {
     let id = *id;
@@ -82,10 +81,15 @@ pub async fn handler(
         .await?
         .ok_or(RouteError::NotFound(id))?;
 
-    info!(%user.id, "Scheduling reactivation of user");
-    repo.queue_job()
-        .schedule_job(&mut rng, &clock, ReactivateUserJob::new(&user, false))
-        .await?;
+    // Call the homeserver synchronously to reactivate the user
+    let mxid = homeserver.mxid(&user.username);
+    homeserver
+        .reactivate_user(&mxid)
+        .await
+        .map_err(RouteError::Homeserver)?;
+
+    // Now reactivate the user in our database
+    let user = repo.user().reactivate(user).await?;
 
     repo.save().await?;
 
@@ -100,7 +104,7 @@ mod tests {
     use hyper::{Request, StatusCode};
     use mas_matrix::{HomeserverConnection, ProvisionRequest};
     use mas_storage::{Clock, RepositoryAccess, user::UserRepository};
-    use sqlx::{PgPool, types::Json};
+    use sqlx::PgPool;
 
     use crate::test_utils::{RequestBuilderExt, ResponseExt, TestState, setup};
 
@@ -150,18 +154,10 @@ mod tests {
             body["data"]["attributes"]["locked_at"],
             serde_json::json!(state.clock.now())
         );
-        // TODO: have test coverage on deactivated_at timestamp
-
-        // It should have scheduled a reactivation job for the user
-        // XXX: we don't have a good way to look for the reactivation job
-        let job: Json<serde_json::Value> = sqlx::query_scalar(
-            "SELECT payload FROM queue_jobs WHERE queue_name = 'reactivate-user'",
-        )
-        .fetch_one(&pool)
-        .await
-        .expect("Reactivation job to be scheduled");
-        assert_eq!(job["user_id"], serde_json::json!(user.id));
-        assert_eq!(job["unlock"], serde_json::Value::Bool(false));
+        assert_eq!(
+            body["data"]["attributes"]["deactivated_at"],
+            serde_json::Value::Null,
+        );
     }
 
     #[sqlx::test(migrator = "mas_storage_pg::MIGRATOR")]
@@ -178,6 +174,14 @@ mod tests {
             .unwrap();
         repo.save().await.unwrap();
 
+        // Provision the user on the homeserver
+        let mxid = state.homeserver_connection.mxid(&user.username);
+        state
+            .homeserver_connection
+            .provision_user(&ProvisionRequest::new(&mxid, &user.sub))
+            .await
+            .unwrap();
+
         let request = Request::post(format!("/api/admin/v1/users/{}/reactivate", user.id))
             .bearer(&token)
             .empty();
@@ -189,18 +193,10 @@ mod tests {
             body["data"]["attributes"]["locked_at"],
             serde_json::Value::Null
         );
-        // TODO: have test coverage on deactivated_at timestamp
-
-        // It should have scheduled a reactivation job for the user
-        // XXX: we don't have a good way to look for the reactivation job
-        let job: Json<serde_json::Value> = sqlx::query_scalar(
-            "SELECT payload FROM queue_jobs WHERE queue_name = 'reactivate-user'",
-        )
-        .fetch_one(&pool)
-        .await
-        .expect("Reactivation job to be scheduled");
-        assert_eq!(job["user_id"], serde_json::json!(user.id));
-        assert_eq!(job["unlock"], serde_json::Value::Bool(false));
+        assert_eq!(
+            body["data"]["attributes"]["deactivated_at"],
+            serde_json::Value::Null
+        );
     }
 
     #[sqlx::test(migrator = "mas_storage_pg::MIGRATOR")]