mas_tasks/
recovery.rs

1// Copyright 2025, 2026 Element Creations Ltd.
2// Copyright 2024, 2025 New Vector Ltd.
3// Copyright 2024 The Matrix.org Foundation C.I.C.
4//
5// SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
6// Please see LICENSE files in the repository root for full details.
7
8use anyhow::Context;
9use async_trait::async_trait;
10use mas_email::{Address, Mailbox};
11use mas_i18n::DataLocale;
12use mas_storage::{
13    Pagination, RepositoryAccess,
14    queue::SendAccountRecoveryEmailsJob,
15    user::{UserEmailFilter, UserRecoveryRepository},
16};
17use mas_templates::{EmailRecoveryContext, TemplateContext};
18use rand::distributions::{Alphanumeric, DistString};
19use tracing::{error, info};
20
21use crate::{
22    State,
23    new_queue::{JobContext, JobError, RunnableJob},
24};
25
26/// Job to send account recovery emails for a given recovery session.
27#[async_trait]
28impl RunnableJob for SendAccountRecoveryEmailsJob {
29    #[tracing::instrument(
30        name = "job.send_account_recovery_email",
31        fields(
32            user_recovery_session.id = %self.user_recovery_session_id(),
33            user_recovery_session.email,
34        ),
35        skip_all,
36    )]
37    async fn run(&self, state: &State, _context: JobContext) -> Result<(), JobError> {
38        let clock = state.clock();
39        let mailer = state.mailer();
40        let url_builder = state.url_builder();
41        let mut rng = state.rng();
42        let mut repo = state.repository().await.map_err(JobError::retry)?;
43
44        let session = repo
45            .user_recovery()
46            .lookup_session(self.user_recovery_session_id())
47            .await
48            .map_err(JobError::retry)?
49            .context("User recovery session not found")
50            .map_err(JobError::fail)?;
51
52        tracing::Span::current().record("user_recovery_session.email", &session.email);
53
54        if session.consumed_at.is_some() {
55            info!("Recovery session already consumed, not sending email");
56            return Ok(());
57        }
58
59        let mut cursor = Pagination::first(50);
60
61        let lang: DataLocale = session
62            .locale
63            .parse()
64            .context("Invalid locale in database on recovery session")
65            .map_err(JobError::fail)?;
66
67        loop {
68            let page = repo
69                .user_email()
70                .list(UserEmailFilter::new().for_email(&session.email), cursor)
71                .await
72                .map_err(JobError::retry)?;
73
74            for edge in page.edges {
75                let ticket = Alphanumeric.sample_string(&mut rng, 32);
76
77                let ticket = repo
78                    .user_recovery()
79                    .add_ticket(&mut rng, clock, &session, &edge.node, ticket)
80                    .await
81                    .map_err(JobError::retry)?;
82
83                let user = repo
84                    .user()
85                    .lookup(edge.node.user_id)
86                    .await
87                    .map_err(JobError::retry)?
88                    .context("User not found")
89                    .map_err(JobError::fail)?;
90
91                let url = url_builder.account_recovery_link(ticket.ticket);
92
93                let address: Address = edge.node.email.parse().map_err(JobError::fail)?;
94                let mailbox = Mailbox::new(Some(user.username.clone()), address);
95
96                info!("Sending recovery email to {}", mailbox);
97                let context =
98                    EmailRecoveryContext::new(user, session.clone(), url).with_language(lang);
99
100                // XXX: we only log if the email fails to send, to avoid stopping the loop
101                if let Err(e) = mailer.send_recovery_email(mailbox, &context).await {
102                    error!(
103                        error = &e as &dyn std::error::Error,
104                        "Failed to send recovery email"
105                    );
106                }
107
108                cursor = cursor.after(edge.cursor);
109            }
110
111            if !page.has_next_page {
112                break;
113            }
114        }
115
116        repo.save().await.map_err(JobError::fail)?;
117
118        Ok(())
119    }
120}
mas_tasks/recovery.rs

mas_tasks/
recovery.rs
