Allow removing email addresses in bulk

crates/storage-pg/src/user/email.rs

@@ -365,6 +365,28 @@ impl UserEmailRepository for PgUserEmailRepository<'_> {
         Ok(())
     }
 
+    #[tracing::instrument(
+        name = "db.user_email.remove_bulk",
+        skip_all,
+        fields(
+            db.query.text,
+        ),
+        err,
+    )]
+    async fn remove_bulk(&mut self, filter: UserEmailFilter<'_>) -> Result<usize, Self::Error> {
+        let (sql, arguments) = Query::delete()
+            .from_table(UserEmails::Table)
+            .apply_filter(filter)
+            .build_sqlx(PostgresQueryBuilder);
+
+        let res = sqlx::query_with(&sql, arguments)
+            .traced()
+            .execute(&mut *self.conn)
+            .await?;
+
+        Ok(res.rows_affected().try_into().unwrap_or(usize::MAX))
+    }
+
     #[tracing::instrument(
         name = "db.user_email.add_authentication_for_session",
         skip_all,

crates/storage-pg/src/user/tests.rs

@@ -290,6 +290,21 @@ async fn test_user_email_repo(pool: PgPool) {
     repo.user_email().remove(user_email).await.unwrap();
     assert_eq!(repo.user_email().count(all).await.unwrap(), 0);
 
+    // Add a few emails
+    for i in 0..5 {
+        let email = format!("email{i}@example.com");
+        repo.user_email()
+            .add(&mut rng, &clock, &user, email)
+            .await
+            .unwrap();
+    }
+    assert_eq!(repo.user_email().count(all).await.unwrap(), 5);
+
+    // Try removing all the emails
+    let affected = repo.user_email().remove_bulk(all).await.unwrap();
+    assert_eq!(affected, 5);
+    assert_eq!(repo.user_email().count(all).await.unwrap(), 0);
+
     repo.save().await.unwrap();
 }
 

crates/storage/src/user/email.rs

@@ -164,6 +164,19 @@ pub trait UserEmailRepository: Send + Sync {
     /// Returns [`Self::Error`] if the underlying repository fails
     async fn remove(&mut self, user_email: UserEmail) -> Result<(), Self::Error>;
 
+    /// Delete all [`UserEmail`] with the given filter
+    ///
+    /// Returns the number of deleted [`UserEmail`]s
+    ///
+    /// # Parameters
+    ///
+    /// * `filter`: The filter parameters
+    ///
+    /// # Errors
+    ///
+    /// Returns [`Self::Error`] if the underlying repository fails
+    async fn remove_bulk(&mut self, filter: UserEmailFilter<'_>) -> Result<usize, Self::Error>;
+
     /// Add a new [`UserEmailAuthentication`] for a [`BrowserSession`]
     ///
     /// # Parameters
@@ -303,6 +316,8 @@ repository_impl!(UserEmailRepository:
     ) -> Result<UserEmail, Self::Error>;
     async fn remove(&mut self, user_email: UserEmail) -> Result<(), Self::Error>;
 
+    async fn remove_bulk(&mut self, filter: UserEmailFilter<'_>) -> Result<usize, Self::Error>;
+
     async fn add_authentication_for_session(
         &mut self,
         rng: &mut (dyn RngCore + Send),