Add cleanup job for finished OAuth2 sessions

Implements hard deletion of OAuth2 sessions that have been finished for more than 30 days, including their associated access and refresh tokens.

crates/storage-pg/.sqlx/query-d8f0b02952e786dd4309eac9de04a359aea3a46e5d4e07764cec56ce5d6609c0.json

@@ -0,0 +1,30 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n                WITH\n                    to_delete AS (\n                        SELECT oauth2_session_id, finished_at\n                        FROM oauth2_sessions\n                        WHERE finished_at IS NOT NULL\n                          AND ($1::timestamptz IS NULL OR finished_at >= $1)\n                          AND finished_at < $2\n                        ORDER BY finished_at ASC\n                        LIMIT $3\n                        FOR UPDATE\n                    ),\n                    deleted_refresh_tokens AS (\n                        DELETE FROM oauth2_refresh_tokens USING to_delete\n                        WHERE oauth2_refresh_tokens.oauth2_session_id = to_delete.oauth2_session_id\n                    ),\n                    deleted_access_tokens AS (\n                        DELETE FROM oauth2_access_tokens USING to_delete\n                        WHERE oauth2_access_tokens.oauth2_session_id = to_delete.oauth2_session_id\n                    ),\n                    deleted_sessions AS (\n                        DELETE FROM oauth2_sessions USING to_delete\n                        WHERE oauth2_sessions.oauth2_session_id = to_delete.oauth2_session_id\n                        RETURNING oauth2_sessions.finished_at\n                    )\n                SELECT COUNT(*) as \"count!\", MAX(finished_at) as last_finished_at FROM deleted_sessions\n            ",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "count!",
+        "type_info": "Int8"
+      },
+      {
+        "ordinal": 1,
+        "name": "last_finished_at",
+        "type_info": "Timestamptz"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Timestamptz",
+        "Timestamptz",
+        "Int8"
+      ]
+    },
+    "nullable": [
+      null,
+      null
+    ]
+  },
+  "hash": "d8f0b02952e786dd4309eac9de04a359aea3a46e5d4e07764cec56ce5d6609c0"
+}

crates/storage-pg/migrations/20260122123211_idx_oauth2_sessions_finished_at.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 a partial index on oauth2_sessions.finished_at to help cleaning up
+-- finished sessions
+CREATE INDEX CONCURRENTLY IF NOT EXISTS "oauth2_sessions_finished_at_idx"
+    ON "oauth2_sessions" ("finished_at")
+    WHERE "finished_at" IS NOT NULL;

crates/storage-pg/src/oauth2/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.
 //
@@ -592,4 +593,63 @@ impl OAuth2SessionRepository for PgOAuth2SessionRepository<'_> {
 
         Ok(session)
     }
+
+    #[tracing::instrument(
+        name = "db.oauth2_session.cleanup_finished",
+        skip_all,
+        fields(
+            db.query.text,
+            since = since.map(tracing::field::display),
+            until = %until,
+            limit = limit,
+        ),
+        err,
+    )]
+    async fn cleanup_finished(
+        &mut self,
+        since: Option<DateTime<Utc>>,
+        until: DateTime<Utc>,
+        limit: usize,
+    ) -> Result<(usize, Option<DateTime<Utc>>), Self::Error> {
+        let res = sqlx::query!(
+            r#"
+                WITH
+                    to_delete AS (
+                        SELECT oauth2_session_id, finished_at
+                        FROM oauth2_sessions
+                        WHERE finished_at IS NOT NULL
+                          AND ($1::timestamptz IS NULL OR finished_at >= $1)
+                          AND finished_at < $2
+                        ORDER BY finished_at ASC
+                        LIMIT $3
+                        FOR UPDATE
+                    ),
+                    deleted_refresh_tokens AS (
+                        DELETE FROM oauth2_refresh_tokens USING to_delete
+                        WHERE oauth2_refresh_tokens.oauth2_session_id = to_delete.oauth2_session_id
+                    ),
+                    deleted_access_tokens AS (
+                        DELETE FROM oauth2_access_tokens USING to_delete
+                        WHERE oauth2_access_tokens.oauth2_session_id = to_delete.oauth2_session_id
+                    ),
+                    deleted_sessions AS (
+                        DELETE FROM oauth2_sessions USING to_delete
+                        WHERE oauth2_sessions.oauth2_session_id = to_delete.oauth2_session_id
+                        RETURNING oauth2_sessions.finished_at
+                    )
+                SELECT COUNT(*) as "count!", MAX(finished_at) as last_finished_at FROM deleted_sessions
+            "#,
+            since,
+            until,
+            i64::try_from(limit).unwrap_or(i64::MAX),
+        )
+        .traced()
+        .fetch_one(&mut *self.conn)
+        .await?;
+
+        Ok((
+            res.count.try_into().unwrap_or(usize::MAX),
+            res.last_finished_at,
+        ))
+    }
 }