compat login: support using client-provided device ID

crates/handlers/src/compat/login.rs

@@ -468,6 +468,10 @@ async fn token_login(
         .await
         .map_err(RouteError::ProvisionDeviceFailed)?;
 
+    repo.app_session()
+        .finish_sessions_to_replace_device(clock, &browser_session.user, &device)
+        .await?;
+
     let compat_session = repo
         .compat_session()
         .add(
@@ -563,6 +567,10 @@ async fn user_password_login(
         .await
         .map_err(RouteError::ProvisionDeviceFailed)?;
 
+    repo.app_session()
+        .finish_sessions_to_replace_device(clock, &user, &device)
+        .await?;
+
     let session = repo
         .compat_session()
         .add(&mut rng, clock, &user, device, None, false)

crates/storage-pg/src/app_session.rs

@@ -7,9 +7,11 @@
 //! A module containing PostgreSQL implementation of repositories for sessions
 
 use async_trait::async_trait;
-use mas_data_model::{CompatSession, CompatSessionState, Device, Session, SessionState, UserAgent};
+use mas_data_model::{
+    CompatSession, CompatSessionState, Device, Session, SessionState, User, UserAgent,
+};
 use mas_storage::{
-    Page, Pagination,
+    Clock, Page, Pagination,
     app_session::{AppSession, AppSessionFilter, AppSessionRepository, AppSessionState},
     compat::CompatSessionFilter,
     oauth2::OAuth2SessionFilter,
@@ -21,6 +23,7 @@ use sea_query::{
 use sea_query_binder::SqlxBinder;
 use sqlx::PgConnection;
 use ulid::Ulid;
+use uuid::Uuid;
 
 use crate::{
     DatabaseError, ExecuteExt,
@@ -457,6 +460,54 @@ impl AppSessionRepository for PgAppSessionRepository<'_> {
             .try_into()
             .map_err(DatabaseError::to_invalid_operation)
     }
+
+    #[tracing::instrument(
+        name = "db.app_session.finish_sessions_to_replace_device",
+        fields(
+            db.query.text,
+            %user.id,
+            %device_id = device.as_str()
+        ),
+        skip_all,
+        err,
+    )]
+    async fn finish_sessions_to_replace_device(
+        &mut self,
+        clock: &dyn Clock,
+        user: &User,
+        device: &Device,
+    ) -> Result<(), Self::Error> {
+        // TODO need to invoke this from all the oauth2 login sites
+        // TODO CREATE A SECOND SPAN FOR THE SECOND QUERY
+        let finished_at = clock.now();
+        sqlx::query!(
+            "
+                UPDATE compat_sessions SET finished_at = $3 WHERE user_id = $1 AND device_id = $2 AND finished_at IS NULL
+            ",
+            Uuid::from(user.id),
+            device.as_str(),
+            finished_at
+        )
+        .traced()
+        .execute(&mut *self.conn)
+        .await?;
+
+        if let Ok(device_as_scope_token) = device.to_scope_token() {
+            sqlx::query!(
+                "
+                    UPDATE oauth2_sessions SET finished_at = $3 WHERE user_id = $1 AND $2 = ANY(scope_list) AND finished_at IS NULL
+                ",
+                Uuid::from(user.id),
+                device_as_scope_token.as_str(),
+                finished_at
+            )
+            .traced()
+            .execute(&mut *self.conn)
+            .await?;
+        }
+
+        Ok(())
+    }
 }
 
 #[cfg(test)]

crates/storage/src/app_session.rs

@@ -10,7 +10,7 @@ use async_trait::async_trait;
 use chrono::{DateTime, Utc};
 use mas_data_model::{BrowserSession, CompatSession, Device, Session, User};
 
-use crate::{Page, Pagination, repository_impl};
+use crate::{Clock, Page, Pagination, repository_impl};
 
 /// The state of a session
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
@@ -188,6 +188,20 @@ pub trait AppSessionRepository: Send + Sync {
     ///
     /// Returns [`Self::Error`] if the underlying repository fails
     async fn count(&mut self, filter: AppSessionFilter<'_>) -> Result<usize, Self::Error>;
+
+    /// Finishes any application sessions that are using the specified device's
+    /// ID.
+    ///
+    /// This is intended for logging in using an existing device ID (i.e.
+    /// replacing a device).
+    ///
+    /// Should be called *before* creating a new session for the device.
+    async fn finish_sessions_to_replace_device(
+        &mut self,
+        clock: &dyn Clock,
+        user: &User,
+        device: &Device,
+    ) -> Result<(), Self::Error>;
 }
 
 repository_impl!(AppSessionRepository:
@@ -198,4 +212,11 @@ repository_impl!(AppSessionRepository:
     ) -> Result<Page<AppSession>, Self::Error>;
 
     async fn count(&mut self, filter: AppSessionFilter<'_>) -> Result<usize, Self::Error>;
+
+    async fn finish_sessions_to_replace_device(
+        &mut self,
+        clock: &dyn Clock,
+        user: &User,
+        device: &Device,
+    ) -> Result<(), Self::Error>;
 );