syn2mas: migrate access tokens, refresh tokens and devices

crates/cli/src/commands/syn2mas.rs

@@ -223,6 +223,7 @@ impl Options {
                 }
                 let mut writer = MasWriter::new(mas_connection, writer_mas_connections).await?;
 
+                let clock = SystemClock::default();
                 // TODO is this rng ok?
                 #[allow(clippy::disallowed_methods)]
                 let mut rng = thread_rng();
@@ -233,6 +234,7 @@ impl Options {
                     &mut reader,
                     &mut writer,
                     &mas_matrix.homeserver,
+                    &clock,
                     &mut rng,
                     &provider_id_mappings,
                 )

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

@@ -72,6 +72,7 @@ pub struct CompatSession {
     pub state: CompatSessionState,
     pub user_id: Ulid,
     pub device: Option<Device>,
+    pub human_name: Option<String>,
     pub user_session_id: Option<Ulid>,
     pub created_at: DateTime<Utc>,
     pub is_synapse_admin: bool,

crates/storage-pg/migrations/20250129154003_compat_sessions_device_name.sql

@@ -0,0 +1,9 @@
+-- Copyright 2025 New Vector Ltd.
+--
+-- SPDX-License-Identifier: AGPL-3.0-only
+-- Please see LICENSE in the repository root for full details.
+
+ALTER TABLE compat_sessions
+  -- Stores a human-readable name for the device.
+  -- syn2mas behaviour: Will be populated from the device name in Synapse.
+  ADD COLUMN human_name TEXT;

crates/storage-pg/migrations/20250130170011_user_is_guest.sql

@@ -0,0 +1,10 @@
+-- Copyright 2025 New Vector Ltd.
+--
+-- SPDX-License-Identifier: AGPL-3.0-only
+-- Please see LICENSE in the repository root for full details.
+
+ALTER TABLE users
+  -- Track whether users are guests.
+  -- Although guest support is not present in MAS yet, syn2mas should import
+  -- these users and therefore we should track their state.
+  ADD COLUMN is_guest BOOLEAN NOT NULL DEFAULT FALSE;

crates/storage-pg/src/app_session.rs

@@ -64,6 +64,7 @@ mod priv_ {
         pub(super) user_id: Option<Uuid>,
         pub(super) scope_list: Option<Vec<String>>,
         pub(super) device_id: Option<String>,
+        pub(super) human_name: Option<String>,
         pub(super) created_at: DateTime<Utc>,
         pub(super) finished_at: Option<DateTime<Utc>>,
         pub(super) is_synapse_admin: Option<bool>,
@@ -91,6 +92,7 @@ impl TryFrom<AppSessionLookup> for AppSession {
             user_id,
             scope_list,
             device_id,
+            human_name,
             created_at,
             finished_at,
             is_synapse_admin,
@@ -141,6 +143,7 @@ impl TryFrom<AppSessionLookup> for AppSession {
                     state,
                     user_id: user_id.into(),
                     device,
+                    human_name,
                     user_session_id,
                     created_at,
                     is_synapse_admin,
@@ -294,6 +297,7 @@ impl AppSessionRepository for PgAppSessionRepository<'_> {
                 AppSessionLookupIden::ScopeList,
             )
             .expr_as(Expr::cust("NULL"), AppSessionLookupIden::DeviceId)
+            .expr_as(Expr::cust("NULL"), AppSessionLookupIden::HumanName)
             .expr_as(
                 Expr::col((OAuth2Sessions::Table, OAuth2Sessions::CreatedAt)),
                 AppSessionLookupIden::CreatedAt,
@@ -343,6 +347,10 @@ impl AppSessionRepository for PgAppSessionRepository<'_> {
                 Expr::col((CompatSessions::Table, CompatSessions::DeviceId)),
                 AppSessionLookupIden::DeviceId,
             )
+            .expr_as(
+                Expr::col((CompatSessions::Table, CompatSessions::HumanName)),
+                AppSessionLookupIden::HumanName,
+            )
             .expr_as(
                 Expr::col((CompatSessions::Table, CompatSessions::CreatedAt)),
                 AppSessionLookupIden::CreatedAt,

crates/storage-pg/src/compat/refresh_token.rs

@@ -204,16 +204,25 @@ impl CompatRefreshTokenRepository for PgCompatRefreshTokenRepository<'_> {
             r#"
                 UPDATE compat_refresh_tokens
                 SET consumed_at = $2
-                WHERE compat_refresh_token_id = $1
+                WHERE compat_session_id = $1
+                  AND consumed_at IS NULL
             "#,
-            Uuid::from(compat_refresh_token.id),
+            Uuid::from(compat_refresh_token.session_id),
             consumed_at,
         )
         .traced()
         .execute(&mut *self.conn)
         .await?;
 
-        DatabaseError::ensure_affected_rows(&res, 1)?;
+        // This can affect multiple rows in case we've imported refresh tokens
+        // from Synapse. What we care about is that it at least affected one,
+        // which is what we're checking here
+        if res.rows_affected() == 0 {
+            return Err(DatabaseError::RowsAffected {
+                expected: 1,
+                actual: 0,
+            });
+        }
 
         let compat_refresh_token = compat_refresh_token
             .consume(consumed_at)

crates/storage-pg/src/compat/session.rs

@@ -48,6 +48,7 @@ impl<'c> PgCompatSessionRepository<'c> {
 struct CompatSessionLookup {
     compat_session_id: Uuid,
     device_id: Option<String>,
+    human_name: Option<String>,
     user_id: Uuid,
     user_session_id: Option<Uuid>,
     created_at: DateTime<Utc>,
@@ -85,6 +86,7 @@ impl TryFrom<CompatSessionLookup> for CompatSession {
             user_id: value.user_id.into(),
             user_session_id: value.user_session_id.map(Ulid::from),
             device,
+            human_name: value.human_name,
             created_at: value.created_at,
             is_synapse_admin: value.is_synapse_admin,
             user_agent: value.user_agent.map(UserAgent::parse),
@@ -101,6 +103,7 @@ impl TryFrom<CompatSessionLookup> for CompatSession {
 struct CompatSessionAndSsoLoginLookup {
     compat_session_id: Uuid,
     device_id: Option<String>,
+    human_name: Option<String>,
     user_id: Uuid,
     user_session_id: Option<Uuid>,
     created_at: DateTime<Utc>,
@@ -143,6 +146,7 @@ impl TryFrom<CompatSessionAndSsoLoginLookup> for (CompatSession, Option<CompatSs
             state,
             user_id: value.user_id.into(),
             device,
+            human_name: value.human_name,
             user_session_id: value.user_session_id.map(Ulid::from),
             created_at: value.created_at,
             is_synapse_admin: value.is_synapse_admin,
@@ -286,6 +290,7 @@ impl CompatSessionRepository for PgCompatSessionRepository<'_> {
             r#"
                 SELECT compat_session_id
                      , device_id
+                     , human_name
                      , user_id
                      , user_session_id
                      , created_at
@@ -356,6 +361,7 @@ impl CompatSessionRepository for PgCompatSessionRepository<'_> {
             state: CompatSessionState::default(),
             user_id: user.id,
             device: Some(device),
+            human_name: None,
             user_session_id: browser_session.map(|s| s.id),
             created_at,
             is_synapse_admin,
@@ -453,6 +459,10 @@ impl CompatSessionRepository for PgCompatSessionRepository<'_> {
                 Expr::col((CompatSessions::Table, CompatSessions::DeviceId)),
                 CompatSessionAndSsoLoginLookupIden::DeviceId,
             )
+            .expr_as(
+                Expr::col((CompatSessions::Table, CompatSessions::HumanName)),
+                CompatSessionAndSsoLoginLookupIden::HumanName,
+            )
             .expr_as(
                 Expr::col((CompatSessions::Table, CompatSessions::UserId)),
                 CompatSessionAndSsoLoginLookupIden::UserId,

crates/storage-pg/src/iden.rs

@@ -43,6 +43,7 @@ pub enum CompatSessions {
     CompatSessionId,
     UserId,
     DeviceId,
+    HumanName,
     UserSessionId,
     CreatedAt,
     FinishedAt,

crates/storage/src/compat/refresh_token.rs

@@ -69,7 +69,13 @@ pub trait CompatRefreshTokenRepository: Send + Sync {
         token: String,
     ) -> Result<CompatRefreshToken, Self::Error>;
 
-    /// Consume a compat refresh token
+    /// Consume a compat refresh token.
+    ///
+    /// This also marks other refresh tokens in the same session as consumed.
+    /// This is desirable because the syn2mas migration process can import
+    /// multiple refresh tokens for one device (compat session).
+    /// But once the user uses one of those, the others should no longer
+    /// be valid.
     ///
     /// Returns the consumed compat refresh token
     ///