headless_lms_server/programs/
email_deliver.rs

1use std::{error::Error as StdError, time::Duration};
2
3use crate::config::program_config::ProgramConfig;
4use crate::prelude::*;
5use crate::setup_tracing;
6use anyhow::{Context, Result};
7use chrono::{DateTime, Utc};
8use futures::{FutureExt, StreamExt};
9use headless_lms_models::email_deliveries::{
10    Email, EmailDeliveryErrorInsert, FETCH_LIMIT, RETRY_WINDOW_SECS, fetch_emails,
11    increment_retry_and_mark_non_retryable, increment_retry_and_schedule,
12    insert_email_delivery_error, mark_as_sent, maybe_purge_expired_recipient_addresses,
13};
14use headless_lms_models::email_layouts::{self, EmailLayout, ResolvedLayout};
15use headless_lms_models::email_templates::EmailTemplateType;
16use headless_lms_models::user_email_codes::UserEmailCodePurpose;
17use headless_lms_models::user_passwords::get_unused_reset_password_token_with_user_id;
18use headless_lms_utils::backoff;
19use headless_lms_utils::email_processor::{
20    self, EmailGutenbergBlock, EmailRenderInput, RenderedEmail,
21};
22use lettre::transport::smtp::Error as SmtpError;
23use lettre::transport::smtp::authentication::Credentials;
24use lettre::{
25    Message, SmtpTransport, Transport,
26    message::{MultiPart, SinglePart, header},
27};
28use once_cell::sync::Lazy;
29use secrecy::ExposeSecret;
30use sqlx::{Connection, PgConnection, PgPool};
31use std::collections::HashMap;
32use uuid::Uuid;
33const BATCH_SIZE: usize = FETCH_LIMIT as usize;
34
35/// How often the sender asks the model to purge expired recipient addresses. The model then rolls its
36/// own 1 in 10 chance, so a sweep lands roughly every ten hours.
37const RECIPIENT_ADDRESS_PURGE_INTERVAL: Duration = Duration::from_secs(60 * 60);
38
39const BASE_BACKOFF_SECS: i64 = 60;
40const MAX_BACKOFF_SECS: i64 = 24 * 60 * 60;
41const JITTER_SECS: i64 = 30;
42
43static SMTP_FROM: Lazy<String> = Lazy::new(|| {
44    ProgramConfig::required("SMTP_FROM").expect("No moocfi email found in the env variables.")
45});
46/// The same variable the server reads into `ApplicationConfiguration::base_url`: a mailed link has to
47/// point at the environment that minted its token. Not `FRONTEND_BASE_URL`, which no overlay sets, so
48/// reading that sends dev and test links to production.
49static BASE_URL: Lazy<String> = Lazy::new(|| {
50    ProgramConfig::required("BASE_URL").expect("No BASE_URL found in the env variables.")
51});
52static SMTP_HOST: Lazy<String> = Lazy::new(|| {
53    ProgramConfig::required("SMTP_HOST").expect("No email relay found in the env variables.")
54});
55static DB_URL: Lazy<String> = Lazy::new(|| {
56    ProgramConfig::required("DATABASE_URL").expect("No db url found in the env variables.")
57});
58static SMTP_MESSAGE_ID_DOMAIN: Lazy<String> = Lazy::new(|| {
59    ProgramConfig::optional("SMTP_MESSAGE_ID_DOMAIN")
60        .and_then(|value| {
61            let trimmed = value.trim();
62            if trimmed.is_empty() {
63                None
64            } else {
65                Some(trimmed.to_string())
66            }
67        })
68        .or_else(|| infer_email_domain(SMTP_FROM.as_str()))
69        .unwrap_or_else(|| "courses.mooc.fi".to_string())
70});
71static SMTP_USER: Lazy<String> = Lazy::new(|| {
72    ProgramConfig::required("SMTP_USER").expect("No smtp user found in env variables.")
73});
74static SMTP_PASS: Lazy<String> = Lazy::new(|| {
75    ProgramConfig::required("SMTP_PASS").expect("No smtp password found in env variables.")
76});
77/// When set, mail goes to `SMTP_HOST` on this port without TLS, for the local Mailpit in dev and
78/// test. Unset, the sender uses implicit TLS on port 465.
79static SMTP_PLAINTEXT_PORT: Lazy<Option<u16>> = Lazy::new(|| {
80    ProgramConfig::optional("SMTP_PLAINTEXT_PORT").map(|port| {
81        port.trim()
82            .parse()
83            .expect("SMTP_PLAINTEXT_PORT is not a port number.")
84    })
85});
86
87/// Sends one batch of due emails. `warned_layouts` is the `updated_at` of each layout last checked
88/// for problems, so each version is warned about once rather than every batch.
89pub async fn mail_sender(
90    pool: &PgPool,
91    mailer: &SmtpTransport,
92    warned_layouts: &mut HashMap<Uuid, DateTime<Utc>>,
93) -> Result<()> {
94    let mut conn = pool.acquire().await?;
95
96    // Read per batch so a replaced layout takes effect without restarting the sender, and before
97    // claiming: a failure after claiming would hold the rows until their lease runs out.
98    let layouts = email_layouts::get_all_live(&mut conn).await?;
99    warn_about_layout_problems(&layouts, warned_layouts);
100
101    let emails = fetch_emails(&mut conn).await?;
102
103    let mut futures = tokio_stream::iter(emails)
104        .map(|email| {
105            let email_id = email.id;
106            let layout = ResolvedLayout::for_language(&layouts, email.language.as_deref());
107            send_message(email, mailer, pool.clone(), layout).inspect(move |r| {
108                if let Err(err) = r {
109                    tracing::error!("Failed to send email {}: {}", email_id, err)
110                }
111            })
112        })
113        .buffer_unordered(BATCH_SIZE);
114
115    while futures.next().await.is_some() {}
116
117    Ok(())
118}
119
120fn warn_about_layout_problems(
121    layouts: &[EmailLayout],
122    warned_layouts: &mut HashMap<Uuid, DateTime<Utc>>,
123) {
124    for layout in layouts {
125        if warned_layouts.get(&layout.id) == Some(&layout.updated_at) {
126            continue;
127        }
128        for warning in email_processor::layout_warnings(&layout.html) {
129            tracing::warn!(
130                "Email layout {} (language {:?}): {}",
131                layout.id,
132                layout.language,
133                warning
134            );
135        }
136    }
137    *warned_layouts = layouts
138        .iter()
139        .map(|layout| (layout.id, layout.updated_at))
140        .collect();
141}
142
143pub async fn send_message(
144    email: Email,
145    mailer: &SmtpTransport,
146    pool: PgPool,
147    layout: ResolvedLayout<'_>,
148) -> Result<()> {
149    let mut conn = pool.acquire().await?;
150    tracing::info!("Email send messages...");
151
152    let now = Utc::now();
153    let attempt = email.retry_count + 1;
154    if retry_window_expired(email.first_failed_at, now) {
155        tracing::warn!(
156            "Retry window expired for email {} (first_failed_at={:?})",
157            email.id,
158            email.first_failed_at
159        );
160        record_non_retryable_failure(
161            &mut conn,
162            email.id,
163            attempt,
164            "retry_window_expired",
165            format!(
166                "Retry window expired before send attempt (email_id={}, user_id={:?}, template={:?}, first_failed_at={:?})",
167                email.id, email.user_id, email.template_type, email.first_failed_at
168            ),
169        )
170        .await?;
171        return Ok(());
172    }
173
174    let parsed = email
175        .subject
176        .as_deref()
177        .context("No subject")
178        .and_then(|subject| {
179            let body = email.body.as_ref().context("No body")?;
180            let blocks: Vec<EmailGutenbergBlock> =
181                serde_json::from_value(body.clone()).context("Failed to parse email body JSON")?;
182            Ok((subject, blocks))
183        });
184    let (subject, blocks) = match parsed {
185        Ok(parsed) => parsed,
186        Err(err) => {
187            record_message_build_failure(&mut conn, &email, attempt, &err).await?;
188            return Ok(());
189        }
190    };
191
192    let replacements = if email.is_test {
193        placeholder_bag_replacements(email.placeholders.as_ref())
194    } else if let Some(template_type) = email.template_type {
195        let template_result = template_replacements(
196            &mut conn,
197            template_type,
198            email.id,
199            email.user_id,
200            email.placeholders.as_ref(),
201            attempt,
202        )
203        .await?;
204        match template_result {
205            TemplateReplacements::Ready(replacements) => replacements,
206            TemplateReplacements::Abandoned => return Ok(()),
207        }
208    } else {
209        HashMap::new()
210    };
211
212    let rendered = email_processor::render_email(EmailRenderInput {
213        layout_html: layout.html,
214        theme: layout.theme,
215        subject,
216        body: blocks,
217        language: email.language.as_deref().unwrap_or_default(),
218        replacements: &replacements,
219    });
220    if !rendered.unfilled_placeholders.is_empty() {
221        tracing::warn!(
222            "Email {} has placeholders with no value, sent as typed: {:?}",
223            email.id,
224            rendered.unfilled_placeholders
225        );
226    }
227
228    let msg = match build_email_message(&email, attempt, rendered) {
229        Ok(msg) => msg,
230        Err(err) => {
231            record_message_build_failure(&mut conn, &email, attempt, &err).await?;
232            return Ok(());
233        }
234    };
235
236    match mailer.send(&msg) {
237        Ok(_) => {
238            tracing::info!("Email sent successfully {}", email.id);
239            mark_as_sent(&mut conn, email.id)
240                .await
241                .context("Couldn't mark as sent")?;
242        }
243        Err(err) => {
244            let is_transient = is_transient_smtp_error(&err);
245            let (error_code, smtp_response, smtp_response_code) = extract_smtp_error_details(&err);
246
247            tracing::error!(
248                "SMTP send failed for {} (attempt {}, transient={}): {:?}",
249                email.id,
250                attempt,
251                is_transient,
252                err
253            );
254
255            let mut tx = (*conn)
256                .begin()
257                .await
258                .context("Couldn't start email failure transaction")?;
259
260            insert_email_delivery_error(
261                &mut tx,
262                EmailDeliveryErrorInsert {
263                    email_delivery_id: email.id,
264                    attempt,
265                    error_message: err.to_string(),
266                    error_code,
267                    smtp_response,
268                    smtp_response_code,
269                    is_transient,
270                },
271            )
272            .await
273            .context("Couldn't insert email delivery error history")?;
274
275            if is_transient {
276                if retry_window_expired(Some(email.first_failed_at.unwrap_or(now)), now) {
277                    increment_retry_and_mark_non_retryable(&mut tx, email.id)
278                        .await
279                        .context("Couldn't close expired retryable email")?;
280                } else {
281                    // `retry_count` is pre-increment from the claimed row; using it here keeps
282                    // backoff aligned with the next failed-attempt number.
283                    let next_retry_at = compute_next_retry_at(now, email.retry_count);
284                    increment_retry_and_schedule(&mut tx, email.id, Some(next_retry_at))
285                        .await
286                        .context("Couldn't schedule retry")?;
287                }
288            } else {
289                increment_retry_and_mark_non_retryable(&mut tx, email.id)
290                    .await
291                    .context("Couldn't close non-retryable email")?;
292            }
293
294            tx.commit()
295                .await
296                .context("Couldn't commit email failure transaction")?;
297        }
298    };
299
300    Ok(())
301}
302
303enum TemplateReplacements {
304    Ready(HashMap<String, String>),
305    Abandoned,
306}
307
308async fn template_replacements(
309    conn: &mut PgConnection,
310    template_type: EmailTemplateType,
311    email_id: Uuid,
312    user_id: Option<Uuid>,
313    placeholders: Option<&serde_json::Value>,
314    attempt: i32,
315) -> anyhow::Result<TemplateReplacements> {
316    let mut replacements = HashMap::new();
317
318    if template_type == EmailTemplateType::Generic {
319        return Ok(TemplateReplacements::Ready(replacements));
320    }
321
322    if template_type.uses_placeholder_bag() {
323        return Ok(TemplateReplacements::Ready(placeholder_bag_replacements(
324            placeholders,
325        )));
326    }
327
328    // Every remaining template derives its values from an account.
329    let Some(user_id) = user_id else {
330        let msg = format!(
331            "Template {template_type:?} requires a user but the delivery is addressed to a raw address"
332        );
333        record_non_retryable_failure(conn, email_id, attempt, "template", msg).await?;
334        return Ok(TemplateReplacements::Abandoned);
335    };
336
337    match template_type {
338        EmailTemplateType::ResetPasswordEmail => {
339            if let Some(token_str) =
340                get_unused_reset_password_token_with_user_id(conn, user_id).await?
341            {
342                let reset_url = format!(
343                    "{}/reset-user-password/{}",
344                    BASE_URL.trim_end_matches('/'),
345                    token_str.token
346                );
347
348                replacements.insert("RESET_LINK".to_string(), reset_url);
349            } else {
350                let msg = anyhow::anyhow!("No reset token found for user {}", user_id);
351                record_non_retryable_failure(conn, email_id, attempt, "template", msg.to_string())
352                    .await?;
353                return Ok(TemplateReplacements::Abandoned);
354            }
355        }
356        EmailTemplateType::DeleteUserEmail => {
357            if let Some(code) =
358                headless_lms_models::user_email_codes::get_unused_user_email_code_with_user_id(
359                    conn,
360                    user_id,
361                    UserEmailCodePurpose::AccountDeletion,
362                )
363                .await?
364            {
365                replacements.insert("CODE".to_string(), code.code.expose_secret().to_string());
366            } else {
367                let msg = anyhow::anyhow!("No deletion code found for user {}", user_id);
368                record_non_retryable_failure(conn, email_id, attempt, "template", msg.to_string())
369                    .await?;
370                return Ok(TemplateReplacements::Abandoned);
371            }
372        }
373        EmailTemplateType::ConfirmEmailCode => {
374            if let Some(code) =
375                headless_lms_models::user_email_codes::get_unused_user_email_code_with_user_id(
376                    conn,
377                    user_id,
378                    UserEmailCodePurpose::StaffLogin,
379                )
380                .await?
381            {
382                replacements.insert("CODE".to_string(), code.code.expose_secret().to_string());
383            } else {
384                let msg = anyhow::anyhow!("No verification code found for user {}", user_id);
385                record_non_retryable_failure(conn, email_id, attempt, "template", msg.to_string())
386                    .await?;
387                return Ok(TemplateReplacements::Abandoned);
388            }
389        }
390        EmailTemplateType::VerifyEmailAddress => {
391            if let Some(code) =
392                headless_lms_models::user_email_codes::get_unused_user_email_code_with_user_id(
393                    conn,
394                    user_id,
395                    UserEmailCodePurpose::EmailOwnershipVerification,
396                )
397                .await?
398            {
399                replacements.insert("CODE".to_string(), code.code.expose_secret().to_string());
400            } else {
401                let msg = anyhow::anyhow!(
402                    "No email ownership verification code found for user {}",
403                    user_id
404                );
405                record_non_retryable_failure(conn, email_id, attempt, "template", msg.to_string())
406                    .await?;
407                return Ok(TemplateReplacements::Abandoned);
408            }
409        }
410        // Handled above. Listed rather than caught by `_` so a new template type is a compile error.
411        EmailTemplateType::Generic
412        | EmailTemplateType::CreditRegistrationAccountLinking
413        | EmailTemplateType::CreditRegistrationActionNeeded
414        | EmailTemplateType::CreditRegistrationRegistered => {}
415    }
416
417    Ok(TemplateReplacements::Ready(replacements))
418}
419
420/// Turns a delivery's placeholder bag into `{{KEY}}` substitutions. Nested objects and arrays are
421/// skipped: there is no sensible rendering for them in body text.
422fn placeholder_bag_replacements(
423    placeholders: Option<&serde_json::Value>,
424) -> HashMap<String, String> {
425    let Some(serde_json::Value::Object(bag)) = placeholders else {
426        return HashMap::new();
427    };
428    bag.iter()
429        .filter_map(|(key, value)| {
430            let rendered = match value {
431                serde_json::Value::String(s) => s.clone(),
432                serde_json::Value::Number(n) => n.to_string(),
433                serde_json::Value::Bool(b) => b.to_string(),
434                _ => return None,
435            };
436            Some((key.clone(), rendered))
437        })
438        .collect()
439}
440
441fn build_email_message(email: &Email, attempt: i32, rendered: RenderedEmail) -> Result<Message> {
442    Message::builder()
443        .from(SMTP_FROM.parse()?)
444        .to(email
445            .to
446            .parse()
447            .with_context(|| format!("Invalid recipient address for email_id {}", email.id))?)
448        .subject(rendered.subject)
449        .message_id(Some(format!(
450            "<{}-{}@{}>",
451            email.id,
452            attempt,
453            SMTP_MESSAGE_ID_DOMAIN.as_str()
454        )))
455        .multipart(
456            MultiPart::alternative()
457                .singlepart(
458                    SinglePart::builder()
459                        .header(header::ContentType::TEXT_PLAIN)
460                        .body(rendered.plain_text),
461                )
462                .singlepart(
463                    SinglePart::builder()
464                        .header(header::ContentType::TEXT_HTML)
465                        .body(rendered.html),
466                ),
467        )
468        .context("Failed to build email message")
469}
470
471fn infer_email_domain(value: &str) -> Option<String> {
472    let candidate = value
473        .rsplit('<')
474        .next()
475        .unwrap_or(value)
476        .trim()
477        .trim_end_matches('>')
478        .trim();
479    let (_, domain) = candidate.rsplit_once('@')?;
480    let domain = domain.trim();
481    if domain.is_empty() || domain.contains(' ') {
482        None
483    } else {
484        Some(domain.to_string())
485    }
486}
487
488async fn record_message_build_failure(
489    conn: &mut PgConnection,
490    email: &Email,
491    attempt: i32,
492    err: &anyhow::Error,
493) -> Result<()> {
494    tracing::error!(
495        "Message construction failed for email {} (attempt {}): {:#}",
496        email.id,
497        attempt,
498        err
499    );
500    record_non_retryable_failure(
501        conn,
502        email.id,
503        attempt,
504        "message_build",
505        format!("Message construction failed: {err:#}"),
506    )
507    .await
508}
509
510pub async fn main() -> anyhow::Result<()> {
511    dotenvy::dotenv().ok();
512    ProgramConfig::ensure_default_rust_log_for_workers();
513    setup_tracing()?;
514    tracing::info!("Email sender starting up...");
515
516    if ProgramConfig::optional("SMTP_USER").is_none()
517        || ProgramConfig::optional("SMTP_PASS").is_none()
518    {
519        tracing::warn!("SMTP user or password is missing or incorrect");
520    }
521
522    let pool = PgPool::connect(&DB_URL.to_string()).await?;
523    let creds = Credentials::new(SMTP_USER.to_string(), SMTP_PASS.to_string());
524
525    let builder = match *SMTP_PLAINTEXT_PORT {
526        Some(port) => {
527            tracing::warn!("Sending email without TLS to {}:{}", *SMTP_HOST, port);
528            SmtpTransport::builder_dangerous(SMTP_HOST.as_str()).port(port)
529        }
530        None => SmtpTransport::relay(&SMTP_HOST).inspect_err(|e| {
531            tracing::error!("Could not configure SMTP transport: {}", e);
532        })?,
533    };
534    let mailer = builder.credentials(creds).build();
535
536    let mut interval = tokio::time::interval(Duration::from_secs(10));
537    // Startup counts as an attempt: pods restart often, and firing on the first tick would turn every
538    // restart into another sweep.
539    let mut last_purge_attempt = tokio::time::Instant::now();
540    let mut warned_layouts = HashMap::new();
541    loop {
542        interval.tick().await;
543        mail_sender(&pool, &mailer, &mut warned_layouts).await?;
544
545        // An elapsed check rather than a second interval: another `tick().await` in this loop would
546        // stall the 10 second send cycle until the hour was up and stop mail going out.
547        if last_purge_attempt.elapsed() >= RECIPIENT_ADDRESS_PURGE_INTERVAL {
548            last_purge_attempt = tokio::time::Instant::now();
549            let purged = async {
550                let mut conn = pool.acquire().await?;
551                maybe_purge_expired_recipient_addresses(&mut conn).await
552            }
553            .await;
554            // Logged, not propagated: an error out of main() restarts the pod, which would stop mail
555            // going out over a retention sweep.
556            match purged {
557                Ok(purged) if purged > 0 => {
558                    tracing::info!(
559                        "Purged retained recipient addresses from {purged} email deliveries"
560                    )
561                }
562                Ok(_) => {}
563                Err(err) => tracing::error!("Failed to purge retained recipient addresses: {err}"),
564            }
565        }
566    }
567}
568
569fn retry_window_expired(first_failed_at: Option<DateTime<Utc>>, now: DateTime<Utc>) -> bool {
570    backoff::window_expired(first_failed_at, now, RETRY_WINDOW_SECS)
571}
572
573fn compute_next_retry_at(now: DateTime<Utc>, retry_count: i32) -> DateTime<Utc> {
574    let delay = backoff::exponential_backoff_secs(BASE_BACKOFF_SECS, MAX_BACKOFF_SECS, retry_count);
575    backoff::next_attempt_at(now, delay, JITTER_SECS)
576}
577
578fn is_transient_smtp_error(err: &SmtpError) -> bool {
579    if err.is_transient() {
580        return true;
581    }
582    if err.is_timeout() || err.is_transport_shutdown() {
583        return true;
584    }
585    has_io_error(err)
586}
587
588fn has_io_error(err: &SmtpError) -> bool {
589    let mut source = err.source();
590    while let Some(inner) = source {
591        if inner.is::<std::io::Error>() {
592            return true;
593        }
594        source = inner.source();
595    }
596    false
597}
598
599fn extract_smtp_error_details(err: &SmtpError) -> (Option<String>, Option<String>, Option<i32>) {
600    let smtp_response_code = err.status().map(|code| i32::from(u16::from(code)));
601
602    let error_code = if err.is_transient() {
603        Some("transient".to_string())
604    } else if err.is_permanent() {
605        Some("permanent".to_string())
606    } else if err.is_timeout() {
607        Some("timeout".to_string())
608    } else if has_io_error(err) {
609        Some("network_io".to_string())
610    } else if err.is_transport_shutdown() {
611        Some("transport_shutdown".to_string())
612    } else if err.is_response() {
613        Some("response".to_string())
614    } else if err.is_client() {
615        Some("client".to_string())
616    } else {
617        None
618    };
619
620    let smtp_response = err.source().map(|source| source.to_string());
621
622    (error_code, smtp_response, smtp_response_code)
623}
624
625async fn record_non_retryable_failure(
626    conn: &mut PgConnection,
627    email_id: Uuid,
628    attempt: i32,
629    error_code: &'static str,
630    message: String,
631) -> Result<()> {
632    let mut tx = (*conn)
633        .begin()
634        .await
635        .context("Couldn't start template failure transaction")?;
636
637    insert_email_delivery_error(
638        &mut tx,
639        EmailDeliveryErrorInsert {
640            email_delivery_id: email_id,
641            attempt,
642            error_message: message,
643            error_code: Some(error_code.to_string()),
644            smtp_response: None,
645            smtp_response_code: None,
646            is_transient: false,
647        },
648    )
649    .await
650    .context("Couldn't insert email delivery error history")?;
651
652    increment_retry_and_mark_non_retryable(&mut tx, email_id)
653        .await
654        .context("Couldn't mark email as non-retryable for template error")?;
655
656    tx.commit()
657        .await
658        .context("Couldn't commit template failure transaction")?;
659
660    Ok(())
661}
headless_lms_server/programs/email_deliver.rs

headless_lms_server/programs/
email_deliver.rs
