Skip to main content

mas_storage_pg/
app_session.rs

1// Copyright 2025, 2026 Element Creations Ltd.
2// Copyright 2024, 2025 New Vector Ltd.
3// Copyright 2023, 2024 The Matrix.org Foundation C.I.C.
4//
5// SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
6// Please see LICENSE files in the repository root for full details.
7
8//! A module containing PostgreSQL implementation of repositories for sessions
9
10use async_trait::async_trait;
11use mas_data_model::{
12    Clock, CompatSession, CompatSessionState, Device, Session, SessionState, User,
13};
14use mas_storage::{
15    Page, Pagination,
16    app_session::{AppSession, AppSessionFilter, AppSessionRepository, AppSessionState},
17    compat::CompatSessionFilter,
18    oauth2::OAuth2SessionFilter,
19};
20use oauth2_types::scope::{Scope, ScopeToken};
21use opentelemetry_semantic_conventions::trace::DB_QUERY_TEXT;
22use sea_query::{
23    Alias, ColumnRef, CommonTableExpression, Expr, PostgresQueryBuilder, Query, UnionType,
24};
25use sea_query_sqlx::SqlxBinder;
26use sqlx::PgConnection;
27use tracing::Instrument;
28use ulid::Ulid;
29use uuid::Uuid;
30
31use crate::{
32    DatabaseError, ExecuteExt,
33    errors::DatabaseInconsistencyError,
34    filter::StatementExt,
35    iden::{CompatSessions, OAuth2Sessions},
36    pagination::QueryBuilderExt,
37};
38
39/// An implementation of [`AppSessionRepository`] for a PostgreSQL connection
40pub struct PgAppSessionRepository<'c> {
41    conn: &'c mut PgConnection,
42}
43
44impl<'c> PgAppSessionRepository<'c> {
45    /// Create a new [`PgAppSessionRepository`] from an active PostgreSQL
46    /// connection
47    pub fn new(conn: &'c mut PgConnection) -> Self {
48        Self { conn }
49    }
50}
51
52mod priv_ {
53    // The enum_def macro generates a public enum, which we don't want, because it
54    // triggers the missing docs warning
55
56    use std::net::IpAddr;
57
58    use chrono::{DateTime, Utc};
59    use mas_storage::pagination::Node;
60    use sea_query::enum_def;
61    use ulid::Ulid;
62    use uuid::Uuid;
63
64    #[derive(sqlx::FromRow)]
65    #[enum_def]
66    pub(super) struct AppSessionLookup {
67        pub(super) cursor: Uuid,
68        pub(super) compat_session_id: Option<Uuid>,
69        pub(super) oauth2_session_id: Option<Uuid>,
70        pub(super) oauth2_client_id: Option<Uuid>,
71        pub(super) user_session_id: Option<Uuid>,
72        pub(super) user_id: Option<Uuid>,
73        pub(super) scope_list: Option<Vec<String>>,
74        pub(super) device_id: Option<String>,
75        pub(super) human_name: Option<String>,
76        pub(super) created_at: DateTime<Utc>,
77        pub(super) finished_at: Option<DateTime<Utc>>,
78        pub(super) is_synapse_admin: Option<bool>,
79        pub(super) user_agent: Option<String>,
80        pub(super) last_active_at: Option<DateTime<Utc>>,
81        pub(super) last_active_ip: Option<IpAddr>,
82    }
83
84    impl Node<Ulid> for AppSessionLookup {
85        fn cursor(&self) -> Ulid {
86            self.cursor.into()
87        }
88    }
89}
90
91use priv_::{AppSessionLookup, AppSessionLookupIden};
92
93impl TryFrom<AppSessionLookup> for AppSession {
94    type Error = DatabaseError;
95
96    fn try_from(value: AppSessionLookup) -> Result<Self, Self::Error> {
97        // This is annoying to do, but we have to match on all the fields to determine
98        // whether it's a compat session or an oauth2 session
99        let AppSessionLookup {
100            cursor,
101            compat_session_id,
102            oauth2_session_id,
103            oauth2_client_id,
104            user_session_id,
105            user_id,
106            scope_list,
107            device_id,
108            human_name,
109            created_at,
110            finished_at,
111            is_synapse_admin,
112            user_agent,
113            last_active_at,
114            last_active_ip,
115        } = value;
116
117        let user_session_id = user_session_id.map(Ulid::from);
118
119        match (
120            compat_session_id,
121            oauth2_session_id,
122            oauth2_client_id,
123            user_id,
124            scope_list,
125            device_id,
126            is_synapse_admin,
127        ) {
128            (
129                Some(compat_session_id),
130                None,
131                None,
132                Some(user_id),
133                None,
134                device_id_opt,
135                Some(is_synapse_admin),
136            ) => {
137                let id = compat_session_id.into();
138                let device = device_id_opt
139                    .map(Device::try_from)
140                    .transpose()
141                    .map_err(|e| {
142                        DatabaseInconsistencyError::on("compat_sessions")
143                            .column("device_id")
144                            .row(id)
145                            .source(e)
146                    })?;
147
148                let state = match finished_at {
149                    None => CompatSessionState::Valid,
150                    Some(finished_at) => CompatSessionState::Finished { finished_at },
151                };
152
153                let session = CompatSession {
154                    id,
155                    state,
156                    user_id: user_id.into(),
157                    device,
158                    human_name,
159                    user_session_id,
160                    created_at,
161                    is_synapse_admin,
162                    user_agent,
163                    last_active_at,
164                    last_active_ip,
165                };
166
167                Ok(AppSession::Compat(Box::new(session)))
168            }
169
170            (
171                None,
172                Some(oauth2_session_id),
173                Some(oauth2_client_id),
174                user_id,
175                Some(scope_list),
176                None,
177                None,
178            ) => {
179                let id = oauth2_session_id.into();
180                let scope: Result<Scope, _> =
181                    scope_list.iter().map(|s| s.parse::<ScopeToken>()).collect();
182                let scope = scope.map_err(|e| {
183                    DatabaseInconsistencyError::on("oauth2_sessions")
184                        .column("scope")
185                        .row(id)
186                        .source(e)
187                })?;
188
189                let state = match value.finished_at {
190                    None => SessionState::Valid,
191                    Some(finished_at) => SessionState::Finished { finished_at },
192                };
193
194                let session = Session {
195                    id,
196                    state,
197                    created_at,
198                    client_id: oauth2_client_id.into(),
199                    user_id: user_id.map(Ulid::from),
200                    user_session_id,
201                    scope,
202                    user_agent,
203                    last_active_at,
204                    last_active_ip,
205                    human_name,
206                };
207
208                Ok(AppSession::OAuth2(Box::new(session)))
209            }
210
211            _ => Err(DatabaseInconsistencyError::on("sessions")
212                .row(cursor.into())
213                .into()),
214        }
215    }
216}
217
218/// Split a [`AppSessionFilter`] into two separate filters: a
219/// [`CompatSessionFilter`] and an [`OAuth2SessionFilter`].
220fn split_filter(
221    filter: AppSessionFilter<'_>,
222) -> (CompatSessionFilter<'_>, OAuth2SessionFilter<'_>) {
223    let mut compat_filter = CompatSessionFilter::new();
224    let mut oauth2_filter = OAuth2SessionFilter::new();
225
226    if let Some(user) = filter.user() {
227        compat_filter = compat_filter.for_user(user);
228        oauth2_filter = oauth2_filter.for_user(user);
229    }
230
231    match filter.state() {
232        Some(AppSessionState::Active) => {
233            compat_filter = compat_filter.active_only();
234            oauth2_filter = oauth2_filter.active_only();
235        }
236        Some(AppSessionState::Finished) => {
237            compat_filter = compat_filter.finished_only();
238            oauth2_filter = oauth2_filter.finished_only();
239        }
240        None => {}
241    }
242
243    if let Some(device) = filter.device() {
244        compat_filter = compat_filter.for_device(device);
245        oauth2_filter = oauth2_filter.for_device(device);
246    }
247
248    if let Some(browser_session) = filter.browser_session() {
249        compat_filter = compat_filter.for_browser_session(browser_session);
250        oauth2_filter = oauth2_filter.for_browser_session(browser_session);
251    }
252
253    if let Some(last_active_before) = filter.last_active_before() {
254        compat_filter = compat_filter.with_last_active_before(last_active_before);
255        oauth2_filter = oauth2_filter.with_last_active_before(last_active_before);
256    }
257
258    if let Some(last_active_after) = filter.last_active_after() {
259        compat_filter = compat_filter.with_last_active_after(last_active_after);
260        oauth2_filter = oauth2_filter.with_last_active_after(last_active_after);
261    }
262
263    if let Some(created_before) = filter.created_before() {
264        compat_filter = compat_filter.with_created_before(created_before);
265        oauth2_filter = oauth2_filter.with_created_before(created_before);
266    }
267
268    if let Some(created_after) = filter.created_after() {
269        compat_filter = compat_filter.with_created_after(created_after);
270        oauth2_filter = oauth2_filter.with_created_after(created_after);
271    }
272
273    (compat_filter, oauth2_filter)
274}
275
276#[async_trait]
277impl AppSessionRepository for PgAppSessionRepository<'_> {
278    type Error = DatabaseError;
279
280    #[tracing::instrument(
281        name = "db.app_session.list",
282        fields(
283            db.query.text,
284        ),
285        skip_all,
286    )]
287    async fn list(
288        &mut self,
289        filter: AppSessionFilter<'_>,
290        pagination: Pagination,
291    ) -> Result<Page<AppSession>, Self::Error> {
292        let (compat_filter, oauth2_filter) = split_filter(filter);
293
294        let mut oauth2_session_select = Query::select()
295            .expr_as(
296                Expr::col((OAuth2Sessions::Table, OAuth2Sessions::OAuth2SessionId)),
297                AppSessionLookupIden::Cursor,
298            )
299            .expr_as(Expr::cust("NULL"), AppSessionLookupIden::CompatSessionId)
300            .expr_as(
301                Expr::col((OAuth2Sessions::Table, OAuth2Sessions::OAuth2SessionId)),
302                AppSessionLookupIden::Oauth2SessionId,
303            )
304            .expr_as(
305                Expr::col((OAuth2Sessions::Table, OAuth2Sessions::OAuth2ClientId)),
306                AppSessionLookupIden::Oauth2ClientId,
307            )
308            .expr_as(
309                Expr::col((OAuth2Sessions::Table, OAuth2Sessions::UserSessionId)),
310                AppSessionLookupIden::UserSessionId,
311            )
312            .expr_as(
313                Expr::col((OAuth2Sessions::Table, OAuth2Sessions::UserId)),
314                AppSessionLookupIden::UserId,
315            )
316            .expr_as(
317                Expr::col((OAuth2Sessions::Table, OAuth2Sessions::ScopeList)),
318                AppSessionLookupIden::ScopeList,
319            )
320            .expr_as(Expr::cust("NULL"), AppSessionLookupIden::DeviceId)
321            .expr_as(
322                Expr::col((OAuth2Sessions::Table, OAuth2Sessions::HumanName)),
323                AppSessionLookupIden::HumanName,
324            )
325            .expr_as(
326                Expr::col((OAuth2Sessions::Table, OAuth2Sessions::CreatedAt)),
327                AppSessionLookupIden::CreatedAt,
328            )
329            .expr_as(
330                Expr::col((OAuth2Sessions::Table, OAuth2Sessions::FinishedAt)),
331                AppSessionLookupIden::FinishedAt,
332            )
333            .expr_as(Expr::cust("NULL"), AppSessionLookupIden::IsSynapseAdmin)
334            .expr_as(
335                Expr::col((OAuth2Sessions::Table, OAuth2Sessions::UserAgent)),
336                AppSessionLookupIden::UserAgent,
337            )
338            .expr_as(
339                Expr::col((OAuth2Sessions::Table, OAuth2Sessions::LastActiveAt)),
340                AppSessionLookupIden::LastActiveAt,
341            )
342            .expr_as(
343                Expr::col((OAuth2Sessions::Table, OAuth2Sessions::LastActiveIp)),
344                AppSessionLookupIden::LastActiveIp,
345            )
346            .from(OAuth2Sessions::Table)
347            .apply_filter(oauth2_filter)
348            .clone();
349
350        let compat_session_select = Query::select()
351            .expr_as(
352                Expr::col((CompatSessions::Table, CompatSessions::CompatSessionId)),
353                AppSessionLookupIden::Cursor,
354            )
355            .expr_as(
356                Expr::col((CompatSessions::Table, CompatSessions::CompatSessionId)),
357                AppSessionLookupIden::CompatSessionId,
358            )
359            .expr_as(Expr::cust("NULL"), AppSessionLookupIden::Oauth2SessionId)
360            .expr_as(Expr::cust("NULL"), AppSessionLookupIden::Oauth2ClientId)
361            .expr_as(
362                Expr::col((CompatSessions::Table, CompatSessions::UserSessionId)),
363                AppSessionLookupIden::UserSessionId,
364            )
365            .expr_as(
366                Expr::col((CompatSessions::Table, CompatSessions::UserId)),
367                AppSessionLookupIden::UserId,
368            )
369            .expr_as(Expr::cust("NULL"), AppSessionLookupIden::ScopeList)
370            .expr_as(
371                Expr::col((CompatSessions::Table, CompatSessions::DeviceId)),
372                AppSessionLookupIden::DeviceId,
373            )
374            .expr_as(
375                Expr::col((CompatSessions::Table, CompatSessions::HumanName)),
376                AppSessionLookupIden::HumanName,
377            )
378            .expr_as(
379                Expr::col((CompatSessions::Table, CompatSessions::CreatedAt)),
380                AppSessionLookupIden::CreatedAt,
381            )
382            .expr_as(
383                Expr::col((CompatSessions::Table, CompatSessions::FinishedAt)),
384                AppSessionLookupIden::FinishedAt,
385            )
386            .expr_as(
387                Expr::col((CompatSessions::Table, CompatSessions::IsSynapseAdmin)),
388                AppSessionLookupIden::IsSynapseAdmin,
389            )
390            .expr_as(
391                Expr::col((CompatSessions::Table, CompatSessions::UserAgent)),
392                AppSessionLookupIden::UserAgent,
393            )
394            .expr_as(
395                Expr::col((CompatSessions::Table, CompatSessions::LastActiveAt)),
396                AppSessionLookupIden::LastActiveAt,
397            )
398            .expr_as(
399                Expr::col((CompatSessions::Table, CompatSessions::LastActiveIp)),
400                AppSessionLookupIden::LastActiveIp,
401            )
402            .from(CompatSessions::Table)
403            .apply_filter(compat_filter)
404            .clone();
405
406        let common_table_expression = CommonTableExpression::new()
407            .query(
408                oauth2_session_select
409                    .union(UnionType::All, compat_session_select)
410                    .clone(),
411            )
412            .table_name(Alias::new("sessions"))
413            .clone();
414
415        let with_clause = Query::with().cte(common_table_expression).clone();
416
417        let select = Query::select()
418            .column(ColumnRef::Asterisk(None))
419            .from(Alias::new("sessions"))
420            .generate_pagination(AppSessionLookupIden::Cursor, pagination)
421            .clone();
422
423        let (sql, arguments) = with_clause.query(select).build_sqlx(PostgresQueryBuilder);
424
425        let edges: Vec<AppSessionLookup> = sqlx::query_as_with(&sql, arguments)
426            .traced()
427            .fetch_all(&mut *self.conn)
428            .await?;
429
430        let page = pagination.process(edges).try_map(TryFrom::try_from)?;
431
432        Ok(page)
433    }
434
435    #[tracing::instrument(
436        name = "db.app_session.count",
437        fields(
438            db.query.text,
439        ),
440        skip_all,
441    )]
442    async fn count(&mut self, filter: AppSessionFilter<'_>) -> Result<usize, Self::Error> {
443        let (compat_filter, oauth2_filter) = split_filter(filter);
444        let mut oauth2_session_select = Query::select()
445            .expr(Expr::cust("1"))
446            .from(OAuth2Sessions::Table)
447            .apply_filter(oauth2_filter)
448            .clone();
449
450        let compat_session_select = Query::select()
451            .expr(Expr::cust("1"))
452            .from(CompatSessions::Table)
453            .apply_filter(compat_filter)
454            .clone();
455
456        let common_table_expression = CommonTableExpression::new()
457            .query(
458                oauth2_session_select
459                    .union(UnionType::All, compat_session_select)
460                    .clone(),
461            )
462            .table_name(Alias::new("sessions"))
463            .clone();
464
465        let with_clause = Query::with().cte(common_table_expression).clone();
466
467        let select = Query::select()
468            .expr(Expr::cust("COUNT(*)"))
469            .from(Alias::new("sessions"))
470            .clone();
471
472        let (sql, arguments) = with_clause.query(select).build_sqlx(PostgresQueryBuilder);
473
474        let count: i64 = sqlx::query_scalar_with(&sql, arguments)
475            .traced()
476            .fetch_one(&mut *self.conn)
477            .await?;
478
479        count
480            .try_into()
481            .map_err(DatabaseError::to_invalid_operation)
482    }
483
484    #[tracing::instrument(
485        name = "db.app_session.finish_sessions_to_replace_device",
486        fields(
487            db.query.text,
488            %user.id,
489            %device_id = device.as_str()
490        ),
491        skip_all,
492    )]
493    async fn finish_sessions_to_replace_device(
494        &mut self,
495        clock: &dyn Clock,
496        user: &User,
497        device: &Device,
498    ) -> Result<bool, Self::Error> {
499        let mut affected = false;
500        // TODO need to invoke this from all the oauth2 login sites
501        let span = tracing::info_span!(
502            "db.app_session.finish_sessions_to_replace_device.compat_sessions",
503            { DB_QUERY_TEXT } = tracing::field::Empty,
504        );
505        let finished_at = clock.now();
506        let compat_affected = sqlx::query!(
507            "
508                UPDATE compat_sessions SET finished_at = $3 WHERE user_id = $1 AND device_id = $2 AND finished_at IS NULL
509            ",
510            Uuid::from(user.id),
511            device.as_str(),
512            finished_at
513        )
514        .record(&span)
515        .execute(&mut *self.conn)
516        .instrument(span)
517        .await?
518        .rows_affected();
519        affected |= compat_affected > 0;
520
521        if let Ok([stable_device_as_scope_token, unstable_device_as_scope_token]) =
522            device.to_scope_token()
523        {
524            let span = tracing::info_span!(
525                "db.app_session.finish_sessions_to_replace_device.oauth2_sessions",
526                { DB_QUERY_TEXT } = tracing::field::Empty,
527            );
528            let oauth2_affected = sqlx::query!(
529                "
530                    UPDATE oauth2_sessions
531                    SET finished_at = $4
532                    WHERE user_id = $1
533                      AND ($2 = ANY(scope_list) OR $3 = ANY(scope_list))
534                      AND finished_at IS NULL
535                ",
536                Uuid::from(user.id),
537                stable_device_as_scope_token.as_str(),
538                unstable_device_as_scope_token.as_str(),
539                finished_at
540            )
541            .record(&span)
542            .execute(&mut *self.conn)
543            .instrument(span)
544            .await?
545            .rows_affected();
546            affected |= oauth2_affected > 0;
547        }
548
549        Ok(affected)
550    }
551}
552
553#[cfg(test)]
554mod tests {
555    use chrono::Duration;
556    use mas_data_model::{Clock, Device, clock::MockClock};
557    use mas_storage::{
558        Pagination, RepositoryAccess,
559        app_session::{AppSession, AppSessionFilter},
560        oauth2::OAuth2SessionRepository,
561    };
562    use oauth2_types::{
563        requests::GrantType,
564        scope::{OPENID, Scope},
565    };
566    use rand::SeedableRng;
567    use rand_chacha::ChaChaRng;
568    use sqlx::PgPool;
569
570    use crate::PgRepository;
571
572    #[sqlx::test(migrator = "crate::MIGRATOR")]
573    async fn test_app_repo(pool: PgPool) {
574        let mut rng = ChaChaRng::seed_from_u64(42);
575        let clock = MockClock::default();
576        let mut repo = PgRepository::from_pool(&pool).await.unwrap();
577
578        // Create a user
579        let user = repo
580            .user()
581            .add(&mut rng, &clock, "john".to_owned())
582            .await
583            .unwrap();
584
585        let all = AppSessionFilter::new().for_user(&user);
586        let active = all.active_only();
587        let finished = all.finished_only();
588        let pagination = Pagination::first(10);
589
590        assert_eq!(repo.app_session().count(all).await.unwrap(), 0);
591        assert_eq!(repo.app_session().count(active).await.unwrap(), 0);
592        assert_eq!(repo.app_session().count(finished).await.unwrap(), 0);
593
594        let full_list = repo.app_session().list(all, pagination).await.unwrap();
595        assert_eq!(full_list.edges, []);
596        let active_list = repo.app_session().list(active, pagination).await.unwrap();
597        assert_eq!(active_list.edges, []);
598        let finished_list = repo.app_session().list(finished, pagination).await.unwrap();
599        assert_eq!(finished_list.edges, []);
600
601        // Start a compat session for that user
602        let device = Device::generate(&mut rng);
603        let compat_session = repo
604            .compat_session()
605            .add(&mut rng, &clock, &user, device.clone(), None, false, None)
606            .await
607            .unwrap();
608
609        assert_eq!(repo.app_session().count(all).await.unwrap(), 1);
610        assert_eq!(repo.app_session().count(active).await.unwrap(), 1);
611        assert_eq!(repo.app_session().count(finished).await.unwrap(), 0);
612
613        let full_list = repo.app_session().list(all, pagination).await.unwrap();
614        assert_eq!(full_list.edges.len(), 1);
615        assert_eq!(
616            full_list.edges[0].node,
617            AppSession::Compat(Box::new(compat_session.clone()))
618        );
619        let active_list = repo.app_session().list(active, pagination).await.unwrap();
620        assert_eq!(active_list.edges.len(), 1);
621        assert_eq!(
622            active_list.edges[0].node,
623            AppSession::Compat(Box::new(compat_session.clone()))
624        );
625        let finished_list = repo.app_session().list(finished, pagination).await.unwrap();
626        assert_eq!(finished_list.edges, []);
627
628        // Finish the session
629        let compat_session = repo
630            .compat_session()
631            .finish(&clock, compat_session)
632            .await
633            .unwrap();
634
635        assert_eq!(repo.app_session().count(all).await.unwrap(), 1);
636        assert_eq!(repo.app_session().count(active).await.unwrap(), 0);
637        assert_eq!(repo.app_session().count(finished).await.unwrap(), 1);
638
639        let full_list = repo.app_session().list(all, pagination).await.unwrap();
640        assert_eq!(full_list.edges.len(), 1);
641        assert_eq!(
642            full_list.edges[0].node,
643            AppSession::Compat(Box::new(compat_session.clone()))
644        );
645        let active_list = repo.app_session().list(active, pagination).await.unwrap();
646        assert_eq!(active_list.edges, []);
647        let finished_list = repo.app_session().list(finished, pagination).await.unwrap();
648        assert_eq!(finished_list.edges.len(), 1);
649        assert_eq!(
650            finished_list.edges[0].node,
651            AppSession::Compat(Box::new(compat_session.clone()))
652        );
653
654        // Start an OAuth2 session
655        let (client, _) = repo
656            .oauth2_client()
657            .add_or_reuse(
658                &mut rng,
659                &clock,
660                vec!["https://example.com/redirect".parse().unwrap()],
661                None,
662                None,
663                None,
664                vec![GrantType::AuthorizationCode],
665                Some("First client".to_owned()),
666                Some("https://example.com/logo.png".parse().unwrap()),
667                Some("https://example.com/".parse().unwrap()),
668                Some("https://example.com/policy".parse().unwrap()),
669                Some("https://example.com/tos".parse().unwrap()),
670                Some("https://example.com/jwks.json".parse().unwrap()),
671                None,
672                None,
673                None,
674                None,
675                None,
676                Some("https://example.com/login".parse().unwrap()),
677            )
678            .await
679            .unwrap();
680
681        let device2 = Device::generate(&mut rng);
682        let scope: Scope = [OPENID]
683            .into_iter()
684            .chain(device2.to_scope_token().unwrap().into_iter())
685            .collect();
686
687        // We're moving the clock forward by 1 minute between each session to ensure
688        // we're getting consistent ordering in lists.
689        clock.advance(Duration::try_minutes(1).unwrap());
690
691        let oauth_session = repo
692            .oauth2_session()
693            .add(&mut rng, &clock, &client, Some(&user), None, scope)
694            .await
695            .unwrap();
696
697        assert_eq!(repo.app_session().count(all).await.unwrap(), 2);
698        assert_eq!(repo.app_session().count(active).await.unwrap(), 1);
699        assert_eq!(repo.app_session().count(finished).await.unwrap(), 1);
700
701        let full_list = repo.app_session().list(all, pagination).await.unwrap();
702        assert_eq!(full_list.edges.len(), 2);
703        assert_eq!(
704            full_list.edges[0].node,
705            AppSession::Compat(Box::new(compat_session.clone()))
706        );
707        assert_eq!(
708            full_list.edges[1].node,
709            AppSession::OAuth2(Box::new(oauth_session.clone()))
710        );
711
712        let active_list = repo.app_session().list(active, pagination).await.unwrap();
713        assert_eq!(active_list.edges.len(), 1);
714        assert_eq!(
715            active_list.edges[0].node,
716            AppSession::OAuth2(Box::new(oauth_session.clone()))
717        );
718
719        let finished_list = repo.app_session().list(finished, pagination).await.unwrap();
720        assert_eq!(finished_list.edges.len(), 1);
721        assert_eq!(
722            finished_list.edges[0].node,
723            AppSession::Compat(Box::new(compat_session.clone()))
724        );
725
726        // Finish the session
727        let oauth_session = repo
728            .oauth2_session()
729            .finish(&clock, oauth_session)
730            .await
731            .unwrap();
732
733        assert_eq!(repo.app_session().count(all).await.unwrap(), 2);
734        assert_eq!(repo.app_session().count(active).await.unwrap(), 0);
735        assert_eq!(repo.app_session().count(finished).await.unwrap(), 2);
736
737        let full_list = repo.app_session().list(all, pagination).await.unwrap();
738        assert_eq!(full_list.edges.len(), 2);
739        assert_eq!(
740            full_list.edges[0].node,
741            AppSession::Compat(Box::new(compat_session.clone()))
742        );
743        assert_eq!(
744            full_list.edges[1].node,
745            AppSession::OAuth2(Box::new(oauth_session.clone()))
746        );
747
748        let active_list = repo.app_session().list(active, pagination).await.unwrap();
749        assert_eq!(active_list.edges, []);
750
751        let finished_list = repo.app_session().list(finished, pagination).await.unwrap();
752        assert_eq!(finished_list.edges.len(), 2);
753        assert_eq!(
754            finished_list.edges[0].node,
755            AppSession::Compat(Box::new(compat_session.clone()))
756        );
757        assert_eq!(
758            full_list.edges[1].node,
759            AppSession::OAuth2(Box::new(oauth_session.clone()))
760        );
761
762        // Query by device
763        let filter = AppSessionFilter::new().for_device(&device);
764        assert_eq!(repo.app_session().count(filter).await.unwrap(), 1);
765        let list = repo.app_session().list(filter, pagination).await.unwrap();
766        assert_eq!(list.edges.len(), 1);
767        assert_eq!(
768            list.edges[0].node,
769            AppSession::Compat(Box::new(compat_session.clone()))
770        );
771
772        let filter = AppSessionFilter::new().for_device(&device2);
773        assert_eq!(repo.app_session().count(filter).await.unwrap(), 1);
774        let list = repo.app_session().list(filter, pagination).await.unwrap();
775        assert_eq!(list.edges.len(), 1);
776        assert_eq!(
777            list.edges[0].node,
778            AppSession::OAuth2(Box::new(oauth_session.clone()))
779        );
780
781        // Create a second user
782        let user2 = repo
783            .user()
784            .add(&mut rng, &clock, "alice".to_owned())
785            .await
786            .unwrap();
787
788        // If we list/count for this user, we should get nothing
789        let filter = AppSessionFilter::new().for_user(&user2);
790        assert_eq!(repo.app_session().count(filter).await.unwrap(), 0);
791        let list = repo.app_session().list(filter, pagination).await.unwrap();
792        assert_eq!(list.edges, []);
793    }
794
795    /// Test the created-at filters on [`AppSessionFilter`]: they should apply
796    /// to both the compat and the `OAuth2` branches of the union.
797    #[sqlx::test(migrator = "crate::MIGRATOR")]
798    async fn test_list_app_sessions_by_created_at(pool: PgPool) {
799        let mut rng = ChaChaRng::seed_from_u64(42);
800        let clock = MockClock::default();
801        let mut repo = PgRepository::from_pool(&pool).await.unwrap();
802
803        let user = repo
804            .user()
805            .add(&mut rng, &clock, "alice".to_owned())
806            .await
807            .unwrap();
808
809        // Create a compat session
810        let device = Device::generate(&mut rng);
811        let compat_session = repo
812            .compat_session()
813            .add(&mut rng, &clock, &user, device, None, false, None)
814            .await
815            .unwrap();
816        clock.advance(Duration::try_minutes(1).unwrap());
817
818        // Capture a cutoff that sits between the compat session and the
819        // OAuth2 one
820        let cutoff = clock.now();
821        clock.advance(Duration::try_minutes(1).unwrap());
822
823        // Create an OAuth2 session for the same user
824        let (client, _) = repo
825            .oauth2_client()
826            .add_or_reuse(
827                &mut rng,
828                &clock,
829                vec!["https://example.com/redirect".parse().unwrap()],
830                None,
831                None,
832                None,
833                vec![GrantType::AuthorizationCode],
834                Some("Test client".to_owned()),
835                Some("https://example.com/logo.png".parse().unwrap()),
836                Some("https://example.com/".parse().unwrap()),
837                Some("https://example.com/policy".parse().unwrap()),
838                Some("https://example.com/tos".parse().unwrap()),
839                Some("https://example.com/jwks.json".parse().unwrap()),
840                None,
841                None,
842                None,
843                None,
844                None,
845                Some("https://example.com/login".parse().unwrap()),
846            )
847            .await
848            .unwrap();
849        let scope: Scope = [OPENID].into_iter().collect();
850        let oauth_session = repo
851            .oauth2_session()
852            .add(&mut rng, &clock, &client, Some(&user), None, scope)
853            .await
854            .unwrap();
855
856        let pagination = Pagination::first(10);
857
858        // Sessions created before the cutoff: only the compat one
859        let filter = AppSessionFilter::new().with_created_before(cutoff);
860        let list = repo.app_session().list(filter, pagination).await.unwrap();
861        assert_eq!(list.edges.len(), 1);
862        assert_eq!(
863            list.edges[0].node,
864            AppSession::Compat(Box::new(compat_session.clone()))
865        );
866        assert_eq!(repo.app_session().count(filter).await.unwrap(), 1);
867
868        // Sessions created after the cutoff: only the OAuth2 one
869        let filter = AppSessionFilter::new().with_created_after(cutoff);
870        let list = repo.app_session().list(filter, pagination).await.unwrap();
871        assert_eq!(list.edges.len(), 1);
872        assert_eq!(
873            list.edges[0].node,
874            AppSession::OAuth2(Box::new(oauth_session.clone()))
875        );
876        assert_eq!(repo.app_session().count(filter).await.unwrap(), 1);
877    }
878}