storage: user and user email repository

crates/cli/src/commands/manage.rs

@@ -20,9 +20,7 @@ use mas_router::UrlBuilder;
 use mas_storage::{
     oauth2::client::{insert_client_from_config, lookup_client, truncate_clients},
     upstream_oauth2::UpstreamOAuthProviderRepository,
-    user::{
+    user::{add_user_password, UserEmailRepository, UserRepository},
-        add_user_password, lookup_user_by_username, lookup_user_email, mark_user_email_as_verified,
-    },
     Clock, Repository,
 };
 use oauth2_types::scope::Scope;
@@ -202,7 +200,9 @@ impl Options {
                 let password_manager = password_manager_from_config(&passwords_config).await?;
 
                 let mut txn = pool.begin().await?;
-                let user = lookup_user_by_username(&mut txn, username)
+                let user = txn
+                    .user()
+                    .find_by_username(username)
                     .await?
                     .context("User not found")?;
 
@@ -232,13 +232,18 @@ impl Options {
                 let pool = database_from_config(&config).await?;
                 let mut txn = pool.begin().await?;
 
-                let user = lookup_user_by_username(&mut txn, username)
+                let user = txn
+                    .user()
+                    .find_by_username(username)
                     .await?
                     .context("User not found")?;
-                let email = lookup_user_email(&mut txn, &user, email)
+
+                let email = txn
+                    .user_email()
+                    .find(&user, email)
                     .await?
                     .context("Email not found")?;
-                let email = mark_user_email_as_verified(&mut txn, &clock, email).await?;
+                let email = txn.user_email().mark_as_verified(&clock, email).await?;
 
                 txn.commit().await?;
                 info!(?email, "Email marked as verified");

crates/data-model/src/users.rs

@@ -22,7 +22,7 @@ pub struct User {
     pub id: Ulid,
     pub username: String,
     pub sub: String,
-    pub primary_email: Option<UserEmail>,
+    pub primary_user_email_id: Option<Ulid>,
 }
 
 impl User {
@@ -32,7 +32,7 @@ impl User {
             id: Ulid::from_datetime_with_source(now.into(), rng),
             username: "john".to_owned(),
             sub: "123-456".to_owned(),
-            primary_email: None,
+            primary_user_email_id: None,
         }]
     }
 }
@@ -89,6 +89,7 @@ impl BrowserSession {
 #[derive(Debug, Clone, PartialEq, Eq, Serialize)]
 pub struct UserEmail {
     pub id: Ulid,
+    pub user_id: Ulid,
     pub email: String,
     pub created_at: DateTime<Utc>,
     pub confirmed_at: Option<DateTime<Utc>>,
@@ -100,12 +101,14 @@ impl UserEmail {
         vec![
             Self {
                 id: Ulid::from_datetime_with_source(now.into(), rng),
+                user_id: Ulid::from_datetime_with_source(now.into(), rng),
                 email: "alice@example.com".to_owned(),
                 created_at: now,
                 confirmed_at: Some(now),
             },
             Self {
                 id: Ulid::from_datetime_with_source(now.into(), rng),
+                user_id: Ulid::from_datetime_with_source(now.into(), rng),
                 email: "bob@example.com".to_owned(),
                 created_at: now,
                 confirmed_at: None,
@@ -124,7 +127,7 @@ pub enum UserEmailVerificationState {
 #[derive(Debug, Clone, PartialEq, Eq, Serialize)]
 pub struct UserEmailVerification {
     pub id: Ulid,
-    pub email: UserEmail,
+    pub user_email_id: Ulid,
     pub code: String,
     pub created_at: DateTime<Utc>,
     pub state: UserEmailVerificationState,
@@ -152,8 +155,8 @@ impl UserEmailVerification {
                     .into_iter()
                     .map(move |email| Self {
                         id: Ulid::from_datetime_with_source(now.into(), &mut rng),
+                        user_email_id: email.id,
                         code: "123456".to_owned(),
-                        email,
                         created_at: now - Duration::minutes(10),
                         state: state.clone(),
                     })

crates/graphql/src/lib.rs

@@ -31,7 +31,8 @@ use async_graphql::{
     Context, Description, EmptyMutation, EmptySubscription, ID,
 };
 use mas_storage::{
-    upstream_oauth2::UpstreamOAuthProviderRepository, Repository, UpstreamOAuthLinkRepository,
+    upstream_oauth2::UpstreamOAuthProviderRepository, user::UserEmailRepository, Repository,
+    UpstreamOAuthLinkRepository,
 };
 use model::CreationEvent;
 use sqlx::PgPool;
@@ -154,8 +155,11 @@ impl RootQuery {
         let Some(session) = session else { return Ok(None) };
         let current_user = session.user;
 
-        let user_email =
+        let user_email = conn
-            mas_storage::user::lookup_user_email_by_id(&mut conn, &current_user, id).await?;
+            .user_email()
+            .lookup(id)
+            .await?
+            .filter(|e| e.user_id == current_user.id);
 
         Ok(user_email.map(UserEmail))
     }

crates/graphql/src/model/upstream_oauth.rs

@@ -15,7 +15,9 @@
 use anyhow::Context as _;
 use async_graphql::{Context, Object, ID};
 use chrono::{DateTime, Utc};
-use mas_storage::{upstream_oauth2::UpstreamOAuthProviderRepository, Repository};
+use mas_storage::{
+    upstream_oauth2::UpstreamOAuthProviderRepository, user::UserRepository, Repository,
+};
 use sqlx::PgPool;
 
 use super::{NodeType, User};
@@ -120,7 +122,10 @@ impl UpstreamOAuth2Link {
             // Fetch on-the-fly
             let database = ctx.data::<PgPool>()?;
             let mut conn = database.acquire().await?;
-            mas_storage::user::lookup_user(&mut conn, *user_id).await?
+            conn.user()
+                .lookup(*user_id)
+                .await?
+                .context("User not found")?
         } else {
             return Ok(None);
         };

crates/graphql/src/model/users.rs

@@ -17,7 +17,7 @@ use async_graphql::{
     Context, Description, Object, ID,
 };
 use chrono::{DateTime, Utc};
-use mas_storage::{Repository, UpstreamOAuthLinkRepository};
+use mas_storage::{user::UserEmailRepository, Repository, UpstreamOAuthLinkRepository};
 use sqlx::PgPool;
 
 use super::{
@@ -54,8 +54,14 @@ impl User {
     }
 
     /// Primary email address of the user.
-    async fn primary_email(&self) -> Option<UserEmail> {
+    async fn primary_email(
-        self.0.primary_email.clone().map(UserEmail)
+        &self,
+        ctx: &Context<'_>,
+    ) -> Result<Option<UserEmail>, async_graphql::Error> {
+        let database = ctx.data::<PgPool>()?;
+        let mut conn = database.acquire().await?;
+
+        Ok(conn.user_email().get_primary(&self.0).await?.map(UserEmail))
     }
 
     /// Get the list of compatibility SSO logins, chronologically sorted
@@ -182,18 +188,17 @@ impl User {
                     .map(|x: OpaqueCursor<NodeCursor>| x.extract_for_type(NodeType::UserEmail))
                     .transpose()?;
 
-                let (has_previous_page, has_next_page, edges) =
+                let page = conn
-                    mas_storage::user::get_paginated_user_emails(
+                    .user_email()
-                        &mut conn, &self.0, before_id, after_id, first, last,
+                    .list_paginated(&self.0, before_id, after_id, first, last)
-                    )
                     .await?;
 
                 let mut connection = Connection::with_additional_fields(
-                    has_previous_page,
+                    page.has_previous_page,
-                    has_next_page,
+                    page.has_next_page,
                     UserEmailsPagination(self.0.clone()),
                 );
-                connection.edges.extend(edges.into_iter().map(|u| {
+                connection.edges.extend(page.edges.into_iter().map(|u| {
                     Edge::new(
                         OpaqueCursor(NodeCursor(NodeType::UserEmail, u.id)),
                         UserEmail(u),
@@ -339,9 +344,9 @@ pub struct UserEmailsPagination(mas_data_model::User);
 #[Object]
 impl UserEmailsPagination {
     /// Identifies the total count of items in the connection.
-    async fn total_count(&self, ctx: &Context<'_>) -> Result<i64, async_graphql::Error> {
+    async fn total_count(&self, ctx: &Context<'_>) -> Result<usize, async_graphql::Error> {
         let mut conn = ctx.data::<PgPool>()?.acquire().await?;
-        let count = mas_storage::user::count_user_emails(&mut conn, &self.0).await?;
+        let count = conn.user_email().count(&self.0).await?;
         Ok(count)
     }
 }

crates/handlers/src/compat/login.rs

@@ -21,8 +21,8 @@ use mas_storage::{
         add_compat_access_token, add_compat_refresh_token, get_compat_sso_login_by_token,
         mark_compat_sso_login_as_exchanged, start_compat_session,
     },
-    user::{add_user_password, lookup_user_by_username, lookup_user_password},
+    user::{add_user_password, lookup_user_password, UserRepository},
-    Clock,
+    Clock, Repository,
 };
 use serde::{Deserialize, Serialize};
 use serde_with::{serde_as, skip_serializing_none, DurationMilliSeconds};
@@ -314,7 +314,9 @@ async fn user_password_login(
     let (clock, mut rng) = crate::clock_and_rng();
 
     // Find the user
-    let user = lookup_user_by_username(&mut *txn, &username)
+    let user = txn
+        .user()
+        .find_by_username(&username)
         .await?
         .ok_or(RouteError::UserNotFound)?;
 

crates/handlers/src/compat/login_sso_complete.rs

@@ -80,14 +80,8 @@ pub async fn get(
         return Ok((cookie_jar, url).into_response());
     };
 
-    // TODO: make that more generic
+    // TODO: make that more generic, check that the email has been confirmed
-    if session
+    if session.user.primary_user_email_id.is_none() {
-        .user
-        .primary_email
-        .as_ref()
-        .and_then(|e| e.confirmed_at)
-        .is_none()
-    {
         let destination = mas_router::AccountAddEmail::default()
             .and_then(PostAuthAction::continue_compat_sso_login(id));
         return Ok((cookie_jar, destination.go()).into_response());
@@ -149,13 +143,7 @@ pub async fn post(
     };
 
     // TODO: make that more generic
-    if session
+    if session.user.primary_user_email_id.is_none() {
-        .user
-        .primary_email
-        .as_ref()
-        .and_then(|e| e.confirmed_at)
-        .is_none()
-    {
         let destination = mas_router::AccountAddEmail::default()
             .and_then(PostAuthAction::continue_compat_sso_login(id));
         return Ok((cookie_jar, destination.go()).into_response());

crates/handlers/src/oauth2/userinfo.rs

@@ -28,6 +28,7 @@ use mas_jose::{
 };
 use mas_keystore::Keystore;
 use mas_router::UrlBuilder;
+use mas_storage::{user::UserEmailRepository, Repository};
 use oauth2_types::scope;
 use serde::Serialize;
 use serde_with::skip_serializing_none;
@@ -66,6 +67,7 @@ pub enum RouteError {
 }
 
 impl_from_error_for_route!(sqlx::Error);
+impl_from_error_for_route!(mas_storage::DatabaseError);
 impl_from_error_for_route!(mas_keystore::WrongAlgorithmError);
 impl_from_error_for_route!(mas_jose::jwt::JwtSignatureError);
 
@@ -92,19 +94,19 @@ pub async fn get(
     let session = user_authorization.protected(&mut conn).await?;
 
     let user = session.browser_session.user;
-    let mut user_info = UserInfo {
+
-        sub: user.sub,
+    let user_email = if session.scope.contains(&scope::EMAIL) {
-        username: user.username,
+        conn.user_email().get_primary(&user).await?
-        email: None,
+    } else {
-        email_verified: None,
+        None
     };
 
-    if session.scope.contains(&scope::EMAIL) {
+    let user_info = UserInfo {
-        if let Some(email) = user.primary_email {
+        sub: user.sub.clone(),
-            user_info.email_verified = Some(email.confirmed_at.is_some());
+        username: user.username.clone(),
-            user_info.email = Some(email.email);
+        email_verified: user_email.as_ref().map(|u| u.confirmed_at.is_some()),
-        }
+        email: user_email.map(|u| u.email),
-    }
+    };
 
     if let Some(alg) = session.client.userinfo_signed_response_alg {
         let key = key_store

crates/handlers/src/upstream_oauth2/link.rs

@@ -26,7 +26,7 @@ use mas_axum_utils::{
 use mas_keystore::Encrypter;
 use mas_storage::{
     upstream_oauth2::UpstreamOAuthSessionRepository,
-    user::{add_user, authenticate_session_with_upstream, lookup_user, start_session},
+    user::{authenticate_session_with_upstream, start_session, UserRepository},
     Repository, UpstreamOAuthLinkRepository,
 };
 use mas_templates::{
@@ -51,6 +51,10 @@ pub(crate) enum RouteError {
     #[error("Session not found")]
     SessionNotFound,
 
+    /// Couldn't find the user
+    #[error("User not found")]
+    UserNotFound,
+
     /// Session was already consumed
     #[error("Session already consumed")]
     SessionConsumed,
@@ -157,7 +161,11 @@ pub(crate) async fn get(
             // Session already linked, but link doesn't match the currently
             // logged user. Suggest logging out of the current user
             // and logging in with the new one
-            let user = lookup_user(&mut txn, user_id).await?;
+            let user = txn
+                .user()
+                .lookup(user_id)
+                .await?
+                .ok_or(RouteError::UserNotFound)?;
 
             let ctx = UpstreamExistingLinkContext::new(user)
                 .with_session(user_session)
@@ -177,7 +185,11 @@ pub(crate) async fn get(
 
         (None, Some(user_id)) => {
             // Session linked, but user not logged in: do the login
-            let user = lookup_user(&mut txn, user_id).await?;
+            let user = txn
+                .user()
+                .lookup(user_id)
+                .await?
+                .ok_or(RouteError::UserNotFound)?;
 
             let ctx = UpstreamExistingLinkContext::new(user).with_csrf(csrf_token.form_value());
 
@@ -250,12 +262,17 @@ pub(crate) async fn post(
         }
 
         (None, Some(user_id), FormData::Login) => {
-            let user = lookup_user(&mut txn, user_id).await?;
+            let user = txn
+                .user()
+                .lookup(user_id)
+                .await?
+                .ok_or(RouteError::UserNotFound)?;
+
             start_session(&mut txn, &mut rng, &clock, user).await?
         }
 
         (None, None, FormData::Register { username }) => {
-            let user = add_user(&mut txn, &mut rng, &clock, &username).await?;
+            let user = txn.user().add(&mut rng, &clock, username).await?;
             txn.upstream_oauth_link()
                 .associate_to_user(&link, &user)
                 .await?;

crates/handlers/src/views/account/emails/add.rs

@@ -24,7 +24,7 @@ use mas_axum_utils::{
 use mas_email::Mailer;
 use mas_keystore::Encrypter;
 use mas_router::Route;
-use mas_storage::user::add_user_email;
+use mas_storage::{user::UserEmailRepository, Repository};
 use mas_templates::{EmailAddContext, TemplateContext, Templates};
 use serde::Deserialize;
 use sqlx::PgPool;
@@ -88,7 +88,11 @@ pub(crate) async fn post(
         return Ok((cookie_jar, login.go()).into_response());
     };
 
-    let user_email = add_user_email(&mut txn, &mut rng, &clock, &session.user, form.email).await?;
+    let user_email = txn
+        .user_email()
+        .add(&mut rng, &clock, &session.user, form.email)
+        .await?;
+
     let next = mas_router::AccountVerifyEmail::new(user_email.id);
     let next = if let Some(action) = query.post_auth_action {
         next.and_then(action)

crates/handlers/src/views/account/emails/mod.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use anyhow::{anyhow, Context};
 use axum::{
     extract::{Form, State},
     response::{Html, IntoResponse, Response},
@@ -27,17 +28,11 @@ use mas_data_model::{BrowserSession, User, UserEmail};
 use mas_email::Mailer;
 use mas_keystore::Encrypter;
 use mas_router::Route;
-use mas_storage::{
+use mas_storage::{user::UserEmailRepository, Clock, Repository};
-    user::{
-        add_user_email, add_user_email_verification_code, get_user_email, get_user_emails,
-        remove_user_email, set_user_email_as_primary,
-    },
-    Clock,
-};
 use mas_templates::{AccountEmailsContext, EmailVerificationContext, TemplateContext, Templates};
 use rand::{distributions::Uniform, Rng};
 use serde::Deserialize;
-use sqlx::{PgExecutor, PgPool};
+use sqlx::{PgConnection, PgPool};
 use tracing::info;
 
 pub mod add;
@@ -79,11 +74,11 @@ async fn render(
     templates: Templates,
     session: BrowserSession,
     cookie_jar: PrivateCookieJar<Encrypter>,
-    executor: impl PgExecutor<'_>,
+    conn: &mut PgConnection,
 ) -> Result<Response, FancyError> {
     let (csrf_token, cookie_jar) = cookie_jar.csrf_token(clock.now(), rng);
 
-    let emails = get_user_emails(executor, &session.user).await?;
+    let emails = conn.user_email().all(&session.user).await?;
 
     let ctx = AccountEmailsContext::new(emails)
         .with_session(session)
@@ -96,7 +91,7 @@ async fn render(
 
 async fn start_email_verification(
     mailer: &Mailer,
-    executor: impl PgExecutor<'_>,
+    conn: &mut PgConnection,
     mut rng: impl Rng + Send,
     clock: &Clock,
     user: &User,
@@ -108,14 +103,9 @@ async fn start_email_verification(
 
     let address: Address = user_email.email.parse()?;
 
-    let verification = add_user_email_verification_code(
+    let verification = conn
-        executor,
+        .user_email()
-        &mut rng,
+        .add_verification_code(&mut rng, clock, &user_email, Duration::hours(8), code)
-        clock,
-        user_email,
-        Duration::hours(8),
-        code,
-    )
         .await?;
 
     // And send the verification email
@@ -126,7 +116,7 @@ async fn start_email_verification(
     mailer.send_verification_email(mailbox, &context).await?;
 
     info!(
-        email.id = %verification.email.id,
+        email.id = %user_email.id,
         "Verification email sent"
     );
     Ok(())
@@ -157,17 +147,13 @@ pub(crate) async fn post(
 
     match form {
         ManagementForm::Add { email } => {
-            let user_email =
+            let email = txn
-                add_user_email(&mut txn, &mut rng, &clock, &session.user, email).await?;
+                .user_email()
-            let next = mas_router::AccountVerifyEmail::new(user_email.id);
+                .add(&mut rng, &clock, &session.user, email)
-            start_email_verification(
+                .await?;
-                &mailer,
+
-                &mut txn,
+            let next = mas_router::AccountVerifyEmail::new(email.id);
-                &mut rng,
+            start_email_verification(&mailer, &mut txn, &mut rng, &clock, &session.user, email)
-                &clock,
-                &session.user,
-                user_email,
-            )
                 .await?;
             txn.commit().await?;
             return Ok((cookie_jar, next.go()).into_response());
@@ -175,16 +161,18 @@ pub(crate) async fn post(
         ManagementForm::ResendConfirmation { id } => {
             let id = id.parse()?;
 
-            let user_email = get_user_email(&mut txn, &session.user, id).await?;
+            let email = txn
-            let next = mas_router::AccountVerifyEmail::new(user_email.id);
+                .user_email()
-            start_email_verification(
+                .lookup(id)
-                &mailer,
+                .await?
-                &mut txn,
+                .context("Email not found")?;
-                &mut rng,
+
-                &clock,
+            if email.user_id != session.user.id {
-                &session.user,
+                return Err(anyhow!("Email not found").into());
-                user_email,
+            }
-            )
+
+            let next = mas_router::AccountVerifyEmail::new(email.id);
+            start_email_verification(&mailer, &mut txn, &mut rng, &clock, &session.user, email)
                 .await?;
             txn.commit().await?;
             return Ok((cookie_jar, next.go()).into_response());
@@ -192,14 +180,32 @@ pub(crate) async fn post(
         ManagementForm::Remove { id } => {
             let id = id.parse()?;
 
-            let email = get_user_email(&mut txn, &session.user, id).await?;
+            let email = txn
-            remove_user_email(&mut txn, email).await?;
+                .user_email()
+                .lookup(id)
+                .await?
+                .context("Email not found")?;
+
+            if email.user_id != session.user.id {
+                return Err(anyhow!("Email not found").into());
+            }
+
+            txn.user_email().remove(email).await?;
         }
         ManagementForm::SetPrimary { id } => {
             let id = id.parse()?;
-            let email = get_user_email(&mut txn, &session.user, id).await?;
+            let email = txn
-            set_user_email_as_primary(&mut txn, &email).await?;
+                .user_email()
-            session.user.primary_email = Some(email);
+                .lookup(id)
+                .await?
+                .context("Email not found")?;
+
+            if email.user_id != session.user.id {
+                return Err(anyhow!("Email not found").into());
+            }
+
+            txn.user_email().set_as_primary(&email).await?;
+            session.user.primary_user_email_id = Some(email.id);
         }
     };
 

crates/handlers/src/views/account/emails/verify.rs

@@ -24,13 +24,7 @@ use mas_axum_utils::{
 };
 use mas_keystore::Encrypter;
 use mas_router::Route;
-use mas_storage::{
+use mas_storage::{user::UserEmailRepository, Clock, Repository};
-    user::{
-        consume_email_verification, lookup_user_email_by_id, lookup_user_email_verification_code,
-        mark_user_email_as_verified, set_user_email_as_primary,
-    },
-    Clock,
-};
 use mas_templates::{EmailVerificationPageContext, TemplateContext, Templates};
 use serde::Deserialize;
 use sqlx::PgPool;
@@ -65,8 +59,11 @@ pub(crate) async fn get(
         return Ok((cookie_jar, login.go()).into_response());
     };
 
-    let user_email = lookup_user_email_by_id(&mut conn, &session.user, id)
+    let user_email = conn
+        .user_email()
+        .lookup(id)
         .await?
+        .filter(|u| u.user_id == session.user.id)
         .context("Could not find user email")?;
 
     if user_email.confirmed_at.is_some() {
@@ -106,23 +103,31 @@ pub(crate) async fn post(
         return Ok((cookie_jar, login.go()).into_response());
     };
 
-    let email = lookup_user_email_by_id(&mut txn, &session.user, id)
+    let user_email = txn
+        .user_email()
+        .lookup(id)
         .await?
+        .filter(|u| u.user_id == session.user.id)
         .context("Could not find user email")?;
 
-    if session.user.primary_email.is_none() {
+    let verification = txn
-        set_user_email_as_primary(&mut txn, &email).await?;
+        .user_email()
-    }
+        .find_verification_code(&clock, &user_email, &form.code)
-
-    // TODO: make those 8 hours configurable
-    let verification = lookup_user_email_verification_code(&mut txn, &clock, email, &form.code)
         .await?
         .context("Invalid code")?;
 
     // TODO: display nice errors if the code was already consumed or expired
-    let verification = consume_email_verification(&mut txn, &clock, verification).await?;
+    txn.user_email()
+        .consume_verification_code(&clock, verification)
+        .await?;
 
-    let _email = mark_user_email_as_verified(&mut txn, &clock, verification.email).await?;
+    if session.user.primary_user_email_id.is_none() {
+        txn.user_email().set_as_primary(&user_email).await?;
+    }
+
+    txn.user_email()
+        .mark_as_verified(&clock, user_email)
+        .await?;
 
     txn.commit().await?;
 

crates/handlers/src/views/account/mod.rs

@@ -23,7 +23,10 @@ use axum_extra::extract::PrivateCookieJar;
 use mas_axum_utils::{csrf::CsrfExt, FancyError, SessionInfoExt};
 use mas_keystore::Encrypter;
 use mas_router::Route;
-use mas_storage::user::{count_active_sessions, get_user_emails};
+use mas_storage::{
+    user::{count_active_sessions, UserEmailRepository},
+    Repository,
+};
 use mas_templates::{AccountContext, TemplateContext, Templates};
 use sqlx::PgPool;
 
@@ -49,7 +52,7 @@ pub(crate) async fn get(
 
     let active_sessions = count_active_sessions(&mut conn, &session.user).await?;
 
-    let emails = get_user_emails(&mut conn, &session.user).await?;
+    let emails = conn.user_email().all(&session.user).await?;
 
     let ctx = AccountContext::new(active_sessions, emails)
         .with_session(session)

crates/handlers/src/views/login.rs

@@ -26,8 +26,8 @@ use mas_keystore::Encrypter;
 use mas_storage::{
     upstream_oauth2::UpstreamOAuthProviderRepository,
     user::{
-        add_user_password, authenticate_session_with_password, lookup_user_by_username,
+        add_user_password, authenticate_session_with_password, lookup_user_password, start_session,
-        lookup_user_password, start_session,
+        UserRepository,
     },
     Clock, Repository,
 };
@@ -130,8 +130,6 @@ pub(crate) async fn post(
         return Ok((cookie_jar, Html(content)).into_response());
     }
 
-    lookup_user_by_username(&mut conn, &form.username).await?;
-
     match login(
         password_manager,
         &mut conn,
@@ -175,7 +173,9 @@ async fn login(
 ) -> Result<BrowserSession, FormError> {
     // XXX: we're loosing the error context here
     // First, lookup the user
-    let user = lookup_user_by_username(&mut *conn, username)
+    let user = conn
+        .user()
+        .find_by_username(username)
         .await
         .map_err(|_e| FormError::Internal)?
         .ok_or(FormError::InvalidCredentials)?;

crates/handlers/src/views/register.rs

@@ -31,9 +31,12 @@ use mas_email::Mailer;
 use mas_keystore::Encrypter;
 use mas_policy::PolicyFactory;
 use mas_router::Route;
-use mas_storage::user::{
+use mas_storage::{
-    add_user, add_user_email, add_user_email_verification_code, add_user_password,
+    user::{
-    authenticate_session_with_password, start_session, username_exists,
+        add_user_password, authenticate_session_with_password, start_session, UserEmailRepository,
+        UserRepository,
+    },
+    Repository,
 };
 use mas_templates::{
     EmailVerificationContext, FieldError, FormError, RegisterContext, RegisterFormField,
@@ -114,7 +117,7 @@ pub(crate) async fn post(
 
         if form.username.is_empty() {
             state.add_error_on_field(RegisterFormField::Username, FieldError::Required);
-        } else if username_exists(&mut txn, &form.username).await? {
+        } else if txn.user().exists(&form.username).await? {
             state.add_error_on_field(RegisterFormField::Username, FieldError::Exists);
         }
 
@@ -185,7 +188,7 @@ pub(crate) async fn post(
         return Ok((cookie_jar, Html(content)).into_response());
     }
 
-    let user = add_user(&mut txn, &mut rng, &clock, &form.username).await?;
+    let user = txn.user().add(&mut rng, &clock, form.username).await?;
     let password = Zeroizing::new(form.password.into_bytes());
     let (version, hashed_password) = password_manager.hash(&mut rng, password).await?;
     let user_password = add_user_password(
@@ -199,7 +202,10 @@ pub(crate) async fn post(
     )
     .await?;
 
-    let user_email = add_user_email(&mut txn, &mut rng, &clock, &user, form.email).await?;
+    let user_email = txn
+        .user_email()
+        .add(&mut rng, &clock, &user, form.email)
+        .await?;
 
     // First, generate a code
     let range = Uniform::<u32>::from(0..1_000_000);
@@ -208,14 +214,9 @@ pub(crate) async fn post(
 
     let address: Address = user_email.email.parse()?;
 
-    let verification = add_user_email_verification_code(
+    let verification = txn
-        &mut txn,
+        .user_email()
-        &mut rng,
+        .add_verification_code(&mut rng, &clock, &user_email, Duration::hours(8), code)
-        &clock,
-        user_email,
-        Duration::hours(8),
-        code,
-    )
         .await?;
 
     // And send the verification email
@@ -225,8 +226,7 @@ pub(crate) async fn post(
 
     mailer.send_verification_email(mailbox, &context).await?;
 
-    let next = mas_router::AccountVerifyEmail::new(verification.email.id)
+    let next = mas_router::AccountVerifyEmail::new(user_email.id).and_maybe(query.post_auth_action);
-        .and_maybe(query.post_auth_action);
 
     let mut session = start_session(&mut txn, &mut rng, &clock, user).await?;
     authenticate_session_with_password(&mut txn, &mut rng, &clock, &mut session, &user_password)

crates/storage/src/compat.rs

@@ -15,7 +15,7 @@
 use chrono::{DateTime, Duration, Utc};
 use mas_data_model::{
     CompatAccessToken, CompatRefreshToken, CompatSession, CompatSsoLogin, CompatSsoLoginState,
-    Device, User, UserEmail,
+    Device, User,
 };
 use rand::Rng;
 use sqlx::{Acquire, PgExecutor, Postgres, QueryBuilder};
@@ -40,10 +40,7 @@ struct CompatAccessTokenLookup {
     compat_session_device_id: String,
     user_id: Uuid,
     user_username: String,
-    user_email_id: Option<Uuid>,
+    user_primary_user_email_id: Option<Uuid>,
-    user_email: Option<String>,
-    user_email_created_at: Option<DateTime<Utc>>,
-    user_email_confirmed_at: Option<DateTime<Utc>>,
 }
 
 #[tracing::instrument(skip_all, err)]
@@ -66,18 +63,13 @@ pub async fn lookup_active_compat_access_token(
                 cs.device_id       AS "compat_session_device_id",
                  u.user_id         AS "user_id!",
                  u.username        AS "user_username!",
-                ue.user_email_id   AS "user_email_id?",
+                 u.primary_user_email_id AS "user_primary_user_email_id"
-                ue.email           AS "user_email?",
-                ue.created_at      AS "user_email_created_at?",
-                ue.confirmed_at    AS "user_email_confirmed_at?"
 
             FROM compat_access_tokens ct
             INNER JOIN compat_sessions cs
               USING (compat_session_id)
             INNER JOIN users u
               USING (user_id)
-            LEFT JOIN user_emails ue
-              ON ue.user_email_id = u.primary_user_email_id
 
             WHERE ct.access_token = $1
               AND (ct.expires_at < $2 OR ct.expires_at IS NULL)
@@ -101,32 +93,11 @@ pub async fn lookup_active_compat_access_token(
     };
 
     let user_id = Ulid::from(res.user_id);
-    let primary_email = match (
-        res.user_email_id,
-        res.user_email,
-        res.user_email_created_at,
-        res.user_email_confirmed_at,
-    ) {
-        (Some(id), Some(email), Some(created_at), confirmed_at) => Some(UserEmail {
-            id: id.into(),
-            email,
-            created_at,
-            confirmed_at,
-        }),
-        (None, None, None, None) => None,
-        _ => {
-            return Err(DatabaseInconsistencyError::on("compat_sessions")
-                .column("user_id")
-                .row(user_id)
-                .into())
-        }
-    };
-
     let user = User {
         id: user_id,
         username: res.user_username,
         sub: user_id.to_string(),
-        primary_email,
+        primary_user_email_id: res.user_primary_user_email_id.map(Into::into),
     };
 
     let id = res.compat_session_id.into();
@@ -162,10 +133,7 @@ pub struct CompatRefreshTokenLookup {
     compat_session_device_id: String,
     user_id: Uuid,
     user_username: String,
-    user_email_id: Option<Uuid>,
+    user_primary_user_email_id: Option<Uuid>,
-    user_email: Option<String>,
-    user_email_created_at: Option<DateTime<Utc>>,
-    user_email_confirmed_at: Option<DateTime<Utc>>,
 }
 
 #[tracing::instrument(skip_all, err)]
@@ -191,10 +159,7 @@ pub async fn lookup_active_compat_refresh_token(
                 cs.device_id       AS "compat_session_device_id",
                 u.user_id,
                 u.username         AS "user_username!",
-                ue.user_email_id   AS "user_email_id?",
+                 u.primary_user_email_id AS "user_primary_user_email_id"
-                ue.email           AS "user_email?",
-                ue.created_at      AS "user_email_created_at?",
-                ue.confirmed_at    AS "user_email_confirmed_at?"
 
             FROM compat_refresh_tokens cr
             INNER JOIN compat_sessions cs
@@ -203,8 +168,6 @@ pub async fn lookup_active_compat_refresh_token(
               USING (compat_access_token_id)
             INNER JOIN users u
               USING (user_id)
-            LEFT JOIN user_emails ue
-              ON ue.user_email_id = u.primary_user_email_id
 
             WHERE cr.refresh_token = $1
               AND cr.consumed_at IS NULL
@@ -233,32 +196,11 @@ pub async fn lookup_active_compat_refresh_token(
     };
 
     let user_id = Ulid::from(res.user_id);
-    let primary_email = match (
-        res.user_email_id,
-        res.user_email,
-        res.user_email_created_at,
-        res.user_email_confirmed_at,
-    ) {
-        (Some(id), Some(email), Some(created_at), confirmed_at) => Some(UserEmail {
-            id: id.into(),
-            email,
-            created_at,
-            confirmed_at,
-        }),
-        (None, None, None, None) => None,
-        _ => {
-            return Err(DatabaseInconsistencyError::on("users")
-                .column("primary_user_email_id")
-                .row(user_id)
-                .into())
-        }
-    };
-
     let user = User {
         id: user_id,
         username: res.user_username,
         sub: user_id.to_string(),
-        primary_email,
+        primary_user_email_id: res.user_primary_user_email_id.map(Into::into),
     };
 
     let session_id = res.compat_session_id.into();
@@ -528,10 +470,7 @@ struct CompatSsoLoginLookup {
     compat_session_device_id: Option<String>,
     user_id: Option<Uuid>,
     user_username: Option<String>,
-    user_email_id: Option<Uuid>,
+    user_primary_user_email_id: Option<Uuid>,
-    user_email: Option<String>,
-    user_email_created_at: Option<DateTime<Utc>>,
-    user_email_confirmed_at: Option<DateTime<Utc>>,
 }
 
 impl TryFrom<CompatSsoLoginLookup> for CompatSsoLogin {
@@ -546,32 +485,18 @@ impl TryFrom<CompatSsoLoginLookup> for CompatSsoLogin {
                 .source(e)
         })?;
 
-        let primary_email = match (
+        let user = match (
-            res.user_email_id,
+            res.user_id,
-            res.user_email,
+            res.user_username,
-            res.user_email_created_at,
+            res.user_primary_user_email_id,
-            res.user_email_confirmed_at,
         ) {
-            (Some(id), Some(email), Some(created_at), confirmed_at) => Some(UserEmail {
+            (Some(id), Some(username), primary_email_id) => {
-                id: id.into(),
-                email,
-                created_at,
-                confirmed_at,
-            }),
-            (None, None, None, None) => None,
-            _ => {
-                return Err(DatabaseInconsistencyError::on("users").column("primary_user_email_id"))
-            }
-        };
-
-        let user = match (res.user_id, res.user_username, primary_email) {
-            (Some(id), Some(username), primary_email) => {
                 let id = Ulid::from(id);
                 Some(User {
                     id,
                     username,
                     sub: id.to_string(),
-                    primary_email,
+                    primary_user_email_id: primary_email_id.map(Into::into),
                 })
             }
 
@@ -667,17 +592,12 @@ pub async fn get_compat_sso_login_by_id(
                 cs.device_id       AS "compat_session_device_id?",
                 u.user_id          AS "user_id?",
                 u.username         AS "user_username?",
-                ue.user_email_id   AS "user_email_id?",
+                u.primary_user_email_id AS "user_primary_user_email_id?"
-                ue.email           AS "user_email?",
-                ue.created_at      AS "user_email_created_at?",
-                ue.confirmed_at    AS "user_email_confirmed_at?"
             FROM compat_sso_logins cl
             LEFT JOIN compat_sessions cs
               USING (compat_session_id)
             LEFT JOIN users u
               USING (user_id)
-            LEFT JOIN user_emails ue
-              ON ue.user_email_id = u.primary_user_email_id
             WHERE cl.compat_sso_login_id = $1
         "#,
         Uuid::from(id),
@@ -725,17 +645,12 @@ pub async fn get_paginated_user_compat_sso_logins(
                 cs.device_id       AS "compat_session_device_id",
                 u.user_id          AS "user_id",
                 u.username         AS "user_username",
-                ue.user_email_id   AS "user_email_id",
+                u.primary_user_email_id AS "user_primary_user_email_id?"
-                ue.email           AS "user_email",
-                ue.created_at      AS "user_email_created_at",
-                ue.confirmed_at    AS "user_email_confirmed_at"
             FROM compat_sso_logins cl
             LEFT JOIN compat_sessions cs
               USING (compat_session_id)
             LEFT JOIN users u
               USING (user_id)
-            LEFT JOIN user_emails ue
-              ON ue.user_email_id = u.primary_user_email_id
         "#,
     );
 
@@ -781,17 +696,12 @@ pub async fn get_compat_sso_login_by_token(
                 cs.device_id       AS "compat_session_device_id?",
                 u.user_id          AS "user_id?",
                 u.username         AS "user_username?",
-                ue.user_email_id   AS "user_email_id?",
+                u.primary_user_email_id AS "user_primary_user_email_id?"
-                ue.email           AS "user_email?",
-                ue.created_at      AS "user_email_created_at?",
-                ue.confirmed_at    AS "user_email_confirmed_at?"
             FROM compat_sso_logins cl
             LEFT JOIN compat_sessions cs
               USING (compat_session_id)
             LEFT JOIN users u
               USING (user_id)
-            LEFT JOIN user_emails ue
-              ON ue.user_email_id = u.primary_user_email_id
             WHERE cl.login_token = $1
         "#,
         token,

crates/storage/src/lib.rs

@@ -161,7 +161,7 @@ impl DatabaseInconsistencyError {
     }
 }
 
-#[derive(Default, Debug, Clone, Copy)]
+#[derive(Default, Debug, Clone)]
 pub struct Clock {
     _private: (),
 }

crates/storage/src/oauth2/access_token.rs

@@ -13,7 +13,7 @@
 // limitations under the License.
 
 use chrono::{DateTime, Duration, Utc};
-use mas_data_model::{AccessToken, Authentication, BrowserSession, Session, User, UserEmail};
+use mas_data_model::{AccessToken, Authentication, BrowserSession, Session, User};
 use rand::Rng;
 use sqlx::{PgConnection, PgExecutor};
 use ulid::Ulid;
@@ -84,12 +84,9 @@ pub struct OAuth2AccessTokenLookup {
     user_session_created_at: DateTime<Utc>,
     user_id: Uuid,
     user_username: String,
+    user_primary_user_email_id: Option<Uuid>,
     user_session_last_authentication_id: Option<Uuid>,
     user_session_last_authentication_created_at: Option<DateTime<Utc>>,
-    user_email_id: Option<Uuid>,
-    user_email: Option<String>,
-    user_email_created_at: Option<DateTime<Utc>>,
-    user_email_confirmed_at: Option<DateTime<Utc>>,
 }
 
 #[allow(clippy::too_many_lines)]
@@ -100,24 +97,20 @@ pub async fn lookup_active_access_token(
     let res = sqlx::query_as!(
         OAuth2AccessTokenLookup,
         r#"
-            SELECT
+            SELECT at.oauth2_access_token_id
-                at.oauth2_access_token_id,
+                 , at.access_token      AS "oauth2_access_token"
-                at.access_token    AS "oauth2_access_token",
+                 , at.created_at        AS "oauth2_access_token_created_at"
-                at.created_at      AS "oauth2_access_token_created_at",
+                 , at.expires_at        AS "oauth2_access_token_expires_at"
-                at.expires_at      AS "oauth2_access_token_expires_at",
+                 , os.oauth2_session_id AS "oauth2_session_id!"
-                os.oauth2_session_id AS "oauth2_session_id!",
+                 , os.oauth2_client_id  AS "oauth2_client_id!"
-                os.oauth2_client_id AS "oauth2_client_id!",
+                 , os.scope             AS "scope!"
-                os.scope           AS "scope!",
+                 , us.user_session_id   AS "user_session_id!"
-                us.user_session_id AS "user_session_id!",
+                 , us.created_at        AS "user_session_created_at!"
-                us.created_at      AS "user_session_created_at!",
+                 ,  u.user_id           AS "user_id!"
-                 u.user_id AS "user_id!",
+                 ,  u.username          AS "user_username!"
-                 u.username        AS "user_username!",
+                 ,  u.primary_user_email_id AS "user_primary_user_email_id"
-                usa.user_session_authentication_id AS "user_session_last_authentication_id?",
+                 , usa.user_session_authentication_id AS "user_session_last_authentication_id?"
-                usa.created_at     AS "user_session_last_authentication_created_at?",
+                 , usa.created_at       AS "user_session_last_authentication_created_at?"
-                ue.user_email_id AS "user_email_id?",
-                ue.email           AS "user_email?",
-                ue.created_at      AS "user_email_created_at?",
-                ue.confirmed_at    AS "user_email_confirmed_at?"
 
             FROM oauth2_access_tokens at
             INNER JOIN oauth2_sessions os
@@ -128,8 +121,6 @@ pub async fn lookup_active_access_token(
               USING (user_id)
             LEFT JOIN user_session_authentications usa
               USING (user_session_id)
-            LEFT JOIN user_emails ue
-              ON ue.user_email_id = u.primary_user_email_id
 
             WHERE at.access_token = $1
               AND at.revoked_at IS NULL
@@ -162,32 +153,11 @@ pub async fn lookup_active_access_token(
         })?;
 
     let user_id = Ulid::from(res.user_id);
-    let primary_email = match (
-        res.user_email_id,
-        res.user_email,
-        res.user_email_created_at,
-        res.user_email_confirmed_at,
-    ) {
-        (Some(id), Some(email), Some(created_at), confirmed_at) => Some(UserEmail {
-            id: id.into(),
-            email,
-            created_at,
-            confirmed_at,
-        }),
-        (None, None, None, None) => None,
-        _ => {
-            return Err(DatabaseInconsistencyError::on("users")
-                .column("primary_user_email_id")
-                .row(user_id)
-                .into())
-        }
-    };
-
     let user = User {
         id: user_id,
         username: res.user_username,
         sub: user_id.to_string(),
-        primary_email,
+        primary_user_email_id: res.user_primary_user_email_id.map(Into::into),
     };
 
     let last_authentication = match (

crates/storage/src/oauth2/authorization_grant.rs

@@ -17,7 +17,7 @@ use std::num::NonZeroU32;
 use chrono::{DateTime, Utc};
 use mas_data_model::{
     Authentication, AuthorizationCode, AuthorizationGrant, AuthorizationGrantStage, BrowserSession,
-    Client, Pkce, Session, User, UserEmail,
+    Client, Pkce, Session, User,
 };
 use mas_iana::oauth::PkceCodeChallengeMethod;
 use oauth2_types::{requests::ResponseMode, scope::Scope};
@@ -154,12 +154,9 @@ struct GrantLookup {
     user_session_created_at: Option<DateTime<Utc>>,
     user_id: Option<Uuid>,
     user_username: Option<String>,
+    user_primary_user_email_id: Option<Uuid>,
     user_session_last_authentication_id: Option<Uuid>,
     user_session_last_authentication_created_at: Option<DateTime<Utc>>,
-    user_email_id: Option<Uuid>,
-    user_email: Option<String>,
-    user_email_created_at: Option<DateTime<Utc>>,
-    user_email_confirmed_at: Option<DateTime<Utc>>,
 }
 
 impl GrantLookup {
@@ -197,34 +194,14 @@ impl GrantLookup {
             _ => return Err(DatabaseInconsistencyError::on("user_session_authentications").into()),
         };
 
-        let primary_email = match (
-            self.user_email_id,
-            self.user_email,
-            self.user_email_created_at,
-            self.user_email_confirmed_at,
-        ) {
-            (Some(id), Some(email), Some(created_at), confirmed_at) => Some(UserEmail {
-                id: id.into(),
-                email,
-                created_at,
-                confirmed_at,
-            }),
-            (None, None, None, None) => None,
-            _ => {
-                return Err(DatabaseInconsistencyError::on("users")
-                    .column("primary_user_email_id")
-                    .into())
-            }
-        };
-
         let session = match (
             self.oauth2_session_id,
             self.user_session_id,
             self.user_session_created_at,
             self.user_id,
             self.user_username,
+            self.user_primary_user_email_id,
             last_authentication,
-            primary_email,
         ) {
             (
                 Some(session_id),
@@ -232,15 +209,15 @@ impl GrantLookup {
                 Some(user_session_created_at),
                 Some(user_id),
                 Some(user_username),
+                user_primary_user_email_id,
                 last_authentication,
-                primary_email,
             ) => {
                 let user_id = Ulid::from(user_id);
                 let user = User {
                     id: user_id,
                     username: user_username,
                     sub: user_id.to_string(),
-                    primary_email,
+                    primary_user_email_id: user_primary_user_email_id.map(Into::into),
                 };
 
                 let browser_session = BrowserSession {
@@ -439,12 +416,9 @@ pub async fn get_grant_by_id(
                 us.created_at              AS "user_session_created_at?",
                  u.user_id                 AS "user_id?",
                  u.username                AS "user_username?",
+                 u.primary_user_email_id   AS "user_primary_user_email_id?",
                 usa.user_session_authentication_id AS "user_session_last_authentication_id?",
-                usa.created_at             AS "user_session_last_authentication_created_at?",
+                usa.created_at             AS "user_session_last_authentication_created_at?"
-                ue.user_email_id           AS "user_email_id?",
-                ue.email                   AS "user_email?",
-                ue.created_at              AS "user_email_created_at?",
-                ue.confirmed_at            AS "user_email_confirmed_at?"
             FROM
                 oauth2_authorization_grants og
             LEFT JOIN oauth2_sessions os
@@ -455,8 +429,6 @@ pub async fn get_grant_by_id(
               USING (user_id)
             LEFT JOIN user_session_authentications usa
               USING (user_session_id)
-            LEFT JOIN user_emails ue
-              ON ue.user_email_id = u.primary_user_email_id
 
             WHERE og.oauth2_authorization_grant_id = $1
 
@@ -508,12 +480,9 @@ pub async fn lookup_grant_by_code(
                 us.created_at              AS "user_session_created_at?",
                  u.user_id                 AS "user_id?",
                  u.username                AS "user_username?",
+                 u.primary_user_email_id   AS "user_primary_user_email_id?",
                 usa.user_session_authentication_id AS "user_session_last_authentication_id?",
-                usa.created_at             AS "user_session_last_authentication_created_at?",
+                usa.created_at             AS "user_session_last_authentication_created_at?"
-                ue.user_email_id           AS "user_email_id?",
-                ue.email                   AS "user_email?",
-                ue.created_at              AS "user_email_created_at?",
-                ue.confirmed_at            AS "user_email_confirmed_at?"
             FROM
                 oauth2_authorization_grants og
             LEFT JOIN oauth2_sessions os
@@ -524,8 +493,6 @@ pub async fn lookup_grant_by_code(
               USING (user_id)
             LEFT JOIN user_session_authentications usa
               USING (user_session_id)
-            LEFT JOIN user_emails ue
-              ON ue.user_email_id = u.primary_user_email_id
 
             WHERE og.authorization_code = $1
 

crates/storage/src/oauth2/refresh_token.rs

@@ -13,9 +13,7 @@
 // limitations under the License.
 
 use chrono::{DateTime, Utc};
-use mas_data_model::{
+use mas_data_model::{AccessToken, Authentication, BrowserSession, RefreshToken, Session, User};
-    AccessToken, Authentication, BrowserSession, RefreshToken, Session, User, UserEmail,
-};
 use rand::Rng;
 use sqlx::{PgConnection, PgExecutor};
 use ulid::Ulid;
@@ -87,12 +85,9 @@ struct OAuth2RefreshTokenLookup {
     user_session_created_at: DateTime<Utc>,
     user_id: Uuid,
     user_username: String,
+    user_primary_user_email_id: Option<Uuid>,
     user_session_last_authentication_id: Option<Uuid>,
     user_session_last_authentication_created_at: Option<DateTime<Utc>>,
-    user_email_id: Option<Uuid>,
-    user_email: Option<String>,
-    user_email_created_at: Option<DateTime<Utc>>,
-    user_email_confirmed_at: Option<DateTime<Utc>>,
 }
 
 #[tracing::instrument(skip_all, err)]
@@ -119,12 +114,9 @@ pub async fn lookup_active_refresh_token(
                 us.created_at        AS "user_session_created_at!",
                  u.user_id           AS "user_id!",
                  u.username          AS "user_username!",
+                 u.primary_user_email_id AS "user_primary_user_email_id",
                 usa.user_session_authentication_id AS "user_session_last_authentication_id?",
-                usa.created_at       AS "user_session_last_authentication_created_at?",
+                usa.created_at       AS "user_session_last_authentication_created_at?"
-                ue.user_email_id     AS "user_email_id?",
-                ue.email             AS "user_email?",
-                ue.created_at        AS "user_email_created_at?",
-                ue.confirmed_at      AS "user_email_confirmed_at?"
             FROM oauth2_refresh_tokens rt
             INNER JOIN oauth2_sessions os
               USING (oauth2_session_id)
@@ -190,32 +182,11 @@ pub async fn lookup_active_refresh_token(
         })?;
 
     let user_id = Ulid::from(res.user_id);
-    let primary_email = match (
-        res.user_email_id,
-        res.user_email,
-        res.user_email_created_at,
-        res.user_email_confirmed_at,
-    ) {
-        (Some(id), Some(email), Some(created_at), confirmed_at) => Some(UserEmail {
-            id: id.into(),
-            email,
-            created_at,
-            confirmed_at,
-        }),
-        (None, None, None, None) => None,
-        _ => {
-            return Err(DatabaseInconsistencyError::on("users")
-                .column("primary_user_email_id")
-                .row(user_id)
-                .into())
-        }
-    };
-
     let user = User {
         id: user_id,
         username: res.user_username,
         sub: user_id.to_string(),
-        primary_email,
+        primary_user_email_id: res.user_primary_user_email_id.map(Into::into),
     };
 
     let last_authentication = match (

crates/storage/src/repository.rs

@@ -14,9 +14,12 @@
 
 use sqlx::{PgConnection, Postgres, Transaction};
 
-use crate::upstream_oauth2::{
+use crate::{
+    upstream_oauth2::{
         PgUpstreamOAuthLinkRepository, PgUpstreamOAuthProviderRepository,
         PgUpstreamOAuthSessionRepository,
+    },
+    user::{PgUserEmailRepository, PgUserRepository},
 };
 
 pub trait Repository {
@@ -32,15 +35,27 @@ pub trait Repository {
     where
         Self: 'c;
 
+    type UserRepository<'c>
+    where
+        Self: 'c;
+
+    type UserEmailRepository<'c>
+    where
+        Self: 'c;
+
     fn upstream_oauth_link(&mut self) -> Self::UpstreamOAuthLinkRepository<'_>;
     fn upstream_oauth_provider(&mut self) -> Self::UpstreamOAuthProviderRepository<'_>;
     fn upstream_oauth_session(&mut self) -> Self::UpstreamOAuthSessionRepository<'_>;
+    fn user(&mut self) -> Self::UserRepository<'_>;
+    fn user_email(&mut self) -> Self::UserEmailRepository<'_>;
 }
 
 impl Repository for PgConnection {
     type UpstreamOAuthLinkRepository<'c> = PgUpstreamOAuthLinkRepository<'c> where Self: 'c;
     type UpstreamOAuthProviderRepository<'c> = PgUpstreamOAuthProviderRepository<'c> where Self: 'c;
     type UpstreamOAuthSessionRepository<'c> = PgUpstreamOAuthSessionRepository<'c> where Self: 'c;
+    type UserRepository<'c> = PgUserRepository<'c> where Self: 'c;
+    type UserEmailRepository<'c> = PgUserEmailRepository<'c> where Self: 'c;
 
     fn upstream_oauth_link(&mut self) -> Self::UpstreamOAuthLinkRepository<'_> {
         PgUpstreamOAuthLinkRepository::new(self)
@@ -53,12 +68,22 @@ impl Repository for PgConnection {
     fn upstream_oauth_session(&mut self) -> Self::UpstreamOAuthSessionRepository<'_> {
         PgUpstreamOAuthSessionRepository::new(self)
     }
+
+    fn user(&mut self) -> Self::UserRepository<'_> {
+        PgUserRepository::new(self)
+    }
+
+    fn user_email(&mut self) -> Self::UserEmailRepository<'_> {
+        PgUserEmailRepository::new(self)
+    }
 }
 
 impl<'t> Repository for Transaction<'t, Postgres> {
     type UpstreamOAuthLinkRepository<'c> = PgUpstreamOAuthLinkRepository<'c> where Self: 'c;
     type UpstreamOAuthProviderRepository<'c> = PgUpstreamOAuthProviderRepository<'c> where Self: 'c;
     type UpstreamOAuthSessionRepository<'c> = PgUpstreamOAuthSessionRepository<'c> where Self: 'c;
+    type UserRepository<'c> = PgUserRepository<'c> where Self: 'c;
+    type UserEmailRepository<'c> = PgUserEmailRepository<'c> where Self: 'c;
 
     fn upstream_oauth_link(&mut self) -> Self::UpstreamOAuthLinkRepository<'_> {
         PgUpstreamOAuthLinkRepository::new(self)
@@ -71,4 +96,12 @@ impl<'t> Repository for Transaction<'t, Postgres> {
     fn upstream_oauth_session(&mut self) -> Self::UpstreamOAuthSessionRepository<'_> {
         PgUpstreamOAuthSessionRepository::new(self)
     }
+
+    fn user(&mut self) -> Self::UserRepository<'_> {
+        PgUserRepository::new(self)
+    }
+
+    fn user_email(&mut self) -> Self::UserEmailRepository<'_> {
+        PgUserEmailRepository::new(self)
+    }
 }

crates/storage/src/user/email.rs

@@ -0,0 +1,555 @@
+// Copyright 2022 The Matrix.org Foundation C.I.C.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use async_trait::async_trait;
+use chrono::{DateTime, Utc};
+use mas_data_model::{User, UserEmail, UserEmailVerification, UserEmailVerificationState};
+use rand::RngCore;
+use sqlx::{PgConnection, QueryBuilder};
+use ulid::Ulid;
+use uuid::Uuid;
+
+use crate::{
+    pagination::{process_page, Page, QueryBuilderExt},
+    Clock, DatabaseError, DatabaseInconsistencyError, LookupResultExt,
+};
+
+#[async_trait]
+pub trait UserEmailRepository: Send + Sync {
+    type Error;
+
+    async fn lookup(&mut self, id: Ulid) -> Result<Option<UserEmail>, Self::Error>;
+    async fn find(&mut self, user: &User, email: &str) -> Result<Option<UserEmail>, Self::Error>;
+    async fn get_primary(&mut self, user: &User) -> Result<Option<UserEmail>, Self::Error>;
+
+    async fn all(&mut self, user: &User) -> Result<Vec<UserEmail>, Self::Error>;
+    async fn list_paginated(
+        &mut self,
+        user: &User,
+        before: Option<Ulid>,
+        after: Option<Ulid>,
+        first: Option<usize>,
+        last: Option<usize>,
+    ) -> Result<Page<UserEmail>, Self::Error>;
+    async fn count(&mut self, user: &User) -> Result<usize, Self::Error>;
+
+    async fn add(
+        &mut self,
+        rng: &mut (dyn RngCore + Send),
+        clock: &Clock,
+        user: &User,
+        email: String,
+    ) -> Result<UserEmail, Self::Error>;
+    async fn remove(&mut self, user_email: UserEmail) -> Result<(), Self::Error>;
+
+    async fn mark_as_verified(
+        &mut self,
+        clock: &Clock,
+        user_email: UserEmail,
+    ) -> Result<UserEmail, Self::Error>;
+
+    async fn set_as_primary(&mut self, user_email: &UserEmail) -> Result<(), Self::Error>;
+
+    async fn add_verification_code(
+        &mut self,
+        rng: &mut (dyn RngCore + Send),
+        clock: &Clock,
+        user_email: &UserEmail,
+        max_age: chrono::Duration,
+        code: String,
+    ) -> Result<UserEmailVerification, Self::Error>;
+
+    async fn find_verification_code(
+        &mut self,
+        clock: &Clock,
+        user_email: &UserEmail,
+        code: &str,
+    ) -> Result<Option<UserEmailVerification>, Self::Error>;
+
+    async fn consume_verification_code(
+        &mut self,
+        clock: &Clock,
+        verification: UserEmailVerification,
+    ) -> Result<UserEmailVerification, Self::Error>;
+}
+
+pub struct PgUserEmailRepository<'c> {
+    conn: &'c mut PgConnection,
+}
+
+impl<'c> PgUserEmailRepository<'c> {
+    pub fn new(conn: &'c mut PgConnection) -> Self {
+        Self { conn }
+    }
+}
+
+#[derive(Debug, Clone, sqlx::FromRow)]
+struct UserEmailLookup {
+    user_email_id: Uuid,
+    user_id: Uuid,
+    email: String,
+    created_at: DateTime<Utc>,
+    confirmed_at: Option<DateTime<Utc>>,
+}
+
+impl From<UserEmailLookup> for UserEmail {
+    fn from(e: UserEmailLookup) -> UserEmail {
+        UserEmail {
+            id: e.user_email_id.into(),
+            user_id: e.user_id.into(),
+            email: e.email,
+            created_at: e.created_at,
+            confirmed_at: e.confirmed_at,
+        }
+    }
+}
+
+struct UserEmailConfirmationCodeLookup {
+    user_email_confirmation_code_id: Uuid,
+    user_email_id: Uuid,
+    code: String,
+    created_at: DateTime<Utc>,
+    expires_at: DateTime<Utc>,
+    consumed_at: Option<DateTime<Utc>>,
+}
+
+impl UserEmailConfirmationCodeLookup {
+    fn into_verification(self, clock: &Clock) -> UserEmailVerification {
+        let now = clock.now();
+        let state = if let Some(when) = self.consumed_at {
+            UserEmailVerificationState::AlreadyUsed { when }
+        } else if self.expires_at < now {
+            UserEmailVerificationState::Expired {
+                when: self.expires_at,
+            }
+        } else {
+            UserEmailVerificationState::Valid
+        };
+
+        UserEmailVerification {
+            id: self.user_email_confirmation_code_id.into(),
+            user_email_id: self.user_email_id.into(),
+            code: self.code,
+            state,
+            created_at: self.created_at,
+        }
+    }
+}
+
+#[async_trait]
+impl<'c> UserEmailRepository for PgUserEmailRepository<'c> {
+    type Error = DatabaseError;
+
+    #[tracing::instrument(
+        skip_all,
+        fields(user_email.id = %id),
+        err,
+    )]
+    async fn lookup(&mut self, id: Ulid) -> Result<Option<UserEmail>, Self::Error> {
+        let res = sqlx::query_as!(
+            UserEmailLookup,
+            r#"
+                SELECT user_email_id
+                     , user_id
+                     , email
+                     , created_at
+                     , confirmed_at
+                FROM user_emails
+
+                WHERE user_email_id = $1
+            "#,
+            Uuid::from(id),
+        )
+        .fetch_one(&mut *self.conn)
+        .await
+        .to_option()?;
+
+        let Some(user_email) = res else { return Ok(None) };
+
+        Ok(Some(user_email.into()))
+    }
+
+    #[tracing::instrument(
+        skip_all,
+        fields(%user.id, user_email.email = email),
+        err,
+    )]
+    async fn find(&mut self, user: &User, email: &str) -> Result<Option<UserEmail>, Self::Error> {
+        let res = sqlx::query_as!(
+            UserEmailLookup,
+            r#"
+                SELECT user_email_id
+                     , user_id
+                     , email
+                     , created_at
+                     , confirmed_at
+                FROM user_emails
+
+                WHERE user_id = $1 AND email = $2
+            "#,
+            Uuid::from(user.id),
+            email,
+        )
+        .fetch_one(&mut *self.conn)
+        .await
+        .to_option()?;
+
+        let Some(user_email) = res else { return Ok(None) };
+
+        Ok(Some(user_email.into()))
+    }
+
+    #[tracing::instrument(
+        skip_all,
+        fields(%user.id),
+        err,
+    )]
+    async fn get_primary(&mut self, user: &User) -> Result<Option<UserEmail>, Self::Error> {
+        let Some(id) = user.primary_user_email_id else { return Ok(None) };
+
+        let user_email = self.lookup(id).await?.ok_or_else(|| {
+            DatabaseInconsistencyError::on("users")
+                .column("primary_user_email_id")
+                .row(user.id)
+        })?;
+
+        Ok(Some(user_email))
+    }
+
+    #[tracing::instrument(
+        skip_all,
+        fields(%user.id),
+        err,
+    )]
+    async fn all(&mut self, user: &User) -> Result<Vec<UserEmail>, Self::Error> {
+        let res = sqlx::query_as!(
+            UserEmailLookup,
+            r#"
+                SELECT user_email_id
+                     , user_id
+                     , email
+                     , created_at
+                     , confirmed_at
+                FROM user_emails
+
+                WHERE user_id = $1
+
+                ORDER BY email ASC
+            "#,
+            Uuid::from(user.id),
+        )
+        .fetch_all(&mut *self.conn)
+        .await?;
+
+        Ok(res.into_iter().map(Into::into).collect())
+    }
+
+    #[tracing::instrument(
+        skip_all,
+        fields(%user.id),
+        err,
+    )]
+    async fn list_paginated(
+        &mut self,
+        user: &User,
+        before: Option<Ulid>,
+        after: Option<Ulid>,
+        first: Option<usize>,
+        last: Option<usize>,
+    ) -> Result<Page<UserEmail>, DatabaseError> {
+        let mut query = QueryBuilder::new(
+            r#"
+                SELECT user_email_id
+                     , user_id
+                     , email
+                     , created_at
+                     , confirmed_at
+                FROM user_emails
+            "#,
+        );
+
+        query
+            .push(" WHERE user_id = ")
+            .push_bind(Uuid::from(user.id))
+            .generate_pagination("ue.user_email_id", before, after, first, last)?;
+
+        let edges: Vec<UserEmailLookup> = query.build_query_as().fetch_all(&mut *self.conn).await?;
+
+        let (has_previous_page, has_next_page, edges) = process_page(edges, first, last)?;
+
+        let edges = edges.into_iter().map(Into::into).collect();
+
+        Ok(Page {
+            has_next_page,
+            has_previous_page,
+            edges,
+        })
+    }
+
+    #[tracing::instrument(
+        skip_all,
+        fields(%user.id),
+        err,
+    )]
+    async fn count(&mut self, user: &User) -> Result<usize, Self::Error> {
+        let res = sqlx::query_scalar!(
+            r#"
+                SELECT COUNT(*)
+                FROM user_emails
+                WHERE user_id = $1
+            "#,
+            Uuid::from(user.id),
+        )
+        .fetch_one(&mut *self.conn)
+        .await?;
+
+        let res = res.unwrap_or_default();
+
+        Ok(res
+            .try_into()
+            .map_err(DatabaseError::to_invalid_operation)?)
+    }
+
+    #[tracing::instrument(
+        skip_all,
+        fields(
+            %user.id,
+            user_email.id,
+            user_email.email = email,
+        ),
+        err,
+    )]
+    async fn add(
+        &mut self,
+        rng: &mut (dyn RngCore + Send),
+        clock: &Clock,
+        user: &User,
+        email: String,
+    ) -> Result<UserEmail, Self::Error> {
+        let created_at = clock.now();
+        let id = Ulid::from_datetime_with_source(created_at.into(), rng);
+        tracing::Span::current().record("user_email.id", tracing::field::display(id));
+
+        sqlx::query!(
+            r#"
+                INSERT INTO user_emails (user_email_id, user_id, email, created_at)
+                VALUES ($1, $2, $3, $4)
+            "#,
+            Uuid::from(id),
+            Uuid::from(user.id),
+            &email,
+            created_at,
+        )
+        .execute(&mut *self.conn)
+        .await?;
+
+        Ok(UserEmail {
+            id,
+            user_id: user.id,
+            email,
+            created_at,
+            confirmed_at: None,
+        })
+    }
+
+    #[tracing::instrument(
+        skip_all,
+        fields(
+            user.id = %user_email.user_id,
+            %user_email.id,
+            %user_email.email,
+        ),
+        err,
+    )]
+    async fn remove(&mut self, user_email: UserEmail) -> Result<(), Self::Error> {
+        sqlx::query!(
+            r#"
+                DELETE FROM user_emails
+                WHERE user_email_id = $1
+            "#,
+            Uuid::from(user_email.id),
+        )
+        .execute(&mut *self.conn)
+        .await?;
+
+        Ok(())
+    }
+
+    async fn mark_as_verified(
+        &mut self,
+        clock: &Clock,
+        mut user_email: UserEmail,
+    ) -> Result<UserEmail, Self::Error> {
+        let confirmed_at = clock.now();
+        sqlx::query!(
+            r#"
+                UPDATE user_emails
+                SET confirmed_at = $2
+                WHERE user_email_id = $1
+            "#,
+            Uuid::from(user_email.id),
+            confirmed_at,
+        )
+        .execute(&mut *self.conn)
+        .await?;
+
+        user_email.confirmed_at = Some(confirmed_at);
+        Ok(user_email)
+    }
+
+    async fn set_as_primary(&mut self, user_email: &UserEmail) -> Result<(), Self::Error> {
+        sqlx::query!(
+            r#"
+                UPDATE users
+                SET primary_user_email_id = user_emails.user_email_id
+                FROM user_emails
+                WHERE user_emails.user_email_id = $1
+                  AND users.user_id = user_emails.user_id
+            "#,
+            Uuid::from(user_email.id),
+        )
+        .execute(&mut *self.conn)
+        .await?;
+
+        Ok(())
+    }
+
+    #[tracing::instrument(
+        skip_all,
+        fields(
+            %user_email.id,
+            %user_email.email,
+            user_email_verification.id,
+            user_email_verification.code = code,
+        ),
+        err,
+    )]
+    async fn add_verification_code(
+        &mut self,
+        rng: &mut (dyn RngCore + Send),
+        clock: &Clock,
+        user_email: &UserEmail,
+        max_age: chrono::Duration,
+        code: String,
+    ) -> Result<UserEmailVerification, Self::Error> {
+        let created_at = clock.now();
+        let id = Ulid::from_datetime_with_source(created_at.into(), rng);
+        tracing::Span::current().record("user_email_confirmation.id", tracing::field::display(id));
+        let expires_at = created_at + max_age;
+
+        sqlx::query!(
+            r#"
+                INSERT INTO user_email_confirmation_codes
+                  (user_email_confirmation_code_id, user_email_id, code, created_at, expires_at)
+                VALUES ($1, $2, $3, $4, $5)
+            "#,
+            Uuid::from(id),
+            Uuid::from(user_email.id),
+            code,
+            created_at,
+            expires_at,
+        )
+        .execute(&mut *self.conn)
+        .await?;
+
+        let verification = UserEmailVerification {
+            id,
+            user_email_id: user_email.id,
+            code,
+            created_at,
+            state: UserEmailVerificationState::Valid,
+        };
+
+        Ok(verification)
+    }
+
+    #[tracing::instrument(
+        skip_all,
+        fields(
+            %user_email.id,
+            user.id = %user_email.user_id,
+        ),
+        err,
+    )]
+    async fn find_verification_code(
+        &mut self,
+        clock: &Clock,
+        user_email: &UserEmail,
+        code: &str,
+    ) -> Result<Option<UserEmailVerification>, Self::Error> {
+        let res = sqlx::query_as!(
+            UserEmailConfirmationCodeLookup,
+            r#"
+                SELECT user_email_confirmation_code_id
+                     , user_email_id
+                     , code
+                     , created_at
+                     , expires_at
+                     , consumed_at
+                FROM user_email_confirmation_codes
+                WHERE code = $1
+                  AND user_email_id = $2
+            "#,
+            code,
+            Uuid::from(user_email.id),
+        )
+        .fetch_one(&mut *self.conn)
+        .await
+        .to_option()?;
+
+        let Some(res) = res else { return Ok(None) };
+
+        Ok(Some(res.into_verification(clock)))
+    }
+
+    #[tracing::instrument(
+        skip_all,
+        fields(
+            %user_email_verification.id,
+            user_email.id = %user_email_verification.user_email_id,
+        ),
+        err,
+    )]
+    async fn consume_verification_code(
+        &mut self,
+        clock: &Clock,
+        mut user_email_verification: UserEmailVerification,
+    ) -> Result<UserEmailVerification, Self::Error> {
+        if !matches!(
+            user_email_verification.state,
+            UserEmailVerificationState::Valid
+        ) {
+            return Err(DatabaseError::invalid_operation());
+        }
+
+        let consumed_at = clock.now();
+
+        sqlx::query!(
+            r#"
+                UPDATE user_email_confirmation_codes
+                SET consumed_at = $2
+                WHERE user_email_confirmation_code_id = $1
+            "#,
+            Uuid::from(user_email_verification.id),
+            consumed_at
+        )
+        .execute(&mut *self.conn)
+        .await?;
+
+        user_email_verification.state =
+            UserEmailVerificationState::AlreadyUsed { when: consumed_at };
+
+        Ok(user_email_verification)
+    }
+}

crates/storage/src/user/mod.rs

@@ -12,13 +12,11 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use async_trait::async_trait;
 use chrono::{DateTime, Utc};
-use mas_data_model::{
+use mas_data_model::{Authentication, BrowserSession, User};
-    Authentication, BrowserSession, User, UserEmail, UserEmailVerification,
+use rand::{Rng, RngCore};
-    UserEmailVerificationState,
+use sqlx::{PgConnection, PgExecutor, QueryBuilder};
-};
-use rand::Rng;
-use sqlx::{PgExecutor, QueryBuilder};
 use tracing::{info_span, Instrument};
 use ulid::Ulid;
 use uuid::Uuid;
@@ -29,35 +27,188 @@ use crate::{
 };
 
 mod authentication;
+mod email;
 mod password;
 
 pub use self::{
     authentication::{authenticate_session_with_password, authenticate_session_with_upstream},
+    email::{PgUserEmailRepository, UserEmailRepository},
     password::{add_user_password, lookup_user_password},
 };
 
+#[async_trait]
+pub trait UserRepository: Send + Sync {
+    type Error;
+
+    async fn lookup(&mut self, id: Ulid) -> Result<Option<User>, Self::Error>;
+    async fn find_by_username(&mut self, username: &str) -> Result<Option<User>, Self::Error>;
+    async fn add(
+        &mut self,
+        rng: &mut (dyn RngCore + Send),
+        clock: &Clock,
+        username: String,
+    ) -> Result<User, Self::Error>;
+    async fn exists(&mut self, username: &str) -> Result<bool, Self::Error>;
+}
+
+pub struct PgUserRepository<'c> {
+    conn: &'c mut PgConnection,
+}
+
+impl<'c> PgUserRepository<'c> {
+    pub fn new(conn: &'c mut PgConnection) -> Self {
+        Self { conn }
+    }
+}
+
 #[derive(Debug, Clone)]
 struct UserLookup {
     user_id: Uuid,
-    user_username: String,
+    username: String,
-    user_email_id: Option<Uuid>,
+    primary_user_email_id: Option<Uuid>,
-    user_email: Option<String>,
+
-    user_email_created_at: Option<DateTime<Utc>>,
+    #[allow(dead_code)]
-    user_email_confirmed_at: Option<DateTime<Utc>>,
+    created_at: DateTime<Utc>,
+}
+
+impl From<UserLookup> for User {
+    fn from(value: UserLookup) -> Self {
+        let id = value.user_id.into();
+        Self {
+            id,
+            username: value.username,
+            sub: id.to_string(),
+            primary_user_email_id: value.primary_user_email_id.map(Into::into),
+        }
+    }
+}
+
+#[async_trait]
+impl<'c> UserRepository for PgUserRepository<'c> {
+    type Error = DatabaseError;
+
+    #[tracing::instrument(
+        skip_all,
+        fields(user.id = %id),
+        err,
+    )]
+    async fn lookup(&mut self, id: Ulid) -> Result<Option<User>, Self::Error> {
+        let res = sqlx::query_as!(
+            UserLookup,
+            r#"
+                SELECT user_id
+                     , username
+                     , primary_user_email_id
+                     , created_at
+                FROM users
+                WHERE user_id = $1
+            "#,
+            Uuid::from(id),
+        )
+        .fetch_one(&mut *self.conn)
+        .await
+        .to_option()?;
+
+        let Some(res) = res else { return Ok(None) };
+
+        Ok(Some(res.into()))
+    }
+
+    #[tracing::instrument(
+        skip_all,
+        fields(user.username = username),
+        err,
+    )]
+    async fn find_by_username(&mut self, username: &str) -> Result<Option<User>, Self::Error> {
+        let res = sqlx::query_as!(
+            UserLookup,
+            r#"
+                SELECT user_id
+                     , username
+                     , primary_user_email_id
+                     , created_at
+                FROM users
+                WHERE username = $1
+            "#,
+            username,
+        )
+        .fetch_one(&mut *self.conn)
+        .await
+        .to_option()?;
+
+        let Some(res) = res else { return Ok(None) };
+
+        Ok(Some(res.into()))
+    }
+
+    #[tracing::instrument(
+        skip_all,
+        fields(
+            user.username = username,
+            user.id,
+        ),
+        err,
+    )]
+    async fn add(
+        &mut self,
+        rng: &mut (dyn RngCore + Send),
+        clock: &Clock,
+        username: String,
+    ) -> Result<User, Self::Error> {
+        let created_at = clock.now();
+        let id = Ulid::from_datetime_with_source(created_at.into(), rng);
+        tracing::Span::current().record("user.id", tracing::field::display(id));
+
+        sqlx::query!(
+            r#"
+                INSERT INTO users (user_id, username, created_at)
+                VALUES ($1, $2, $3)
+            "#,
+            Uuid::from(id),
+            username,
+            created_at,
+        )
+        .execute(&mut *self.conn)
+        .await?;
+
+        Ok(User {
+            id,
+            username,
+            sub: id.to_string(),
+            primary_user_email_id: None,
+        })
+    }
+
+    #[tracing::instrument(
+        skip_all,
+        fields(user.username = username),
+        err,
+    )]
+    async fn exists(&mut self, username: &str) -> Result<bool, Self::Error> {
+        let exists = sqlx::query_scalar!(
+            r#"
+                SELECT EXISTS(
+                    SELECT 1 FROM users WHERE username = $1
+                ) AS "exists!"
+            "#,
+            username
+        )
+        .fetch_one(&mut *self.conn)
+        .await?;
+
+        Ok(exists)
+    }
 }
 
 #[derive(sqlx::FromRow)]
 struct SessionLookup {
     user_session_id: Uuid,
+    user_session_created_at: DateTime<Utc>,
     user_id: Uuid,
-    username: String,
+    user_username: String,
-    created_at: DateTime<Utc>,
+    user_primary_user_email_id: Option<Uuid>,
     last_authentication_id: Option<Uuid>,
     last_authd_at: Option<DateTime<Utc>>,
-    user_email_id: Option<Uuid>,
-    user_email: Option<String>,
-    user_email_created_at: Option<DateTime<Utc>>,
-    user_email_confirmed_at: Option<DateTime<Utc>>,
 }
 
 impl TryInto<BrowserSession> for SessionLookup {
@@ -65,31 +216,11 @@ impl TryInto<BrowserSession> for SessionLookup {
 
     fn try_into(self) -> Result<BrowserSession, Self::Error> {
         let id = Ulid::from(self.user_id);
-        let primary_email = match (
-            self.user_email_id,
-            self.user_email,
-            self.user_email_created_at,
-            self.user_email_confirmed_at,
-        ) {
-            (Some(id), Some(email), Some(created_at), confirmed_at) => Some(UserEmail {
-                id: id.into(),
-                email,
-                created_at,
-                confirmed_at,
-            }),
-            (None, None, None, None) => None,
-            _ => {
-                return Err(DatabaseInconsistencyError::on("users")
-                    .column("primary_user_email_id")
-                    .row(id))
-            }
-        };
-
         let user = User {
             id,
-            username: self.username,
+            username: self.user_username,
             sub: id.to_string(),
-            primary_email,
+            primary_user_email_id: self.user_primary_user_email_id.map(Into::into),
         };
 
         let last_authentication = match (self.last_authentication_id, self.last_authd_at) {
@@ -108,7 +239,7 @@ impl TryInto<BrowserSession> for SessionLookup {
         Ok(BrowserSession {
             id: self.user_session_id.into(),
             user,
-            created_at: self.created_at,
+            created_at: self.user_session_created_at,
             last_authentication,
         })
     }
@@ -126,24 +257,18 @@ pub async fn lookup_active_session(
     let res = sqlx::query_as!(
         SessionLookup,
         r#"
-            SELECT
+            SELECT s.user_session_id
-                s.user_session_id,
+                 , s.created_at                     AS "user_session_created_at"
-                u.user_id,
+                 , u.user_id
-                u.username,
+                 , u.username                       AS "user_username"
-                s.created_at,
+                 , u.primary_user_email_id          AS "user_primary_user_email_id"
-                a.user_session_authentication_id AS "last_authentication_id?",
+                 , a.user_session_authentication_id AS "last_authentication_id?"
-                a.created_at                     AS "last_authd_at?",
+                 , a.created_at                     AS "last_authd_at?"
-                ue.user_email_id   AS "user_email_id?",
-                ue.email           AS "user_email?",
-                ue.created_at      AS "user_email_created_at?",
-                ue.confirmed_at    AS "user_email_confirmed_at?"
             FROM user_sessions s
             INNER JOIN users u
                 USING (user_id)
             LEFT JOIN user_session_authentications a
                 USING (user_session_id)
-            LEFT JOIN user_emails ue
-              ON ue.user_email_id = u.primary_user_email_id
             WHERE s.user_session_id = $1 AND s.finished_at IS NULL
             ORDER BY a.created_at DESC
             LIMIT 1
@@ -279,44 +404,6 @@ pub async fn count_active_sessions(
     Ok(res)
 }
 
-#[tracing::instrument(
-    skip_all,
-    fields(
-        user.username = username,
-        user.id,
-    ),
-    err,
-)]
-pub async fn add_user(
-    executor: impl PgExecutor<'_>,
-    mut rng: impl Rng + Send,
-    clock: &Clock,
-    username: &str,
-) -> Result<User, sqlx::Error> {
-    let created_at = clock.now();
-    let id = Ulid::from_datetime_with_source(created_at.into(), &mut rng);
-    tracing::Span::current().record("user.id", tracing::field::display(id));
-
-    sqlx::query!(
-        r#"
-            INSERT INTO users (user_id, username, created_at)
-            VALUES ($1, $2, $3)
-        "#,
-        Uuid::from(id),
-        username,
-        created_at,
-    )
-    .execute(executor)
-    .await?;
-
-    Ok(User {
-        id,
-        username: username.to_owned(),
-        sub: id.to_string(),
-        primary_email: None,
-    })
-}
-
 #[tracing::instrument(
     skip_all,
     fields(%user_session.id),
@@ -343,662 +430,3 @@ pub async fn end_session(
 
     DatabaseError::ensure_affected_rows(&res, 1)
 }
-
-#[tracing::instrument(
-    skip_all,
-    fields(user.username = username),
-    err,
-)]
-pub async fn lookup_user_by_username(
-    executor: impl PgExecutor<'_>,
-    username: &str,
-) -> Result<Option<User>, DatabaseError> {
-    let res = sqlx::query_as!(
-        UserLookup,
-        r#"
-            SELECT
-                u.user_id,
-                u.username       AS user_username,
-                ue.user_email_id AS "user_email_id?",
-                ue.email         AS "user_email?",
-                ue.created_at    AS "user_email_created_at?",
-                ue.confirmed_at  AS "user_email_confirmed_at?"
-            FROM users u
-
-            LEFT JOIN user_emails ue
-              USING (user_id)
-
-            WHERE u.username = $1
-        "#,
-        username,
-    )
-    .fetch_one(executor)
-    .instrument(info_span!("Fetch user"))
-    .await
-    .to_option()?;
-
-    let Some(res) = res else { return Ok(None) };
-
-    let id = Ulid::from(res.user_id);
-    let primary_email = match (
-        res.user_email_id,
-        res.user_email,
-        res.user_email_created_at,
-        res.user_email_confirmed_at,
-    ) {
-        (Some(id), Some(email), Some(created_at), confirmed_at) => Some(UserEmail {
-            id: id.into(),
-            email,
-            created_at,
-            confirmed_at,
-        }),
-        (None, None, None, None) => None,
-        _ => {
-            return Err(DatabaseInconsistencyError::on("users")
-                .column("primary_user_email_id")
-                .row(id)
-                .into())
-        }
-    };
-
-    Ok(Some(User {
-        id,
-        username: res.user_username,
-        sub: id.to_string(),
-        primary_email,
-    }))
-}
-
-#[tracing::instrument(
-    skip_all,
-    fields(user.id = %id),
-    err,
-)]
-pub async fn lookup_user(executor: impl PgExecutor<'_>, id: Ulid) -> Result<User, DatabaseError> {
-    let res = sqlx::query_as!(
-        UserLookup,
-        r#"
-            SELECT
-                u.user_id,
-                u.username       AS user_username,
-                ue.user_email_id AS "user_email_id?",
-                ue.email         AS "user_email?",
-                ue.created_at    AS "user_email_created_at?",
-                ue.confirmed_at  AS "user_email_confirmed_at?"
-            FROM users u
-
-            LEFT JOIN user_emails ue
-              USING (user_id)
-
-            WHERE u.user_id = $1
-        "#,
-        Uuid::from(id),
-    )
-    .fetch_one(executor)
-    .instrument(info_span!("Fetch user"))
-    .await?;
-
-    let id = Ulid::from(res.user_id);
-    let primary_email = match (
-        res.user_email_id,
-        res.user_email,
-        res.user_email_created_at,
-        res.user_email_confirmed_at,
-    ) {
-        (Some(id), Some(email), Some(created_at), confirmed_at) => Some(UserEmail {
-            id: id.into(),
-            email,
-            created_at,
-            confirmed_at,
-        }),
-        (None, None, None, None) => None,
-        _ => {
-            return Err(DatabaseInconsistencyError::on("users")
-                .column("primary_user_email_id")
-                .row(id)
-                .into())
-        }
-    };
-
-    Ok(User {
-        id,
-        username: res.user_username,
-        sub: id.to_string(),
-        primary_email,
-    })
-}
-
-#[tracing::instrument(
-    skip_all,
-    fields(user.username = username),
-    err,
-)]
-pub async fn username_exists(
-    executor: impl PgExecutor<'_>,
-    username: &str,
-) -> Result<bool, sqlx::Error> {
-    sqlx::query_scalar!(
-        r#"
-            SELECT EXISTS(
-                SELECT 1 FROM users WHERE username = $1
-            ) AS "exists!"
-        "#,
-        username
-    )
-    .fetch_one(executor)
-    .await
-}
-
-#[derive(Debug, Clone, sqlx::FromRow)]
-struct UserEmailLookup {
-    user_email_id: Uuid,
-    user_email: String,
-    user_email_created_at: DateTime<Utc>,
-    user_email_confirmed_at: Option<DateTime<Utc>>,
-}
-
-impl From<UserEmailLookup> for UserEmail {
-    fn from(e: UserEmailLookup) -> UserEmail {
-        UserEmail {
-            id: e.user_email_id.into(),
-            email: e.user_email,
-            created_at: e.user_email_created_at,
-            confirmed_at: e.user_email_confirmed_at,
-        }
-    }
-}
-
-#[tracing::instrument(
-    skip_all,
-    fields(%user.id, %user.username),
-    err,
-)]
-pub async fn get_user_emails(
-    executor: impl PgExecutor<'_>,
-    user: &User,
-) -> Result<Vec<UserEmail>, sqlx::Error> {
-    let res = sqlx::query_as!(
-        UserEmailLookup,
-        r#"
-            SELECT
-                ue.user_email_id,
-                ue.email        AS "user_email",
-                ue.created_at   AS "user_email_created_at",
-                ue.confirmed_at AS "user_email_confirmed_at"
-            FROM user_emails ue
-
-            WHERE ue.user_id = $1
-
-            ORDER BY ue.email ASC
-        "#,
-        Uuid::from(user.id),
-    )
-    .fetch_all(executor)
-    .instrument(info_span!("Fetch user emails"))
-    .await?;
-
-    Ok(res.into_iter().map(Into::into).collect())
-}
-
-#[tracing::instrument(
-    skip_all,
-    fields(%user.id, %user.username),
-    err,
-)]
-pub async fn count_user_emails(
-    executor: impl PgExecutor<'_>,
-    user: &User,
-) -> Result<i64, sqlx::Error> {
-    let res = sqlx::query_scalar!(
-        r#"
-            SELECT COUNT(*)
-            FROM user_emails ue
-            WHERE ue.user_id = $1
-        "#,
-        Uuid::from(user.id),
-    )
-    .fetch_one(executor)
-    .instrument(info_span!("Count user emails"))
-    .await?;
-
-    Ok(res.unwrap_or_default())
-}
-
-#[tracing::instrument(
-    skip_all,
-    fields(%user.id, %user.username),
-    err,
-)]
-pub async fn get_paginated_user_emails(
-    executor: impl PgExecutor<'_>,
-    user: &User,
-    before: Option<Ulid>,
-    after: Option<Ulid>,
-    first: Option<usize>,
-    last: Option<usize>,
-) -> Result<(bool, bool, Vec<UserEmail>), DatabaseError> {
-    let mut query = QueryBuilder::new(
-        r#"
-            SELECT
-                ue.user_email_id,
-                ue.email        AS "user_email",
-                ue.created_at   AS "user_email_created_at",
-                ue.confirmed_at AS "user_email_confirmed_at"
-            FROM user_emails ue
-        "#,
-    );
-
-    query
-        .push(" WHERE ue.user_id = ")
-        .push_bind(Uuid::from(user.id))
-        .generate_pagination("ue.user_email_id", before, after, first, last)?;
-
-    let span = info_span!("Fetch paginated user sessions", db.statement = query.sql());
-    let page: Vec<UserEmailLookup> = query
-        .build_query_as()
-        .fetch_all(executor)
-        .instrument(span)
-        .await?;
-
-    let (has_previous_page, has_next_page, page) = process_page(page, first, last)?;
-
-    Ok((
-        has_previous_page,
-        has_next_page,
-        page.into_iter().map(Into::into).collect(),
-    ))
-}
-
-#[tracing::instrument(
-    skip_all,
-    fields(
-        %user.id,
-        %user.username,
-        user_email.id = %id,
-    ),
-    err,
-)]
-pub async fn get_user_email(
-    executor: impl PgExecutor<'_>,
-    user: &User,
-    id: Ulid,
-) -> Result<UserEmail, sqlx::Error> {
-    let res = sqlx::query_as!(
-        UserEmailLookup,
-        r#"
-            SELECT
-                ue.user_email_id,
-                ue.email        AS "user_email",
-                ue.created_at   AS "user_email_created_at",
-                ue.confirmed_at AS "user_email_confirmed_at"
-            FROM user_emails ue
-
-            WHERE ue.user_id = $1
-              AND ue.user_email_id = $2
-        "#,
-        Uuid::from(user.id),
-        Uuid::from(id),
-    )
-    .fetch_one(executor)
-    .instrument(info_span!("Fetch user emails"))
-    .await?;
-
-    Ok(res.into())
-}
-
-#[tracing::instrument(
-    skip_all,
-    fields(
-        %user.id,
-        %user.username,
-        user_email.id,
-        user_email.email = %email,
-    ),
-    err,
-)]
-pub async fn add_user_email(
-    executor: impl PgExecutor<'_>,
-    mut rng: impl Rng + Send,
-    clock: &Clock,
-    user: &User,
-    email: String,
-) -> Result<UserEmail, sqlx::Error> {
-    let created_at = clock.now();
-    let id = Ulid::from_datetime_with_source(created_at.into(), &mut rng);
-    tracing::Span::current().record("user_email.id", tracing::field::display(id));
-
-    sqlx::query!(
-        r#"
-            INSERT INTO user_emails (user_email_id, user_id, email, created_at)
-            VALUES ($1, $2, $3, $4)
-        "#,
-        Uuid::from(id),
-        Uuid::from(user.id),
-        &email,
-        created_at,
-    )
-    .execute(executor)
-    .instrument(info_span!("Add user email"))
-    .await?;
-
-    Ok(UserEmail {
-        id,
-        email,
-        created_at,
-        confirmed_at: None,
-    })
-}
-
-#[tracing::instrument(
-    skip_all,
-    fields(
-        %user_email.id,
-        %user_email.email,
-    ),
-    err,
-)]
-pub async fn set_user_email_as_primary(
-    executor: impl PgExecutor<'_>,
-    user_email: &UserEmail,
-) -> Result<(), sqlx::Error> {
-    sqlx::query!(
-        r#"
-            UPDATE users
-            SET primary_user_email_id = user_emails.user_email_id
-            FROM user_emails
-            WHERE user_emails.user_email_id = $1
-              AND users.user_id = user_emails.user_id
-        "#,
-        Uuid::from(user_email.id),
-    )
-    .execute(executor)
-    .instrument(info_span!("Add user email"))
-    .await?;
-
-    Ok(())
-}
-
-#[tracing::instrument(
-    skip_all,
-    fields(
-        %user_email.id,
-        %user_email.email,
-    ),
-    err,
-)]
-pub async fn remove_user_email(
-    executor: impl PgExecutor<'_>,
-    user_email: UserEmail,
-) -> Result<(), sqlx::Error> {
-    sqlx::query!(
-        r#"
-            DELETE FROM user_emails
-            WHERE user_emails.user_email_id = $1
-        "#,
-        Uuid::from(user_email.id),
-    )
-    .execute(executor)
-    .instrument(info_span!("Remove user email"))
-    .await?;
-
-    Ok(())
-}
-
-#[tracing::instrument(
-    skip_all,
-    fields(
-        %user.id,
-        user_email.email = email,
-    ),
-    err,
-)]
-pub async fn lookup_user_email(
-    executor: impl PgExecutor<'_>,
-    user: &User,
-    email: &str,
-) -> Result<Option<UserEmail>, sqlx::Error> {
-    let res = sqlx::query_as!(
-        UserEmailLookup,
-        r#"
-            SELECT
-                ue.user_email_id,
-                ue.email        AS "user_email",
-                ue.created_at   AS "user_email_created_at",
-                ue.confirmed_at AS "user_email_confirmed_at"
-            FROM user_emails ue
-
-            WHERE ue.user_id = $1
-              AND ue.email = $2
-        "#,
-        Uuid::from(user.id),
-        email,
-    )
-    .fetch_one(executor)
-    .instrument(info_span!("Lookup user email"))
-    .await
-    .to_option()?;
-
-    let Some(res) = res else { return Ok(None) };
-
-    Ok(Some(res.into()))
-}
-
-#[tracing::instrument(
-    skip_all,
-    fields(
-        %user.id,
-        user_email.id = %id,
-    ),
-    err,
-)]
-pub async fn lookup_user_email_by_id(
-    executor: impl PgExecutor<'_>,
-    user: &User,
-    id: Ulid,
-) -> Result<Option<UserEmail>, DatabaseError> {
-    let res = sqlx::query_as!(
-        UserEmailLookup,
-        r#"
-            SELECT
-                ue.user_email_id,
-                ue.email        AS "user_email",
-                ue.created_at   AS "user_email_created_at",
-                ue.confirmed_at AS "user_email_confirmed_at"
-            FROM user_emails ue
-
-            WHERE ue.user_id = $1
-              AND ue.user_email_id = $2
-        "#,
-        Uuid::from(user.id),
-        Uuid::from(id),
-    )
-    .fetch_one(executor)
-    .instrument(info_span!("Lookup user email"))
-    .await
-    .to_option()?;
-
-    let Some(res) = res else { return Ok(None) };
-
-    Ok(Some(res.into()))
-}
-
-#[tracing::instrument(
-    skip_all,
-    fields(%user_email.id),
-    err,
-)]
-pub async fn mark_user_email_as_verified(
-    executor: impl PgExecutor<'_>,
-    clock: &Clock,
-    mut user_email: UserEmail,
-) -> Result<UserEmail, sqlx::Error> {
-    let confirmed_at = clock.now();
-    sqlx::query!(
-        r#"
-            UPDATE user_emails
-            SET confirmed_at = $2
-            WHERE user_email_id = $1
-        "#,
-        Uuid::from(user_email.id),
-        confirmed_at,
-    )
-    .execute(executor)
-    .instrument(info_span!("Confirm user email"))
-    .await?;
-
-    user_email.confirmed_at = Some(confirmed_at);
-
-    Ok(user_email)
-}
-
-struct UserEmailConfirmationCodeLookup {
-    user_email_confirmation_code_id: Uuid,
-    code: String,
-    created_at: DateTime<Utc>,
-    expires_at: DateTime<Utc>,
-    consumed_at: Option<DateTime<Utc>>,
-}
-
-#[tracing::instrument(
-    skip_all,
-    fields(%user_email.id),
-    err,
-)]
-pub async fn lookup_user_email_verification_code(
-    executor: impl PgExecutor<'_>,
-    clock: &Clock,
-    user_email: UserEmail,
-    code: &str,
-) -> Result<Option<UserEmailVerification>, DatabaseError> {
-    let now = clock.now();
-
-    let res = sqlx::query_as!(
-        UserEmailConfirmationCodeLookup,
-        r#"
-            SELECT
-                ec.user_email_confirmation_code_id,
-                ec.code,
-                ec.created_at,
-                ec.expires_at,
-                ec.consumed_at
-            FROM user_email_confirmation_codes ec
-            WHERE ec.code = $1
-              AND ec.user_email_id = $2
-        "#,
-        code,
-        Uuid::from(user_email.id),
-    )
-    .fetch_one(executor)
-    .instrument(info_span!("Lookup user email verification"))
-    .await
-    .to_option()?;
-
-    let Some(res) = res else { return Ok(None) };
-
-    let state = if let Some(when) = res.consumed_at {
-        UserEmailVerificationState::AlreadyUsed { when }
-    } else if res.expires_at < now {
-        UserEmailVerificationState::Expired {
-            when: res.expires_at,
-        }
-    } else {
-        UserEmailVerificationState::Valid
-    };
-
-    Ok(Some(UserEmailVerification {
-        id: res.user_email_confirmation_code_id.into(),
-        code: res.code,
-        email: user_email,
-        state,
-        created_at: res.created_at,
-    }))
-}
-
-#[tracing::instrument(
-    skip_all,
-    fields(
-        %user_email_verification.id,
-    ),
-    err,
-)]
-pub async fn consume_email_verification(
-    executor: impl PgExecutor<'_>,
-    clock: &Clock,
-    mut user_email_verification: UserEmailVerification,
-) -> Result<UserEmailVerification, DatabaseError> {
-    if !matches!(
-        user_email_verification.state,
-        UserEmailVerificationState::Valid
-    ) {
-        return Err(DatabaseError::invalid_operation());
-    }
-
-    let consumed_at = clock.now();
-
-    sqlx::query!(
-        r#"
-            UPDATE user_email_confirmation_codes
-            SET consumed_at = $2
-            WHERE user_email_confirmation_code_id = $1
-        "#,
-        Uuid::from(user_email_verification.id),
-        consumed_at
-    )
-    .execute(executor)
-    .instrument(info_span!("Consume user email verification"))
-    .await?;
-
-    user_email_verification.state = UserEmailVerificationState::AlreadyUsed { when: consumed_at };
-
-    Ok(user_email_verification)
-}
-
-#[tracing::instrument(
-    skip_all,
-    fields(
-        %user_email.id,
-        %user_email.email,
-        user_email_confirmation.id,
-        user_email_confirmation.code = code,
-    ),
-    err,
-)]
-pub async fn add_user_email_verification_code(
-    executor: impl PgExecutor<'_>,
-    mut rng: impl Rng + Send,
-    clock: &Clock,
-    user_email: UserEmail,
-    max_age: chrono::Duration,
-    code: String,
-) -> Result<UserEmailVerification, sqlx::Error> {
-    let created_at = clock.now();
-    let id = Ulid::from_datetime_with_source(created_at.into(), &mut rng);
-    tracing::Span::current().record("user_email_confirmation.id", tracing::field::display(id));
-    let expires_at = created_at + max_age;
-
-    sqlx::query!(
-        r#"
-            INSERT INTO user_email_confirmation_codes
-              (user_email_confirmation_code_id, user_email_id, code, created_at, expires_at)
-            VALUES ($1, $2, $3, $4, $5)
-        "#,
-        Uuid::from(id),
-        Uuid::from(user_email.id),
-        code,
-        created_at,
-        expires_at,
-    )
-    .execute(executor)
-    .instrument(info_span!("Add user email verification code"))
-    .await?;
-
-    let verification = UserEmailVerification {
-        id,
-        email: user_email,
-        code,
-        created_at,
-        state: UserEmailVerificationState::Valid,
-    };
-
-    Ok(verification)
-}

crates/templates/src/context.rs

@@ -618,6 +618,7 @@ impl TemplateContext for EmailVerificationContext {
             .map(|user| {
                 let email = UserEmail {
                     id: Ulid::from_datetime_with_source(now.into(), rng),
+                    user_id: user.id,
                     email: "foobar@example.com".to_owned(),
                     created_at: now,
                     confirmed_at: None,
@@ -625,8 +626,8 @@ impl TemplateContext for EmailVerificationContext {
 
                 let verification = UserEmailVerification {
                     id: Ulid::from_datetime_with_source(now.into(), rng),
+                    user_email_id: email.id,
                     code: "123456".to_owned(),
-                    email,
                     created_at: now,
                     state: mas_data_model::UserEmailVerificationState::Valid,
                 };
@@ -684,6 +685,7 @@ impl TemplateContext for EmailVerificationPageContext {
     {
         let email = UserEmail {
             id: Ulid::from_datetime_with_source(now.into(), rng),
+            user_id: Ulid::from_datetime_with_source(now.into(), rng),
             email: "foobar@example.com".to_owned(),
             created_at: now,
             confirmed_at: None,

docs/development/database.md

@@ -35,6 +35,8 @@ Note that migrations are embedded in the final binary and can be run from the se
 
 ## Writing database interactions
 
+**TODO**: *This section is outdated.*
+
 A typical interaction with the database look like this:
 
 ```rust