Re-enable the upstream authentication sessions cleanup job (#5439)

crates/handlers/src/upstream_oauth2/backchannel_logout.rs

@@ -1,3 +1,4 @@
+// Copyright 2025, 2026 Element Creations Ltd.
 // Copyright 2025 New Vector Ltd.
 //
 // SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
@@ -249,14 +250,14 @@ pub(crate) async fn post(
         }
         UpstreamOAuthProviderOnBackchannelLogout::LogoutBrowserOnly => {
             let filter = BrowserSessionFilter::new()
-                .authenticated_by_upstream_sessions_only(auth_session_filter)
+                .linked_to_upstream_sessions_only(auth_session_filter)
                 .active_only();
             let affected = repo.browser_session().finish_bulk(&clock, filter).await?;
             tracing::info!("Finished {affected} browser sessions");
         }
         UpstreamOAuthProviderOnBackchannelLogout::LogoutAll => {
-            let browser_session_filter = BrowserSessionFilter::new()
+            let browser_session_filter =
-                .authenticated_by_upstream_sessions_only(auth_session_filter);
+                BrowserSessionFilter::new().linked_to_upstream_sessions_only(auth_session_filter);
 
             // We need to loop through all the browser sessions to find all the
             // users affected so that we can trigger a device sync job for them

crates/storage-pg/migrations/20260121104214_upstream_auth_user_session_fk_idx.sql

@@ -0,0 +1,11 @@
+-- no-transaction
+-- Copyright 2026 Element Creations Ltd.
+--
+-- SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
+-- Please see LICENSE files in the repository root for full details.
+
+-- Adds an index on the user_session_id column on the
+-- upstream_oauth_authorization_sessions table
+CREATE INDEX CONCURRENTLY IF NOT EXISTS
+  upstream_oauth_authorization_sessions_user_session_id_idx
+  ON upstream_oauth_authorization_sessions (user_session_id);

crates/storage-pg/migrations/20260121121140_upstream_oauth_track_user_session_trigger.sql

@@ -0,0 +1,27 @@
+-- Copyright 2026 Element Creations Ltd.
+--
+-- SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
+-- Please see LICENSE files in the repository root for full details.
+
+-- Adds a trigger which will backfill the user_session_id column when inserting
+-- a new user_session_authentications row. This is to help supporting rolling
+-- back to previous releases and should be dropped in a future version.
+CREATE OR REPLACE FUNCTION upstream_oauth_authorization_sessions_insert_trigger()
+RETURNS TRIGGER AS $$
+BEGIN
+    IF NEW.upstream_oauth_authorization_session_id IS NOT NULL THEN
+        UPDATE upstream_oauth_authorization_sessions
+        SET user_session_id = NEW.user_session_id
+        WHERE upstream_oauth_authorization_sessions.upstream_oauth_authorization_session_id
+               = NEW.upstream_oauth_authorization_session_id
+          AND upstream_oauth_authorization_sessions.user_session_id IS NULL;
+    END IF;
+    RETURN NEW;
+END;
+$$ LANGUAGE plpgsql;
+
+-- Create the trigger
+CREATE TRIGGER upstream_oauth_authorization_sessions_insert_trigger
+  AFTER INSERT ON user_session_authentications
+  FOR EACH ROW
+  EXECUTE FUNCTION upstream_oauth_authorization_sessions_insert_trigger();

crates/storage-pg/migrations/20260121121150_upstream_oauth_track_user_session_backfill.sql

@@ -0,0 +1,13 @@
+-- Copyright 2026 Element Creations Ltd.
+--
+-- SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
+-- Please see LICENSE files in the repository root for full details.
+
+-- Backfill the upstream_oauth_authorization_sessions.user_session_id column
+-- based on session authentications
+UPDATE upstream_oauth_authorization_sessions
+SET user_session_id = user_session_authentications.user_session_id
+FROM user_session_authentications
+WHERE upstream_oauth_authorization_sessions.user_session_id IS NULL
+  AND upstream_oauth_authorization_sessions.upstream_oauth_authorization_session_id
+             = user_session_authentications.upstream_oauth_authorization_session_id;

crates/storage-pg/src/iden.rs

@@ -1,3 +1,4 @@
+// Copyright 2025, 2026 Element Creations Ltd.
 // Copyright 2024, 2025 New Vector Ltd.
 // Copyright 2023, 2024 The Matrix.org Foundation C.I.C.
 //
@@ -18,18 +19,6 @@ pub enum UserSessions {
     LastActiveIp,
 }
 
-#[derive(sea_query::Iden)]
-#[expect(dead_code)]
-pub enum UserSessionAuthentications {
-    Table,
-    UserSessionAuthenticationId,
-    UserSessionId,
-    CreatedAt,
-    UserPasswordId,
-    #[iden = "upstream_oauth_authorization_session_id"]
-    UpstreamOAuthAuthorizationSessionId,
-}
-
 #[derive(sea_query::Iden)]
 pub enum Users {
     Table,
@@ -204,6 +193,7 @@ pub enum UpstreamOAuthAuthorizationSessions {
     CompletedAt,
     ConsumedAt,
     UnlinkedAt,
+    UserSessionId,
 }
 
 #[derive(sea_query::Iden)]

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

@@ -1,3 +1,4 @@
+// Copyright 2025, 2026 Element Creations Ltd.
 // Copyright 2024, 2025 New Vector Ltd.
 // Copyright 2022-2024 The Matrix.org Foundation C.I.C.
 //
@@ -27,7 +28,7 @@ use uuid::Uuid;
 use crate::{
     DatabaseError, DatabaseInconsistencyError,
     filter::StatementExt,
-    iden::{UpstreamOAuthAuthorizationSessions, UserSessionAuthentications, UserSessions, Users},
+    iden::{UpstreamOAuthAuthorizationSessions, UserSessions, Users},
     pagination::QueryBuilderExt,
     tracing::ExecuteExt,
 };
@@ -154,26 +155,14 @@ impl crate::filter::Filter for BrowserSessionFilter<'_> {
             .add_option(self.last_active_before().map(|last_active_before| {
                 Expr::col((UserSessions::Table, UserSessions::LastActiveAt)).lt(last_active_before)
             }))
-            .add_option(self.authenticated_by_upstream_sessions().map(|filter| {
+            .add_option(self.linked_to_upstream_sessions().map(|filter| {
-                // For filtering by upstream sessions, we need to hop over the
-                // `user_session_authentications` table
-                let join_expr = Expr::col((
-                    UserSessionAuthentications::Table,
-                    UserSessionAuthentications::UpstreamOAuthAuthorizationSessionId,
-                ))
-                .eq(Expr::col((
-                    UpstreamOAuthAuthorizationSessions::Table,
-                    UpstreamOAuthAuthorizationSessions::UpstreamOAuthAuthorizationSessionId,
-                )));
-
                 Expr::col((UserSessions::Table, UserSessions::UserSessionId)).in_subquery(
                     Query::select()
                         .expr(Expr::col((
-                            UserSessionAuthentications::Table,
+                            UpstreamOAuthAuthorizationSessions::Table,
-                            UserSessionAuthentications::UserSessionId,
+                            UpstreamOAuthAuthorizationSessions::UserSessionId,
                         )))
-                        .from(UserSessionAuthentications::Table)
+                        .from(UpstreamOAuthAuthorizationSessions::Table)
-                        .inner_join(UpstreamOAuthAuthorizationSessions::Table, join_expr)
                         .apply_filter(filter)
                         .take(),
                 )

crates/storage-pg/src/user/tests.rs

@@ -1,3 +1,4 @@
+// Copyright 2025, 2026 Element Creations Ltd.
 // Copyright 2024, 2025 New Vector Ltd.
 // Copyright 2023, 2024 The Matrix.org Foundation C.I.C.
 //
@@ -799,8 +800,8 @@ async fn test_user_session(pool: PgPool) {
     // This will match all authorization sessions, which matches exactly that one
     // authorization session
     let upstream_oauth_session_filter = UpstreamOAuthSessionFilter::new();
-    let filter = BrowserSessionFilter::new()
+    let filter =
-        .authenticated_by_upstream_sessions_only(upstream_oauth_session_filter);
+        BrowserSessionFilter::new().linked_to_upstream_sessions_only(upstream_oauth_session_filter);
 
     // Now try to look it up
     let page = repo

crates/storage/src/user/session.rs

@@ -1,3 +1,4 @@
+// Copyright 2025, 2026 Element Creations Ltd.
 // Copyright 2024, 2025 New Vector Ltd.
 // Copyright 2022-2024 The Matrix.org Foundation C.I.C.
 //
@@ -41,7 +42,7 @@ pub struct BrowserSessionFilter<'a> {
     state: Option<BrowserSessionState>,
     last_active_before: Option<DateTime<Utc>>,
     last_active_after: Option<DateTime<Utc>>,
-    authenticated_by_upstream_sessions: Option<UpstreamOAuthSessionFilter<'a>>,
+    linked_to_upstream_sessions: Option<UpstreamOAuthSessionFilter<'a>>,
 }
 
 impl<'a> BrowserSessionFilter<'a> {
@@ -114,21 +115,20 @@ impl<'a> BrowserSessionFilter<'a> {
         self.state
     }
 
-    /// Only return browser sessions authenticated by the given upstream OAuth
+    /// Only return browser sessions linked to the given upstream OAuth sessions
-    /// sessions
     #[must_use]
-    pub fn authenticated_by_upstream_sessions_only(
+    pub fn linked_to_upstream_sessions_only(
         mut self,
         filter: UpstreamOAuthSessionFilter<'a>,
     ) -> Self {
-        self.authenticated_by_upstream_sessions = Some(filter);
+        self.linked_to_upstream_sessions = Some(filter);
         self
     }
 
     /// Get the upstream OAuth session filter
     #[must_use]
-    pub fn authenticated_by_upstream_sessions(&self) -> Option<UpstreamOAuthSessionFilter<'a>> {
+    pub fn linked_to_upstream_sessions(&self) -> Option<UpstreamOAuthSessionFilter<'a>> {
-        self.authenticated_by_upstream_sessions
+        self.linked_to_upstream_sessions
     }
 }
 

crates/tasks/src/lib.rs

@@ -211,16 +211,12 @@ pub async fn init(
             "0 57 * * * *".parse()?,
             mas_storage::queue::CleanupUserEmailAuthenticationsJob,
         )
-        // This job is currently disabled, as it needs a database backfill to
+        .add_schedule(
-        // happen, which will happen in the next release. Some context in
+            "cleanup-upstream-oauth-sessions",
-        // https://github.com/element-hq/matrix-authentication-service/issues/5435
+            // Run this job every hour
-        //
+            "0 58 * * * *".parse()?,
-        //.add_schedule(
+            mas_storage::queue::CleanupUpstreamOAuthSessionsJob,
-        //    "cleanup-upstream-oauth-sessions",
+        )
-        //    // Run this job every hour
-        //    "0 58 * * * *".parse()?,
-        //    mas_storage::queue::CleanupUpstreamOAuthSessionsJob,
-        //)
         .add_schedule(
             "cleanup-upstream-oauth-links",
             // Run this job every hour