mas_storage_pg/user/
email.rs

1// Copyright 2025, 2026 Element Creations Ltd.
2// Copyright 2024, 2025 New Vector Ltd.
3// Copyright 2022-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 async_trait::async_trait;
9use chrono::{DateTime, Utc};
10use mas_data_model::{
11    BrowserSession, Clock, UlidExt as _, UpstreamOAuthAuthorizationSession, User, UserEmail,
12    UserEmailAuthentication, UserEmailAuthenticationCode, UserRegistration,
13};
14use mas_storage::{
15    Page, Pagination,
16    pagination::Node,
17    user::{UserEmailFilter, UserEmailRepository},
18};
19use rand::Rng;
20use sea_query::{Expr, ExprTrait, Func, PostgresQueryBuilder, Query, SimpleExpr, enum_def};
21use sea_query_sqlx::SqlxBinder;
22use sqlx::PgConnection;
23use ulid::Ulid;
24use uuid::Uuid;
25
26use crate::{
27    DatabaseError,
28    filter::{Filter, StatementExt},
29    iden::UserEmails,
30    pagination::QueryBuilderExt,
31    tracing::ExecuteExt,
32};
33
34/// An implementation of [`UserEmailRepository`] for a PostgreSQL connection
35pub struct PgUserEmailRepository<'c> {
36    conn: &'c mut PgConnection,
37}
38
39impl<'c> PgUserEmailRepository<'c> {
40    /// Create a new [`PgUserEmailRepository`] from an active PostgreSQL
41    /// connection
42    pub fn new(conn: &'c mut PgConnection) -> Self {
43        Self { conn }
44    }
45}
46
47#[derive(Debug, Clone, sqlx::FromRow)]
48#[enum_def]
49struct UserEmailLookup {
50    user_email_id: Uuid,
51    user_id: Uuid,
52    email: String,
53    created_at: DateTime<Utc>,
54}
55
56impl Node<Ulid> for UserEmailLookup {
57    fn cursor(&self) -> Ulid {
58        self.user_email_id.into()
59    }
60}
61
62impl From<UserEmailLookup> for UserEmail {
63    fn from(e: UserEmailLookup) -> UserEmail {
64        UserEmail {
65            id: e.user_email_id.into(),
66            user_id: e.user_id.into(),
67            email: e.email,
68            created_at: e.created_at,
69        }
70    }
71}
72
73struct UserEmailAuthenticationLookup {
74    user_email_authentication_id: Uuid,
75    user_session_id: Option<Uuid>,
76    user_registration_id: Option<Uuid>,
77    email: String,
78    created_at: DateTime<Utc>,
79    completed_at: Option<DateTime<Utc>>,
80}
81
82impl From<UserEmailAuthenticationLookup> for UserEmailAuthentication {
83    fn from(value: UserEmailAuthenticationLookup) -> Self {
84        UserEmailAuthentication {
85            id: value.user_email_authentication_id.into(),
86            user_session_id: value.user_session_id.map(Ulid::from),
87            user_registration_id: value.user_registration_id.map(Ulid::from),
88            email: value.email,
89            created_at: value.created_at,
90            completed_at: value.completed_at,
91        }
92    }
93}
94
95struct UserEmailAuthenticationCodeLookup {
96    user_email_authentication_code_id: Uuid,
97    user_email_authentication_id: Uuid,
98    code: String,
99    created_at: DateTime<Utc>,
100    expires_at: DateTime<Utc>,
101}
102
103impl From<UserEmailAuthenticationCodeLookup> for UserEmailAuthenticationCode {
104    fn from(value: UserEmailAuthenticationCodeLookup) -> Self {
105        UserEmailAuthenticationCode {
106            id: value.user_email_authentication_code_id.into(),
107            user_email_authentication_id: value.user_email_authentication_id.into(),
108            code: value.code,
109            created_at: value.created_at,
110            expires_at: value.expires_at,
111        }
112    }
113}
114
115impl Filter for UserEmailFilter<'_> {
116    fn generate_condition(&self, _has_joins: bool) -> impl sea_query::IntoCondition {
117        sea_query::Condition::all()
118            .add_option(self.user().map(|user| {
119                Expr::col((UserEmails::Table, UserEmails::UserId)).eq(Uuid::from(user.id))
120            }))
121            .add_option(self.email().map(|email| {
122                SimpleExpr::from(Func::lower(Expr::col((
123                    UserEmails::Table,
124                    UserEmails::Email,
125                ))))
126                .eq(Func::lower(email))
127            }))
128    }
129}
130
131#[async_trait]
132impl UserEmailRepository for PgUserEmailRepository<'_> {
133    type Error = DatabaseError;
134
135    #[tracing::instrument(
136        name = "db.user_email.lookup",
137        skip_all,
138        fields(
139            db.query.text,
140            user_email.id = %id,
141        ),
142    )]
143    async fn lookup(&mut self, id: Ulid) -> Result<Option<UserEmail>, Self::Error> {
144        let res = sqlx::query_as!(
145            UserEmailLookup,
146            r#"
147                SELECT user_email_id
148                     , user_id
149                     , email
150                     , created_at
151                FROM user_emails
152
153                WHERE user_email_id = $1
154            "#,
155            Uuid::from(id),
156        )
157        .traced()
158        .fetch_optional(&mut *self.conn)
159        .await?;
160
161        let Some(user_email) = res else {
162            return Ok(None);
163        };
164
165        Ok(Some(user_email.into()))
166    }
167
168    #[tracing::instrument(
169        name = "db.user_email.find",
170        skip_all,
171        fields(
172            db.query.text,
173            %user.id,
174            user_email.email = email,
175        ),
176    )]
177    async fn find(&mut self, user: &User, email: &str) -> Result<Option<UserEmail>, Self::Error> {
178        let res = sqlx::query_as!(
179            UserEmailLookup,
180            r#"
181                SELECT user_email_id
182                     , user_id
183                     , email
184                     , created_at
185                FROM user_emails
186
187                WHERE user_id = $1 AND LOWER(email) = LOWER($2)
188            "#,
189            Uuid::from(user.id),
190            email,
191        )
192        .traced()
193        .fetch_optional(&mut *self.conn)
194        .await?;
195
196        let Some(user_email) = res else {
197            return Ok(None);
198        };
199
200        Ok(Some(user_email.into()))
201    }
202
203    #[tracing::instrument(
204        name = "db.user_email.find_by_email",
205        skip_all,
206        fields(
207            db.query.text,
208            user_email.email = email,
209        ),
210    )]
211    async fn find_by_email(&mut self, email: &str) -> Result<Option<UserEmail>, Self::Error> {
212        let res = sqlx::query_as!(
213            UserEmailLookup,
214            r#"
215                SELECT user_email_id
216                     , user_id
217                     , email
218                     , created_at
219                FROM user_emails
220                WHERE LOWER(email) = LOWER($1)
221            "#,
222            email,
223        )
224        .traced()
225        .fetch_all(&mut *self.conn)
226        .await?;
227
228        if res.len() != 1 {
229            return Ok(None);
230        }
231
232        let Some(user_email) = res.into_iter().next() else {
233            return Ok(None);
234        };
235
236        Ok(Some(user_email.into()))
237    }
238
239    #[tracing::instrument(
240        name = "db.user_email.all",
241        skip_all,
242        fields(
243            db.query.text,
244            %user.id,
245        ),
246    )]
247    async fn all(&mut self, user: &User) -> Result<Vec<UserEmail>, Self::Error> {
248        let res = sqlx::query_as!(
249            UserEmailLookup,
250            r#"
251                SELECT user_email_id
252                     , user_id
253                     , email
254                     , created_at
255                FROM user_emails
256
257                WHERE user_id = $1
258
259                ORDER BY email ASC
260            "#,
261            Uuid::from(user.id),
262        )
263        .traced()
264        .fetch_all(&mut *self.conn)
265        .await?;
266
267        Ok(res.into_iter().map(Into::into).collect())
268    }
269
270    #[tracing::instrument(
271        name = "db.user_email.list",
272        skip_all,
273        fields(
274            db.query.text,
275        ),
276    )]
277    async fn list(
278        &mut self,
279        filter: UserEmailFilter<'_>,
280        pagination: Pagination,
281    ) -> Result<Page<UserEmail>, DatabaseError> {
282        let (sql, arguments) = Query::select()
283            .expr_as(
284                Expr::col((UserEmails::Table, UserEmails::UserEmailId)),
285                UserEmailLookupIden::UserEmailId,
286            )
287            .expr_as(
288                Expr::col((UserEmails::Table, UserEmails::UserId)),
289                UserEmailLookupIden::UserId,
290            )
291            .expr_as(
292                Expr::col((UserEmails::Table, UserEmails::Email)),
293                UserEmailLookupIden::Email,
294            )
295            .expr_as(
296                Expr::col((UserEmails::Table, UserEmails::CreatedAt)),
297                UserEmailLookupIden::CreatedAt,
298            )
299            .from(UserEmails::Table)
300            .apply_filter(filter)
301            .generate_pagination((UserEmails::Table, UserEmails::UserEmailId), pagination)
302            .build_sqlx(PostgresQueryBuilder);
303
304        let edges: Vec<UserEmailLookup> = sqlx::query_as_with(&sql, arguments)
305            .traced()
306            .fetch_all(&mut *self.conn)
307            .await?;
308
309        let page = pagination.process(edges).map(UserEmail::from);
310
311        Ok(page)
312    }
313
314    #[tracing::instrument(
315        name = "db.user_email.count",
316        skip_all,
317        fields(
318            db.query.text,
319        ),
320    )]
321    async fn count(&mut self, filter: UserEmailFilter<'_>) -> Result<usize, Self::Error> {
322        let (sql, arguments) = Query::select()
323            .expr(Expr::col((UserEmails::Table, UserEmails::UserEmailId)).count())
324            .from(UserEmails::Table)
325            .apply_filter(filter)
326            .build_sqlx(PostgresQueryBuilder);
327
328        let count: i64 = sqlx::query_scalar_with(&sql, arguments)
329            .traced()
330            .fetch_one(&mut *self.conn)
331            .await?;
332
333        count
334            .try_into()
335            .map_err(DatabaseError::to_invalid_operation)
336    }
337
338    #[tracing::instrument(
339        name = "db.user_email.add",
340        skip_all,
341        fields(
342            db.query.text,
343            %user.id,
344            user_email.id,
345            user_email.email = email,
346        ),
347    )]
348    async fn add(
349        &mut self,
350        rng: &mut (dyn Rng + Send),
351        clock: &dyn Clock,
352        user: &User,
353        email: String,
354    ) -> Result<UserEmail, Self::Error> {
355        let created_at = clock.now();
356        let id = Ulid::from_datetime_with_rng(created_at, rng);
357        tracing::Span::current().record("user_email.id", tracing::field::display(id));
358
359        sqlx::query!(
360            r#"
361                INSERT INTO user_emails (user_email_id, user_id, email, created_at)
362                VALUES ($1, $2, $3, $4)
363            "#,
364            Uuid::from(id),
365            Uuid::from(user.id),
366            &email,
367            created_at,
368        )
369        .traced()
370        .execute(&mut *self.conn)
371        .await?;
372
373        Ok(UserEmail {
374            id,
375            user_id: user.id,
376            email,
377            created_at,
378        })
379    }
380
381    #[tracing::instrument(
382        name = "db.user_email.remove",
383        skip_all,
384        fields(
385            db.query.text,
386            user.id = %user_email.user_id,
387            %user_email.id,
388            %user_email.email,
389        ),
390    )]
391    async fn remove(&mut self, user_email: UserEmail) -> Result<(), Self::Error> {
392        let res = sqlx::query!(
393            r#"
394                DELETE FROM user_emails
395                WHERE user_email_id = $1
396            "#,
397            Uuid::from(user_email.id),
398        )
399        .traced()
400        .execute(&mut *self.conn)
401        .await?;
402
403        DatabaseError::ensure_affected_rows(&res, 1)?;
404
405        Ok(())
406    }
407
408    #[tracing::instrument(
409        name = "db.user_email.remove_bulk",
410        skip_all,
411        fields(
412            db.query.text,
413        ),
414    )]
415    async fn remove_bulk(&mut self, filter: UserEmailFilter<'_>) -> Result<usize, Self::Error> {
416        let (sql, arguments) = Query::delete()
417            .from_table(UserEmails::Table)
418            .apply_filter(filter)
419            .build_sqlx(PostgresQueryBuilder);
420
421        let res = sqlx::query_with(&sql, arguments)
422            .traced()
423            .execute(&mut *self.conn)
424            .await?;
425
426        Ok(res.rows_affected().try_into().unwrap_or(usize::MAX))
427    }
428
429    #[tracing::instrument(
430        name = "db.user_email.add_authentication_for_session",
431        skip_all,
432        fields(
433            db.query.text,
434            %session.id,
435            user_email_authentication.id,
436            user_email_authentication.email = email,
437        ),
438    )]
439    async fn add_authentication_for_session(
440        &mut self,
441        rng: &mut (dyn Rng + Send),
442        clock: &dyn Clock,
443        email: String,
444        session: &BrowserSession,
445    ) -> Result<UserEmailAuthentication, Self::Error> {
446        let created_at = clock.now();
447        let id = Ulid::from_datetime_with_rng(created_at, rng);
448        tracing::Span::current()
449            .record("user_email_authentication.id", tracing::field::display(id));
450
451        sqlx::query!(
452            r#"
453                INSERT INTO user_email_authentications
454                  ( user_email_authentication_id
455                  , user_session_id
456                  , email
457                  , created_at
458                  )
459                VALUES ($1, $2, $3, $4)
460            "#,
461            Uuid::from(id),
462            Uuid::from(session.id),
463            &email,
464            created_at,
465        )
466        .traced()
467        .execute(&mut *self.conn)
468        .await?;
469
470        Ok(UserEmailAuthentication {
471            id,
472            user_session_id: Some(session.id),
473            user_registration_id: None,
474            email,
475            created_at,
476            completed_at: None,
477        })
478    }
479
480    #[tracing::instrument(
481        name = "db.user_email.add_authentication_for_registration",
482        skip_all,
483        fields(
484            db.query.text,
485            %user_registration.id,
486            user_email_authentication.id,
487            user_email_authentication.email = email,
488        ),
489    )]
490    async fn add_authentication_for_registration(
491        &mut self,
492        rng: &mut (dyn Rng + Send),
493        clock: &dyn Clock,
494        email: String,
495        user_registration: &UserRegistration,
496    ) -> Result<UserEmailAuthentication, Self::Error> {
497        let created_at = clock.now();
498        let id = Ulid::from_datetime_with_rng(created_at, rng);
499        tracing::Span::current()
500            .record("user_email_authentication.id", tracing::field::display(id));
501
502        sqlx::query!(
503            r#"
504                INSERT INTO user_email_authentications
505                  ( user_email_authentication_id
506                  , user_registration_id
507                  , email
508                  , created_at
509                  )
510                VALUES ($1, $2, $3, $4)
511            "#,
512            Uuid::from(id),
513            Uuid::from(user_registration.id),
514            &email,
515            created_at,
516        )
517        .traced()
518        .execute(&mut *self.conn)
519        .await?;
520
521        Ok(UserEmailAuthentication {
522            id,
523            user_session_id: None,
524            user_registration_id: Some(user_registration.id),
525            email,
526            created_at,
527            completed_at: None,
528        })
529    }
530
531    #[tracing::instrument(
532        name = "db.user_email.add_authentication_code",
533        skip_all,
534        fields(
535            db.query.text,
536            %user_email_authentication.id,
537            %user_email_authentication.email,
538            user_email_authentication_code.id,
539            user_email_authentication_code.code = code,
540        ),
541    )]
542    async fn add_authentication_code(
543        &mut self,
544        rng: &mut (dyn Rng + Send),
545        clock: &dyn Clock,
546        duration: chrono::Duration,
547        user_email_authentication: &UserEmailAuthentication,
548        code: String,
549    ) -> Result<UserEmailAuthenticationCode, Self::Error> {
550        let created_at = clock.now();
551        let expires_at = created_at + duration;
552        let id = Ulid::from_datetime_with_rng(created_at, rng);
553        tracing::Span::current().record(
554            "user_email_authentication_code.id",
555            tracing::field::display(id),
556        );
557
558        sqlx::query!(
559            r#"
560                INSERT INTO user_email_authentication_codes
561                  ( user_email_authentication_code_id
562                  , user_email_authentication_id
563                  , code
564                  , created_at
565                  , expires_at
566                  )
567                VALUES ($1, $2, $3, $4, $5)
568            "#,
569            Uuid::from(id),
570            Uuid::from(user_email_authentication.id),
571            &code,
572            created_at,
573            expires_at,
574        )
575        .traced()
576        .execute(&mut *self.conn)
577        .await?;
578
579        Ok(UserEmailAuthenticationCode {
580            id,
581            user_email_authentication_id: user_email_authentication.id,
582            code,
583            created_at,
584            expires_at,
585        })
586    }
587
588    #[tracing::instrument(
589        name = "db.user_email.lookup_authentication",
590        skip_all,
591        fields(
592            db.query.text,
593            user_email_authentication.id = %id,
594        ),
595    )]
596    async fn lookup_authentication(
597        &mut self,
598        id: Ulid,
599    ) -> Result<Option<UserEmailAuthentication>, Self::Error> {
600        let res = sqlx::query_as!(
601            UserEmailAuthenticationLookup,
602            r#"
603                SELECT user_email_authentication_id
604                     , user_session_id
605                     , user_registration_id
606                     , email
607                     , created_at
608                     , completed_at
609                FROM user_email_authentications
610                WHERE user_email_authentication_id = $1
611            "#,
612            Uuid::from(id),
613        )
614        .traced()
615        .fetch_optional(&mut *self.conn)
616        .await?;
617
618        Ok(res.map(UserEmailAuthentication::from))
619    }
620
621    #[tracing::instrument(
622        name = "db.user_email.find_authentication_by_code",
623        skip_all,
624        fields(
625            db.query.text,
626            %authentication.id,
627            user_email_authentication_code.code = code,
628        ),
629    )]
630    async fn find_authentication_code(
631        &mut self,
632        authentication: &UserEmailAuthentication,
633        code: &str,
634    ) -> Result<Option<UserEmailAuthenticationCode>, Self::Error> {
635        let res = sqlx::query_as!(
636            UserEmailAuthenticationCodeLookup,
637            r#"
638                SELECT user_email_authentication_code_id
639                     , user_email_authentication_id
640                     , code
641                     , created_at
642                     , expires_at
643                FROM user_email_authentication_codes
644                WHERE user_email_authentication_id = $1
645                  AND code = $2
646            "#,
647            Uuid::from(authentication.id),
648            code,
649        )
650        .traced()
651        .fetch_optional(&mut *self.conn)
652        .await?;
653
654        Ok(res.map(UserEmailAuthenticationCode::from))
655    }
656
657    #[tracing::instrument(
658        name = "db.user_email.complete_email_authentication_with_code",
659        skip_all,
660        fields(
661            db.query.text,
662            %user_email_authentication.id,
663            %user_email_authentication.email,
664            %user_email_authentication_code.id,
665            %user_email_authentication_code.code,
666        ),
667    )]
668    async fn complete_authentication_with_code(
669        &mut self,
670        clock: &dyn Clock,
671        mut user_email_authentication: UserEmailAuthentication,
672        user_email_authentication_code: &UserEmailAuthenticationCode,
673    ) -> Result<UserEmailAuthentication, Self::Error> {
674        // We technically don't use the authentication code here (other than
675        // recording it in the span), but this is to make sure the caller has
676        // fetched one before calling this
677        let completed_at = clock.now();
678
679        // We'll assume the caller has checked that completed_at is None, so in case
680        // they haven't, the update will not affect any rows, which will raise
681        // an error
682        let res = sqlx::query!(
683            r#"
684                UPDATE user_email_authentications
685                SET completed_at = $2
686                WHERE user_email_authentication_id = $1
687                  AND completed_at IS NULL
688            "#,
689            Uuid::from(user_email_authentication.id),
690            completed_at,
691        )
692        .traced()
693        .execute(&mut *self.conn)
694        .await?;
695
696        DatabaseError::ensure_affected_rows(&res, 1)?;
697
698        user_email_authentication.completed_at = Some(completed_at);
699        Ok(user_email_authentication)
700    }
701
702    #[tracing::instrument(
703        name = "db.user_email.complete_email_authentication_with_upstream",
704        skip_all,
705        fields(
706            db.query.text,
707            %user_email_authentication.id,
708            %user_email_authentication.email,
709            %upstream_oauth_authorization_session.id,
710        ),
711    )]
712    async fn complete_authentication_with_upstream(
713        &mut self,
714        clock: &dyn Clock,
715        mut user_email_authentication: UserEmailAuthentication,
716        upstream_oauth_authorization_session: &UpstreamOAuthAuthorizationSession,
717    ) -> Result<UserEmailAuthentication, Self::Error> {
718        // We technically don't use the upstream_oauth_authorization_session here (other
719        // than recording it in the span), but this is to make sure the caller
720        // has fetched one before calling this
721        let completed_at = clock.now();
722
723        // We'll assume the caller has checked that completed_at is None, so in case
724        // they haven't, the update will not affect any rows, which will raise
725        // an error
726        let res = sqlx::query!(
727            r#"
728                UPDATE user_email_authentications
729                SET completed_at = $2
730                WHERE user_email_authentication_id = $1
731                  AND completed_at IS NULL
732            "#,
733            Uuid::from(user_email_authentication.id),
734            completed_at,
735        )
736        .traced()
737        .execute(&mut *self.conn)
738        .await?;
739
740        DatabaseError::ensure_affected_rows(&res, 1)?;
741
742        user_email_authentication.completed_at = Some(completed_at);
743        Ok(user_email_authentication)
744    }
745
746    #[tracing::instrument(
747        name = "db.user_email.cleanup_authentications",
748        skip_all,
749        fields(
750            db.query.text,
751            since = since.map(tracing::field::display),
752            until = %until,
753            limit = limit,
754        ),
755    )]
756    async fn cleanup_authentications(
757        &mut self,
758        since: Option<Ulid>,
759        until: Ulid,
760        limit: usize,
761    ) -> Result<(usize, Option<Ulid>), Self::Error> {
762        // Use ULID cursor-based pagination. Since ULIDs contain a timestamp,
763        // we can efficiently delete old authentications without needing an index.
764        // `MAX(uuid)` isn't a thing in Postgres, so we aggregate on the client side.
765        let res = sqlx::query_scalar!(
766            r#"
767                WITH
768                  to_delete AS (
769                    SELECT user_email_authentication_id
770                    FROM user_email_authentications
771                    WHERE ($1::uuid IS NULL OR user_email_authentication_id > $1)
772                      AND user_email_authentication_id <= $2
773                    ORDER BY user_email_authentication_id
774                    LIMIT $3
775                  ),
776                  deleted_codes AS (
777                    DELETE FROM user_email_authentication_codes
778                    USING to_delete
779                    WHERE user_email_authentication_codes.user_email_authentication_id = to_delete.user_email_authentication_id
780                    RETURNING user_email_authentication_codes.user_email_authentication_code_id
781                  )
782                DELETE FROM user_email_authentications
783                USING to_delete
784                WHERE user_email_authentications.user_email_authentication_id = to_delete.user_email_authentication_id
785                RETURNING user_email_authentications.user_email_authentication_id
786            "#,
787            since.map(Uuid::from),
788            Uuid::from(until),
789            i64::try_from(limit).unwrap_or(i64::MAX)
790        )
791        .traced()
792        .fetch_all(&mut *self.conn)
793        .await?;
794
795        let count = res.len();
796        let max_id = res.into_iter().max();
797
798        Ok((count, max_id.map(Ulid::from)))
799    }
800}
mas_storage_pg/user/email.rs

mas_storage_pg/user/
email.rs
