storage: filter compat sessions by creation time

crates/storage-pg/src/compat/mod.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.
 //
@@ -312,6 +313,73 @@ mod tests {
         assert_eq!(repo.compat_session().count(active).await.unwrap(), 0);
     }
 
+    /// Test the created-at filters on [`CompatSessionFilter`].
+    #[sqlx::test(migrator = "crate::MIGRATOR")]
+    async fn test_list_compat_sessions_by_created_at(pool: PgPool) {
+        let mut rng = ChaChaRng::seed_from_u64(42);
+        let clock = MockClock::default();
+        let mut repo = PgRepository::from_pool(&pool).await.unwrap();
+
+        let user = repo
+            .user()
+            .add(&mut rng, &clock, "alice".to_owned())
+            .await
+            .unwrap();
+
+        // Three sessions created one minute apart, with a cutoff captured
+        // between the second and the third.
+        let device = Device::generate(&mut rng);
+        let session1 = repo
+            .compat_session()
+            .add(&mut rng, &clock, &user, device, None, false, None)
+            .await
+            .unwrap();
+        clock.advance(Duration::try_minutes(1).unwrap());
+
+        let device = Device::generate(&mut rng);
+        let session2 = repo
+            .compat_session()
+            .add(&mut rng, &clock, &user, device, None, false, None)
+            .await
+            .unwrap();
+        clock.advance(Duration::try_minutes(1).unwrap());
+
+        let cutoff = clock.now();
+
+        clock.advance(Duration::try_minutes(1).unwrap());
+        let device = Device::generate(&mut rng);
+        let session3 = repo
+            .compat_session()
+            .add(&mut rng, &clock, &user, device, None, false, None)
+            .await
+            .unwrap();
+
+        let pagination = Pagination::first(10);
+
+        // Sessions created before the cutoff
+        let filter = CompatSessionFilter::new().with_created_before(cutoff);
+        let list = repo
+            .compat_session()
+            .list(filter, pagination)
+            .await
+            .unwrap();
+        assert_eq!(list.edges.len(), 2);
+        assert_eq!(list.edges[0].node.0, session1);
+        assert_eq!(list.edges[1].node.0, session2);
+        assert_eq!(repo.compat_session().count(filter).await.unwrap(), 2);
+
+        // Sessions created after the cutoff
+        let filter = CompatSessionFilter::new().with_created_after(cutoff);
+        let list = repo
+            .compat_session()
+            .list(filter, pagination)
+            .await
+            .unwrap();
+        assert_eq!(list.edges.len(), 1);
+        assert_eq!(list.edges[0].node.0, session3);
+        assert_eq!(repo.compat_session().count(filter).await.unwrap(), 1);
+    }
+
     #[sqlx::test(migrator = "crate::MIGRATOR")]
     async fn test_access_token_repository(pool: PgPool) {
         const FIRST_TOKEN: &str = "first_access_token";

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

@@ -32,6 +32,7 @@ use crate::{
     iden::{CompatSessions, CompatSsoLogins, UserSessions},
     pagination::QueryBuilderExt,
     tracing::ExecuteExt,
+    ulid_at::{max_ulid_at, min_ulid_at},
 };
 
 /// An implementation of [`CompatSessionRepository`] for a PostgreSQL connection
@@ -263,6 +264,17 @@ impl Filter for CompatSessionFilter<'_> {
                 Expr::col((CompatSessions::Table, CompatSessions::LastActiveAt))
                     .lt(last_active_before)
             }))
+            .add_option(self.created_after().map(|created_after| {
+                // ULIDs encode the creation time in their high 48 bits, so we
+                // can use the primary key index to filter on creation time
+                // without touching the `created_at` column.
+                Expr::col((CompatSessions::Table, CompatSessions::CompatSessionId))
+                    .gt(max_ulid_at(created_after))
+            }))
+            .add_option(self.created_before().map(|created_before| {
+                Expr::col((CompatSessions::Table, CompatSessions::CompatSessionId))
+                    .lt(min_ulid_at(created_before))
+            }))
             .add_option(self.device().map(|device| {
                 Expr::col((CompatSessions::Table, CompatSessions::DeviceId)).eq(device.as_str())
             }))

crates/storage/src/compat/session.rs

@@ -66,6 +66,8 @@ pub struct CompatSessionFilter<'a> {
     device: Option<&'a Device>,
     last_active_before: Option<DateTime<Utc>>,
     last_active_after: Option<DateTime<Utc>>,
+    created_before: Option<DateTime<Utc>>,
+    created_after: Option<DateTime<Utc>>,
 }
 
 impl<'a> CompatSessionFilter<'a> {
@@ -160,6 +162,36 @@ impl<'a> CompatSessionFilter<'a> {
         self.last_active_after
     }
 
+    /// Only return sessions created before the given time
+    #[must_use]
+    pub fn with_created_before(mut self, created_before: DateTime<Utc>) -> Self {
+        self.created_before = Some(created_before);
+        self
+    }
+
+    /// Only return sessions created after the given time
+    #[must_use]
+    pub fn with_created_after(mut self, created_after: DateTime<Utc>) -> Self {
+        self.created_after = Some(created_after);
+        self
+    }
+
+    /// Get the created-before filter
+    ///
+    /// Returns [`None`] if no filter was set
+    #[must_use]
+    pub fn created_before(&self) -> Option<DateTime<Utc>> {
+        self.created_before
+    }
+
+    /// Get the created-after filter
+    ///
+    /// Returns [`None`] if no filter was set
+    #[must_use]
+    pub fn created_after(&self) -> Option<DateTime<Utc>> {
+        self.created_after
+    }
+
     /// Only return active compatibility sessions
     #[must_use]
     pub fn active_only(mut self) -> Self {