mas_storage_pg/compat/
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
8use std::net::IpAddr;
9
10use async_trait::async_trait;
11use chrono::{DateTime, Utc};
12use mas_data_model::{
13    BrowserSession, Clock, CompatSession, CompatSessionState, CompatSsoLogin, CompatSsoLoginState,
14    Device, UlidExt as _, User,
15};
16use mas_storage::{
17    Page, Pagination,
18    compat::{CompatSessionFilter, CompatSessionRepository},
19    pagination::Node,
20};
21use rand::Rng;
22use sea_query::{Expr, ExprTrait, PostgresQueryBuilder, Query, enum_def};
23use sea_query_sqlx::SqlxBinder;
24use sqlx::PgConnection;
25use ulid::Ulid;
26use url::Url;
27use uuid::Uuid;
28
29use crate::{
30    DatabaseError, DatabaseInconsistencyError,
31    filter::{Filter, StatementExt, StatementWithJoinsExt},
32    iden::{CompatSessions, CompatSsoLogins, UserSessions},
33    pagination::QueryBuilderExt,
34    tracing::ExecuteExt,
35    ulid_at::{max_ulid_at, min_ulid_at},
36};
37
38/// An implementation of [`CompatSessionRepository`] for a PostgreSQL connection
39pub struct PgCompatSessionRepository<'c> {
40    conn: &'c mut PgConnection,
41}
42
43impl<'c> PgCompatSessionRepository<'c> {
44    /// Create a new [`PgCompatSessionRepository`] from an active PostgreSQL
45    /// connection
46    pub fn new(conn: &'c mut PgConnection) -> Self {
47        Self { conn }
48    }
49}
50
51struct CompatSessionLookup {
52    compat_session_id: Uuid,
53    device_id: Option<String>,
54    human_name: Option<String>,
55    user_id: Uuid,
56    user_session_id: Option<Uuid>,
57    created_at: DateTime<Utc>,
58    finished_at: Option<DateTime<Utc>>,
59    is_synapse_admin: bool,
60    user_agent: Option<String>,
61    last_active_at: Option<DateTime<Utc>>,
62    last_active_ip: Option<IpAddr>,
63}
64
65impl Node<Ulid> for CompatSessionLookup {
66    fn cursor(&self) -> Ulid {
67        self.compat_session_id.into()
68    }
69}
70
71impl From<CompatSessionLookup> for CompatSession {
72    fn from(value: CompatSessionLookup) -> Self {
73        let id = value.compat_session_id.into();
74
75        let state = match value.finished_at {
76            None => CompatSessionState::Valid,
77            Some(finished_at) => CompatSessionState::Finished { finished_at },
78        };
79
80        CompatSession {
81            id,
82            state,
83            user_id: value.user_id.into(),
84            user_session_id: value.user_session_id.map(Ulid::from),
85            device: value.device_id.map(Device::from),
86            human_name: value.human_name,
87            created_at: value.created_at,
88            is_synapse_admin: value.is_synapse_admin,
89            user_agent: value.user_agent,
90            last_active_at: value.last_active_at,
91            last_active_ip: value.last_active_ip,
92        }
93    }
94}
95
96#[derive(sqlx::FromRow)]
97#[enum_def]
98struct CompatSessionAndSsoLoginLookup {
99    compat_session_id: Uuid,
100    device_id: Option<String>,
101    human_name: Option<String>,
102    user_id: Uuid,
103    user_session_id: Option<Uuid>,
104    created_at: DateTime<Utc>,
105    finished_at: Option<DateTime<Utc>>,
106    is_synapse_admin: bool,
107    user_agent: Option<String>,
108    last_active_at: Option<DateTime<Utc>>,
109    last_active_ip: Option<IpAddr>,
110    compat_sso_login_id: Option<Uuid>,
111    compat_sso_login_token: Option<String>,
112    compat_sso_login_redirect_uri: Option<String>,
113    compat_sso_login_created_at: Option<DateTime<Utc>>,
114    compat_sso_login_fulfilled_at: Option<DateTime<Utc>>,
115    compat_sso_login_exchanged_at: Option<DateTime<Utc>>,
116}
117
118impl Node<Ulid> for CompatSessionAndSsoLoginLookup {
119    fn cursor(&self) -> Ulid {
120        self.compat_session_id.into()
121    }
122}
123
124impl TryFrom<CompatSessionAndSsoLoginLookup> for (CompatSession, Option<CompatSsoLogin>) {
125    type Error = DatabaseInconsistencyError;
126
127    fn try_from(value: CompatSessionAndSsoLoginLookup) -> Result<Self, Self::Error> {
128        let id = value.compat_session_id.into();
129
130        let state = match value.finished_at {
131            None => CompatSessionState::Valid,
132            Some(finished_at) => CompatSessionState::Finished { finished_at },
133        };
134
135        let session = CompatSession {
136            id,
137            state,
138            user_id: value.user_id.into(),
139            device: value.device_id.map(Device::from),
140            human_name: value.human_name,
141            user_session_id: value.user_session_id.map(Ulid::from),
142            created_at: value.created_at,
143            is_synapse_admin: value.is_synapse_admin,
144            user_agent: value.user_agent,
145            last_active_at: value.last_active_at,
146            last_active_ip: value.last_active_ip,
147        };
148
149        match (
150            value.compat_sso_login_id,
151            value.compat_sso_login_token,
152            value.compat_sso_login_redirect_uri,
153            value.compat_sso_login_created_at,
154            value.compat_sso_login_fulfilled_at,
155            value.compat_sso_login_exchanged_at,
156        ) {
157            (None, None, None, None, None, None) => Ok((session, None)),
158            (
159                Some(id),
160                Some(login_token),
161                Some(redirect_uri),
162                Some(created_at),
163                fulfilled_at,
164                exchanged_at,
165            ) => {
166                let id = id.into();
167                let redirect_uri = Url::parse(&redirect_uri).map_err(|e| {
168                    DatabaseInconsistencyError::on("compat_sso_logins")
169                        .column("redirect_uri")
170                        .row(id)
171                        .source(e)
172                })?;
173
174                let state = match (fulfilled_at, exchanged_at) {
175                    (Some(fulfilled_at), Some(exchanged_at)) => CompatSsoLoginState::Exchanged {
176                        fulfilled_at,
177                        exchanged_at,
178                        compat_session_id: session.id,
179                    },
180                    _ => return Err(DatabaseInconsistencyError::on("compat_sso_logins").row(id)),
181                };
182
183                let login = CompatSsoLogin {
184                    id,
185                    redirect_uri,
186                    login_token,
187                    created_at,
188                    state,
189                };
190
191                Ok((session, Some(login)))
192            }
193            _ => Err(DatabaseInconsistencyError::on("compat_sso_logins").row(id)),
194        }
195    }
196}
197
198impl Filter for CompatSessionFilter<'_> {
199    fn generate_condition(&self, has_joins: bool) -> impl sea_query::IntoCondition {
200        sea_query::Condition::all()
201            .add_option(self.user().map(|user| {
202                Expr::col((CompatSessions::Table, CompatSessions::UserId)).eq(Uuid::from(user.id))
203            }))
204            .add_option(self.browser_session().map(|browser_session| {
205                Expr::col((CompatSessions::Table, CompatSessions::UserSessionId))
206                    .eq(Uuid::from(browser_session.id))
207            }))
208            .add_option(self.browser_session_filter().map(|browser_session_filter| {
209                Expr::col((CompatSessions::Table, CompatSessions::UserSessionId)).in_subquery(
210                    Query::select()
211                        .expr(Expr::col((
212                            UserSessions::Table,
213                            UserSessions::UserSessionId,
214                        )))
215                        .apply_filter(browser_session_filter)
216                        .from(UserSessions::Table)
217                        .take(),
218                )
219            }))
220            .add_option(self.state().map(|state| {
221                if state.is_active() {
222                    Expr::col((CompatSessions::Table, CompatSessions::FinishedAt)).is_null()
223                } else {
224                    Expr::col((CompatSessions::Table, CompatSessions::FinishedAt)).is_not_null()
225                }
226            }))
227            .add_option(self.auth_type().map(|auth_type| {
228                // In in the SELECT to list sessions, we can rely on the JOINed table, whereas
229                // in other queries we need to do a subquery
230                if has_joins {
231                    if auth_type.is_sso_login() {
232                        Expr::col((CompatSsoLogins::Table, CompatSsoLogins::CompatSsoLoginId))
233                            .is_not_null()
234                    } else {
235                        Expr::col((CompatSsoLogins::Table, CompatSsoLogins::CompatSsoLoginId))
236                            .is_null()
237                    }
238                } else {
239                    // This builds either a:
240                    // `WHERE compat_session_id = ANY(...)`
241                    // or a `WHERE compat_session_id <> ALL(...)`
242                    let compat_sso_logins = Query::select()
243                        .expr(Expr::col((
244                            CompatSsoLogins::Table,
245                            CompatSsoLogins::CompatSessionId,
246                        )))
247                        .from(CompatSsoLogins::Table)
248                        .take();
249
250                    if auth_type.is_sso_login() {
251                        Expr::col((CompatSessions::Table, CompatSessions::CompatSessionId))
252                            .eq(Expr::any(compat_sso_logins))
253                    } else {
254                        Expr::col((CompatSessions::Table, CompatSessions::CompatSessionId))
255                            .ne(Expr::all(compat_sso_logins))
256                    }
257                }
258            }))
259            .add_option(self.last_active_after().map(|last_active_after| {
260                Expr::col((CompatSessions::Table, CompatSessions::LastActiveAt))
261                    .gt(last_active_after)
262            }))
263            .add_option(self.last_active_before().map(|last_active_before| {
264                Expr::col((CompatSessions::Table, CompatSessions::LastActiveAt))
265                    .lt(last_active_before)
266            }))
267            .add_option(self.created_after().map(|created_after| {
268                // ULIDs encode the creation time in their high 48 bits, so we
269                // can use the primary key index to filter on creation time
270                // without touching the `created_at` column.
271                Expr::col((CompatSessions::Table, CompatSessions::CompatSessionId))
272                    .gt(max_ulid_at(created_after))
273            }))
274            .add_option(self.created_before().map(|created_before| {
275                Expr::col((CompatSessions::Table, CompatSessions::CompatSessionId))
276                    .lt(min_ulid_at(created_before))
277            }))
278            .add_option(self.device().map(|device| {
279                Expr::col((CompatSessions::Table, CompatSessions::DeviceId)).eq(device.as_str())
280            }))
281    }
282}
283
284#[async_trait]
285impl CompatSessionRepository for PgCompatSessionRepository<'_> {
286    type Error = DatabaseError;
287
288    #[tracing::instrument(
289        name = "db.compat_session.lookup",
290        skip_all,
291        fields(
292            db.query.text,
293            compat_session.id = %id,
294        ),
295    )]
296    async fn lookup(&mut self, id: Ulid) -> Result<Option<CompatSession>, Self::Error> {
297        let res = sqlx::query_as!(
298            CompatSessionLookup,
299            r#"
300                SELECT compat_session_id
301                     , device_id
302                     , human_name
303                     , user_id
304                     , user_session_id
305                     , created_at
306                     , finished_at
307                     , is_synapse_admin
308                     , user_agent
309                     , last_active_at
310                     , last_active_ip as "last_active_ip: IpAddr"
311                FROM compat_sessions
312                WHERE compat_session_id = $1
313            "#,
314            Uuid::from(id),
315        )
316        .traced()
317        .fetch_optional(&mut *self.conn)
318        .await?;
319
320        let Some(res) = res else { return Ok(None) };
321
322        Ok(Some(res.into()))
323    }
324
325    #[tracing::instrument(
326        name = "db.compat_session.add",
327        skip_all,
328        fields(
329            db.query.text,
330            compat_session.id,
331            %user.id,
332            %user.username,
333            compat_session.device.id = device.as_str(),
334        ),
335    )]
336    async fn add(
337        &mut self,
338        rng: &mut (dyn Rng + Send),
339        clock: &dyn Clock,
340        user: &User,
341        device: Device,
342        browser_session: Option<&BrowserSession>,
343        is_synapse_admin: bool,
344        human_name: Option<String>,
345    ) -> Result<CompatSession, Self::Error> {
346        let created_at = clock.now();
347        let id = Ulid::from_datetime_with_rng(created_at, rng);
348        tracing::Span::current().record("compat_session.id", tracing::field::display(id));
349
350        sqlx::query!(
351            r#"
352                INSERT INTO compat_sessions
353                    (compat_session_id, user_id, device_id,
354                     user_session_id, created_at, is_synapse_admin,
355                     human_name)
356                VALUES ($1, $2, $3, $4, $5, $6, $7)
357            "#,
358            Uuid::from(id),
359            Uuid::from(user.id),
360            device.as_str(),
361            browser_session.map(|s| Uuid::from(s.id)),
362            created_at,
363            is_synapse_admin,
364            human_name.as_deref(),
365        )
366        .traced()
367        .execute(&mut *self.conn)
368        .await?;
369
370        Ok(CompatSession {
371            id,
372            state: CompatSessionState::default(),
373            user_id: user.id,
374            device: Some(device),
375            human_name,
376            user_session_id: browser_session.map(|s| s.id),
377            created_at,
378            is_synapse_admin,
379            user_agent: None,
380            last_active_at: None,
381            last_active_ip: None,
382        })
383    }
384
385    #[tracing::instrument(
386        name = "db.compat_session.finish",
387        skip_all,
388        fields(
389            db.query.text,
390            %compat_session.id,
391            user.id = %compat_session.user_id,
392            compat_session.device.id = compat_session.device.as_ref().map(mas_data_model::Device::as_str),
393        ),
394    )]
395    async fn finish(
396        &mut self,
397        clock: &dyn Clock,
398        compat_session: CompatSession,
399    ) -> Result<CompatSession, Self::Error> {
400        let finished_at = clock.now();
401
402        let res = sqlx::query!(
403            r#"
404                UPDATE compat_sessions cs
405                SET finished_at = $2
406                WHERE compat_session_id = $1
407            "#,
408            Uuid::from(compat_session.id),
409            finished_at,
410        )
411        .traced()
412        .execute(&mut *self.conn)
413        .await?;
414
415        DatabaseError::ensure_affected_rows(&res, 1)?;
416
417        let compat_session = compat_session
418            .finish(finished_at)
419            .map_err(DatabaseError::to_invalid_operation)?;
420
421        Ok(compat_session)
422    }
423
424    #[tracing::instrument(
425        name = "db.compat_session.finish_bulk",
426        skip_all,
427        fields(db.query.text),
428    )]
429    async fn finish_bulk(
430        &mut self,
431        clock: &dyn Clock,
432        filter: CompatSessionFilter<'_>,
433    ) -> Result<usize, Self::Error> {
434        let finished_at = clock.now();
435        let (sql, arguments) = Query::update()
436            .table(CompatSessions::Table)
437            .value(CompatSessions::FinishedAt, finished_at)
438            .apply_filter(filter)
439            .build_sqlx(PostgresQueryBuilder);
440
441        let res = sqlx::query_with(&sql, arguments)
442            .traced()
443            .execute(&mut *self.conn)
444            .await?;
445
446        Ok(res.rows_affected().try_into().unwrap_or(usize::MAX))
447    }
448
449    #[tracing::instrument(
450        name = "db.compat_session.list",
451        skip_all,
452        fields(
453            db.query.text,
454        ),
455    )]
456    async fn list(
457        &mut self,
458        filter: CompatSessionFilter<'_>,
459        pagination: Pagination,
460    ) -> Result<Page<(CompatSession, Option<CompatSsoLogin>)>, Self::Error> {
461        let (sql, arguments) = Query::select()
462            .expr_as(
463                Expr::col((CompatSessions::Table, CompatSessions::CompatSessionId)),
464                CompatSessionAndSsoLoginLookupIden::CompatSessionId,
465            )
466            .expr_as(
467                Expr::col((CompatSessions::Table, CompatSessions::DeviceId)),
468                CompatSessionAndSsoLoginLookupIden::DeviceId,
469            )
470            .expr_as(
471                Expr::col((CompatSessions::Table, CompatSessions::HumanName)),
472                CompatSessionAndSsoLoginLookupIden::HumanName,
473            )
474            .expr_as(
475                Expr::col((CompatSessions::Table, CompatSessions::UserId)),
476                CompatSessionAndSsoLoginLookupIden::UserId,
477            )
478            .expr_as(
479                Expr::col((CompatSessions::Table, CompatSessions::UserSessionId)),
480                CompatSessionAndSsoLoginLookupIden::UserSessionId,
481            )
482            .expr_as(
483                Expr::col((CompatSessions::Table, CompatSessions::CreatedAt)),
484                CompatSessionAndSsoLoginLookupIden::CreatedAt,
485            )
486            .expr_as(
487                Expr::col((CompatSessions::Table, CompatSessions::FinishedAt)),
488                CompatSessionAndSsoLoginLookupIden::FinishedAt,
489            )
490            .expr_as(
491                Expr::col((CompatSessions::Table, CompatSessions::IsSynapseAdmin)),
492                CompatSessionAndSsoLoginLookupIden::IsSynapseAdmin,
493            )
494            .expr_as(
495                Expr::col((CompatSessions::Table, CompatSessions::UserAgent)),
496                CompatSessionAndSsoLoginLookupIden::UserAgent,
497            )
498            .expr_as(
499                Expr::col((CompatSessions::Table, CompatSessions::LastActiveAt)),
500                CompatSessionAndSsoLoginLookupIden::LastActiveAt,
501            )
502            .expr_as(
503                Expr::col((CompatSessions::Table, CompatSessions::LastActiveIp)),
504                CompatSessionAndSsoLoginLookupIden::LastActiveIp,
505            )
506            .expr_as(
507                Expr::col((CompatSsoLogins::Table, CompatSsoLogins::CompatSsoLoginId)),
508                CompatSessionAndSsoLoginLookupIden::CompatSsoLoginId,
509            )
510            .expr_as(
511                Expr::col((CompatSsoLogins::Table, CompatSsoLogins::LoginToken)),
512                CompatSessionAndSsoLoginLookupIden::CompatSsoLoginToken,
513            )
514            .expr_as(
515                Expr::col((CompatSsoLogins::Table, CompatSsoLogins::RedirectUri)),
516                CompatSessionAndSsoLoginLookupIden::CompatSsoLoginRedirectUri,
517            )
518            .expr_as(
519                Expr::col((CompatSsoLogins::Table, CompatSsoLogins::CreatedAt)),
520                CompatSessionAndSsoLoginLookupIden::CompatSsoLoginCreatedAt,
521            )
522            .expr_as(
523                Expr::col((CompatSsoLogins::Table, CompatSsoLogins::FulfilledAt)),
524                CompatSessionAndSsoLoginLookupIden::CompatSsoLoginFulfilledAt,
525            )
526            .expr_as(
527                Expr::col((CompatSsoLogins::Table, CompatSsoLogins::ExchangedAt)),
528                CompatSessionAndSsoLoginLookupIden::CompatSsoLoginExchangedAt,
529            )
530            .from(CompatSessions::Table)
531            .left_join(
532                CompatSsoLogins::Table,
533                Expr::col((CompatSessions::Table, CompatSessions::CompatSessionId))
534                    .equals((CompatSsoLogins::Table, CompatSsoLogins::CompatSessionId)),
535            )
536            .apply_filter_with_joins(filter)
537            .generate_pagination(
538                (CompatSessions::Table, CompatSessions::CompatSessionId),
539                pagination,
540            )
541            .build_sqlx(PostgresQueryBuilder);
542
543        let edges: Vec<CompatSessionAndSsoLoginLookup> = sqlx::query_as_with(&sql, arguments)
544            .traced()
545            .fetch_all(&mut *self.conn)
546            .await?;
547
548        let page = pagination.process(edges).try_map(TryFrom::try_from)?;
549
550        Ok(page)
551    }
552
553    #[tracing::instrument(
554        name = "db.compat_session.count",
555        skip_all,
556        fields(
557            db.query.text,
558        ),
559    )]
560    async fn count(&mut self, filter: CompatSessionFilter<'_>) -> Result<usize, Self::Error> {
561        let (sql, arguments) = sea_query::Query::select()
562            .expr(Expr::col((CompatSessions::Table, CompatSessions::CompatSessionId)).count())
563            .from(CompatSessions::Table)
564            .apply_filter(filter)
565            .build_sqlx(PostgresQueryBuilder);
566
567        let count: i64 = sqlx::query_scalar_with(&sql, arguments)
568            .traced()
569            .fetch_one(&mut *self.conn)
570            .await?;
571
572        count
573            .try_into()
574            .map_err(DatabaseError::to_invalid_operation)
575    }
576
577    #[tracing::instrument(
578        name = "db.compat_session.record_batch_activity",
579        skip_all,
580        fields(
581            db.query.text,
582        ),
583    )]
584    async fn record_batch_activity(
585        &mut self,
586        mut activities: Vec<(Ulid, DateTime<Utc>, Option<IpAddr>)>,
587    ) -> Result<(), Self::Error> {
588        // Sort the activity by ID, so that when batching the updates, Postgres
589        // locks the rows in a stable order, preventing deadlocks
590        activities.sort_unstable();
591        let mut ids = Vec::with_capacity(activities.len());
592        let mut last_activities = Vec::with_capacity(activities.len());
593        let mut ips = Vec::with_capacity(activities.len());
594
595        for (id, last_activity, ip) in activities {
596            ids.push(Uuid::from(id));
597            last_activities.push(last_activity);
598            ips.push(ip);
599        }
600
601        let res = sqlx::query!(
602            r#"
603                UPDATE compat_sessions
604                SET last_active_at = GREATEST(t.last_active_at, compat_sessions.last_active_at)
605                  , last_active_ip = COALESCE(t.last_active_ip, compat_sessions.last_active_ip)
606                FROM (
607                    SELECT *
608                    FROM UNNEST($1::uuid[], $2::timestamptz[], $3::inet[])
609                        AS t(compat_session_id, last_active_at, last_active_ip)
610                ) AS t
611                WHERE compat_sessions.compat_session_id = t.compat_session_id
612            "#,
613            &ids,
614            &last_activities,
615            &ips as &[Option<IpAddr>],
616        )
617        .traced()
618        .execute(&mut *self.conn)
619        .await?;
620
621        DatabaseError::ensure_affected_rows(&res, ids.len().try_into().unwrap_or(u64::MAX))?;
622
623        Ok(())
624    }
625
626    #[tracing::instrument(
627        name = "db.compat_session.record_user_agent",
628        skip_all,
629        fields(
630            db.query.text,
631            %compat_session.id,
632        ),
633    )]
634    async fn record_user_agent(
635        &mut self,
636        mut compat_session: CompatSession,
637        user_agent: String,
638    ) -> Result<CompatSession, Self::Error> {
639        let res = sqlx::query!(
640            r#"
641            UPDATE compat_sessions
642            SET user_agent = $2
643            WHERE compat_session_id = $1
644        "#,
645            Uuid::from(compat_session.id),
646            &*user_agent,
647        )
648        .traced()
649        .execute(&mut *self.conn)
650        .await?;
651
652        compat_session.user_agent = Some(user_agent);
653
654        DatabaseError::ensure_affected_rows(&res, 1)?;
655
656        Ok(compat_session)
657    }
658
659    #[tracing::instrument(
660        name = "repository.compat_session.set_human_name",
661        skip(self),
662        fields(
663            compat_session.id = %compat_session.id,
664            compat_session.human_name = ?human_name,
665        ),
666    )]
667    async fn set_human_name(
668        &mut self,
669        mut compat_session: CompatSession,
670        human_name: Option<String>,
671    ) -> Result<CompatSession, Self::Error> {
672        let res = sqlx::query!(
673            r#"
674            UPDATE compat_sessions
675            SET human_name = $2
676            WHERE compat_session_id = $1
677        "#,
678            Uuid::from(compat_session.id),
679            human_name.as_deref(),
680        )
681        .traced()
682        .execute(&mut *self.conn)
683        .await?;
684
685        compat_session.human_name = human_name;
686
687        DatabaseError::ensure_affected_rows(&res, 1)?;
688
689        Ok(compat_session)
690    }
691
692    #[tracing::instrument(
693        name = "db.compat_session.cleanup_finished",
694        skip_all,
695        fields(
696            db.query.text,
697        ),
698    )]
699    async fn cleanup_finished(
700        &mut self,
701        since: Option<DateTime<Utc>>,
702        until: DateTime<Utc>,
703        limit: usize,
704    ) -> Result<(usize, Option<DateTime<Utc>>), Self::Error> {
705        let res = sqlx::query!(
706            r#"
707                WITH
708                    to_delete AS (
709                        SELECT compat_session_id, finished_at
710                        FROM compat_sessions
711                        WHERE finished_at IS NOT NULL
712                          AND ($1::timestamptz IS NULL OR finished_at >= $1)
713                          AND finished_at < $2
714                        ORDER BY finished_at ASC
715                        LIMIT $3
716                        FOR UPDATE
717                    ),
718
719                    -- Delete refresh tokens first because they reference access tokens
720                    deleted_refresh_tokens AS (
721                        DELETE FROM compat_refresh_tokens
722                        USING to_delete
723                        WHERE compat_refresh_tokens.compat_session_id = to_delete.compat_session_id
724                    ),
725
726                    deleted_access_tokens AS (
727                        DELETE FROM compat_access_tokens
728                        USING to_delete
729                        WHERE compat_access_tokens.compat_session_id = to_delete.compat_session_id
730                    ),
731
732                    deleted_sso_logins AS (
733                        DELETE FROM compat_sso_logins
734                        USING to_delete
735                        WHERE compat_sso_logins.compat_session_id = to_delete.compat_session_id
736                    ),
737
738                    deleted_sessions AS (
739                        DELETE FROM compat_sessions
740                        USING to_delete
741                        WHERE compat_sessions.compat_session_id = to_delete.compat_session_id
742                        RETURNING compat_sessions.finished_at
743                    )
744
745                SELECT
746                    COUNT(*) as "count!",
747                    MAX(finished_at) as last_finished_at
748                FROM deleted_sessions
749            "#,
750            since,
751            until,
752            i64::try_from(limit).unwrap_or(i64::MAX),
753        )
754        .traced()
755        .fetch_one(&mut *self.conn)
756        .await?;
757
758        Ok((
759            res.count.try_into().unwrap_or(usize::MAX),
760            res.last_finished_at,
761        ))
762    }
763
764    #[tracing::instrument(
765        name = "db.compat_session.cleanup_inactive_ips",
766        skip_all,
767        fields(
768            db.query.text,
769            since = since.map(tracing::field::display),
770            threshold = %threshold,
771            limit = limit,
772        ),
773    )]
774    async fn cleanup_inactive_ips(
775        &mut self,
776        since: Option<DateTime<Utc>>,
777        threshold: DateTime<Utc>,
778        limit: usize,
779    ) -> Result<(usize, Option<DateTime<Utc>>), Self::Error> {
780        let res = sqlx::query!(
781            r#"
782                WITH to_update AS (
783                    SELECT compat_session_id, last_active_at
784                    FROM compat_sessions
785                    WHERE last_active_ip IS NOT NULL
786                      AND last_active_at IS NOT NULL
787                      AND ($1::timestamptz IS NULL OR last_active_at >= $1)
788                      AND last_active_at < $2
789                    ORDER BY last_active_at ASC
790                    LIMIT $3
791                    FOR UPDATE
792                ),
793                updated AS (
794                    UPDATE compat_sessions
795                    SET last_active_ip = NULL
796                    FROM to_update
797                    WHERE compat_sessions.compat_session_id = to_update.compat_session_id
798                    RETURNING compat_sessions.last_active_at
799                )
800                SELECT COUNT(*) AS "count!", MAX(last_active_at) AS last_active_at FROM updated
801            "#,
802            since,
803            threshold,
804            i64::try_from(limit).unwrap_or(i64::MAX),
805        )
806        .traced()
807        .fetch_one(&mut *self.conn)
808        .await?;
809
810        Ok((
811            res.count.try_into().unwrap_or(usize::MAX),
812            res.last_active_at,
813        ))
814    }
815}
mas_storage_pg/compat/session.rs

mas_storage_pg/compat/
session.rs
