When adding or revoking personal sessions, schedule device synchronisation (#5182)

crates/data-model/src/personal/session.rs

@@ -10,7 +10,7 @@ use oauth2_types::scope::Scope;
 use serde::Serialize;
 use ulid::Ulid;
 
-use crate::{Client, InvalidTransitionError, User};
+use crate::{Client, Device, InvalidTransitionError, User};
 
 #[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)]
 pub enum SessionState {
@@ -129,4 +129,13 @@ impl PersonalSession {
         self.state = self.state.revoke(revoked_at)?;
         Ok(self)
     }
+
+    /// Returns whether the scope of this session contains a device scope;
+    /// in other words: whether this session has a device.
+    #[must_use]
+    pub fn has_device(&self) -> bool {
+        self.scope
+            .iter()
+            .any(|scope_token| Device::from_scope_token(scope_token).is_some())
+    }
 }

crates/handlers/src/admin/v1/personal_sessions/add.rs

@@ -3,12 +3,16 @@
 // SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
 // Please see LICENSE files in the repository root for full details.
 
+use std::sync::Arc;
+
 use aide::{NoApi, OperationIo, transform::TransformOperation};
-use axum::{Json, response::IntoResponse};
+use anyhow::Context;
+use axum::{Json, extract::State, response::IntoResponse};
 use chrono::Duration;
 use hyper::StatusCode;
 use mas_axum_utils::record_error;
-use mas_data_model::{BoxRng, TokenType};
+use mas_data_model::{BoxRng, Device, TokenType};
+use mas_matrix::HomeserverConnection;
 use oauth2_types::scope::Scope;
 use schemars::JsonSchema;
 use serde::Deserialize;
@@ -99,6 +103,7 @@ pub async fn handler(
         ..
     }: CallContext,
     NoApi(mut rng): NoApi<BoxRng>,
+    NoApi(State(homeserver)): NoApi<State<Arc<dyn HomeserverConnection>>>,
     Json(params): Json<Request>,
 ) -> Result<(StatusCode, Json<SingleResponse<PersonalSession>>), RouteError> {
     let owner = personal_session_owner_from_caller(&session);
@@ -139,6 +144,28 @@ pub async fn handler(
         )
         .await?;
 
+    // If the session has a device, we should add those to the homeserver now
+    if session.has_device() {
+        // Lock the user sync to make sure we don't get into a race condition
+        repo.user().acquire_lock_for_sync(&actor_user).await?;
+
+        for scope in &*session.scope {
+            if let Some(device) = Device::from_scope_token(scope) {
+                // NOTE: We haven't relinquished the repo at this point,
+                // so we are holding a transaction across the homeserver
+                // operation.
+                // This is suboptimal, but simpler.
+                // Given this is an administrative endpoint, this is a tolerable
+                // compromise for now.
+                homeserver
+                    .upsert_device(&actor_user.username, device.as_str(), None)
+                    .await
+                    .context("Failed to provision device")
+                    .map_err(|e| RouteError::Internal(e.into()))?;
+            }
+        }
+    }
+
     repo.save().await?;
 
     Ok((

crates/handlers/src/admin/v1/personal_sessions/revoke.rs

@@ -3,10 +3,12 @@
 // SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
 // Please see LICENSE files in the repository root for full details.
 
-use aide::{OperationIo, transform::TransformOperation};
+use aide::{NoApi, OperationIo, transform::TransformOperation};
 use axum::{Json, response::IntoResponse};
 use hyper::StatusCode;
 use mas_axum_utils::record_error;
+use mas_data_model::BoxRng;
+use mas_storage::queue::{QueueJobRepositoryExt as _, SyncDevicesJob};
 use ulid::Ulid;
 
 use crate::{
@@ -80,6 +82,7 @@ pub async fn handler(
     CallContext {
         mut repo, clock, ..
     }: CallContext,
+    NoApi(mut rng): NoApi<BoxRng>,
     session_id: UlidPathParam,
 ) -> Result<Json<SingleResponse<PersonalSession>>, RouteError> {
     let session_id = *session_id;
@@ -95,6 +98,18 @@ pub async fn handler(
 
     let session = repo.personal_session().revoke(&clock, session).await?;
 
+    if session.has_device() {
+        // If the session has a device, then we are now
+        // deleting a device and should schedule a device sync to clean up.
+        repo.queue_job()
+            .schedule_job(
+                &mut rng,
+                &clock,
+                SyncDevicesJob::new_for_id(session.actor_user_id),
+            )
+            .await?;
+    }
+
     repo.save().await?;
 
     Ok(Json(SingleResponse::new_canonical(