Skip to main content

headless_lms_models/
email_deliveries.rs

1use std::collections::HashMap;
2
3use rand::RngExt;
4use utoipa::ToSchema;
5
6use crate::email_templates::EmailTemplateType;
7use crate::prelude::*;
8
9pub const FETCH_LIMIT: i64 = 20;
10
11/// How long a delivery keeps being retried after its first failure.
12///
13/// Shared by the sender and [`derive_email_send_status`], which must agree on what has failed.
14pub const RETRY_WINDOW_SECS: i64 = 3 * 24 * 60 * 60;
15
16/// How long a delivery to a raw address may keep that address after being queued.
17const RECIPIENT_ADDRESS_RETENTION: &str = "1 month";
18
19/// One purge in this many calls, matching the token cleanups. The caller schedules how often it asks.
20const PURGE_CHANCE_IN: u32 = 10;
21
22#[derive(Debug, Serialize, Deserialize, PartialEq, Clone)]
23pub struct EmailDelivery {
24    pub id: Uuid,
25    pub created_at: DateTime<Utc>,
26    pub updated_at: DateTime<Utc>,
27    pub deleted_at: Option<DateTime<Utc>>,
28    pub email_template_id: Uuid,
29    pub sent: bool,
30    /// `None` for deliveries addressed to a raw address; see `recipient_email`.
31    pub user_id: Option<Uuid>,
32    pub recipient_email: Option<String>,
33    pub placeholders: Option<serde_json::Value>,
34    /// Number of failed send attempts recorded so far.
35    pub retry_count: i32,
36    pub next_retry_at: Option<DateTime<Utc>>,
37    pub retryable: bool,
38    pub first_failed_at: Option<DateTime<Utc>>,
39    pub last_attempt_at: Option<DateTime<Utc>>,
40}
41
42pub struct Email {
43    pub id: Uuid,
44    /// `None` when the mail goes to a raw address with no account here; substitutions must cope.
45    pub user_id: Option<Uuid>,
46    pub to: String,
47    pub subject: Option<String>,
48    pub body: Option<serde_json::Value>,
49    pub template_type: Option<EmailTemplateType>,
50    /// The template's language, else its course's; `None` for a global template without one.
51    pub language: Option<String>,
52    /// Substitutions carried on the delivery row, so a mail to a raw address needs no user lookup.
53    pub placeholders: Option<serde_json::Value>,
54    /// A test send from the email editor: `subject` and `body` are its snapshot rather than the
55    /// saved template, and every substitution comes from `placeholders`.
56    pub is_test: bool,
57    /// Number of failed send attempts recorded so far.
58    pub retry_count: i32,
59    pub next_retry_at: Option<DateTime<Utc>>,
60    pub retryable: bool,
61    pub first_failed_at: Option<DateTime<Utc>>,
62    pub last_attempt_at: Option<DateTime<Utc>>,
63}
64
65/// Inserts an email delivery; fails if the user or email template is soft-deleted.
66///
67/// For a template whose substitutions the sender cannot look up from the account, use
68/// [`insert_email_delivery_with_placeholders`].
69pub async fn insert_email_delivery(
70    conn: &mut PgConnection,
71    user_id: Uuid,
72    email_template_id: Uuid,
73) -> ModelResult<Uuid> {
74    insert_delivery_for_user(conn, user_id, email_template_id, None).await
75}
76
77/// Queues a mail to an account, carrying its substitutions on the delivery row.
78///
79/// The sibling of [`insert_email_delivery_to_address`] for recipients we do have an account for:
80/// the address comes from the account, the values in the message do not.
81pub async fn insert_email_delivery_with_placeholders(
82    conn: &mut PgConnection,
83    user_id: Uuid,
84    email_template_id: Uuid,
85    placeholders: &serde_json::Value,
86) -> ModelResult<Uuid> {
87    insert_delivery_for_user(conn, user_id, email_template_id, Some(placeholders)).await
88}
89
90async fn insert_delivery_for_user(
91    conn: &mut PgConnection,
92    user_id: Uuid,
93    email_template_id: Uuid,
94    placeholders: Option<&serde_json::Value>,
95) -> ModelResult<Uuid> {
96    let check = sqlx::query_as!(
97        CheckUserAndTemplateRow,
98        r#"
99SELECT
100    EXISTS(SELECT 1 FROM users WHERE id = $1 AND deleted_at IS NULL) AS "user_ok!",
101    EXISTS(SELECT 1 FROM email_templates WHERE id = $2 AND deleted_at IS NULL) AS "template_ok!"
102        "#,
103        user_id,
104        email_template_id
105    )
106    .fetch_one(&mut *conn)
107    .await?;
108    if !check.user_ok {
109        return Err(ModelError::new(
110            ModelErrorType::PreconditionFailed,
111            "User not found or deleted".to_string(),
112            None,
113        ));
114    }
115    if !check.template_ok {
116        return Err(ModelError::new(
117            ModelErrorType::PreconditionFailed,
118            "Email template not found or deleted".to_string(),
119            None,
120        ));
121    }
122
123    let id = Uuid::new_v4();
124    sqlx::query!(
125        r#"
126INSERT INTO email_deliveries (
127    id,
128    user_id,
129    email_template_id,
130    placeholders
131)
132VALUES ($1, $2, $3, $4)
133        "#,
134        id,
135        user_id,
136        email_template_id,
137        placeholders,
138    )
139    .execute(conn)
140    .await?;
141
142    Ok(id)
143}
144
145struct CheckUserAndTemplateRow {
146    user_ok: bool,
147    template_ok: bool,
148}
149
150/// Queues an email to a raw address, for recipients who may have no account here.
151pub async fn insert_email_delivery_to_address(
152    conn: &mut PgConnection,
153    recipient_email: &str,
154    email_template_id: Uuid,
155    placeholders: &serde_json::Value,
156) -> ModelResult<Uuid> {
157    let template_ok = sqlx::query_scalar!(
158        r#"
159SELECT EXISTS(SELECT 1 FROM email_templates WHERE id = $1 AND deleted_at IS NULL) AS "template_ok!"
160        "#,
161        email_template_id
162    )
163    .fetch_one(&mut *conn)
164    .await?;
165    if !template_ok {
166        return Err(ModelError::new(
167            ModelErrorType::PreconditionFailed,
168            "Email template not found or deleted".to_string(),
169            None,
170        ));
171    }
172
173    let id = Uuid::new_v4();
174    sqlx::query!(
175        r#"
176INSERT INTO email_deliveries (
177    id,
178    recipient_email,
179    email_template_id,
180    placeholders
181)
182VALUES ($1, $2, $3, $4)
183        "#,
184        id,
185        recipient_email,
186        email_template_id,
187        placeholders
188    )
189    .execute(conn)
190    .await?;
191
192    Ok(id)
193}
194
195/// Queues a test send of an unsaved template body to the account's own address, unless the
196/// account already queued `max_per_minute` in the last minute, in which case returns `None`.
197/// `placeholders` must hold every substitution: the sender derives none for a test send.
198pub async fn insert_test_email_delivery_within_limit(
199    conn: &mut PgConnection,
200    user_id: Uuid,
201    email_template_id: Uuid,
202    subject: &str,
203    content: &serde_json::Value,
204    placeholders: &serde_json::Value,
205    max_per_minute: i64,
206) -> ModelResult<Option<Uuid>> {
207    let mut tx = conn.begin().await?;
208    // Serializes concurrent sends of one user, so the count below cannot go stale before the insert.
209    sqlx::query!(
210        "SELECT pg_advisory_xact_lock(hashtextextended($1, 0))",
211        user_id.to_string()
212    )
213    .execute(&mut *tx)
214    .await?;
215    let recent = sqlx::query_scalar!(
216        r#"
217SELECT COUNT(*) AS "count!"
218FROM email_deliveries
219WHERE user_id = $1
220  AND test_content IS NOT NULL
221  AND created_at > now() - interval '1 minute'
222        "#,
223        user_id
224    )
225    .fetch_one(&mut *tx)
226    .await?;
227    if recent >= max_per_minute {
228        tx.rollback().await?;
229        return Ok(None);
230    }
231    let id = Uuid::new_v4();
232    sqlx::query!(
233        r#"
234INSERT INTO email_deliveries (
235    id,
236    user_id,
237    email_template_id,
238    placeholders,
239    test_subject,
240    test_content
241)
242VALUES ($1, $2, $3, $4, $5, $6)
243        "#,
244        id,
245        user_id,
246        email_template_id,
247        placeholders,
248        subject,
249        content,
250    )
251    .execute(&mut *tx)
252    .await?;
253    tx.commit().await?;
254    Ok(Some(id))
255}
256
257pub async fn fetch_emails(conn: &mut PgConnection) -> ModelResult<Vec<Email>> {
258    let emails = sqlx::query_as!(
259        Email,
260        r#"
261WITH due AS (
262    SELECT
263        ed.id
264    FROM email_deliveries ed
265    LEFT JOIN users u ON u.id = ed.user_id
266    LEFT JOIN user_details ud ON ud.user_id = ed.user_id
267    JOIN email_templates et ON et.id = ed.email_template_id
268    WHERE ed.deleted_at IS NULL
269      AND ed.sent = FALSE
270      AND ed.retryable = TRUE
271      AND (ed.user_id IS NULL OR u.deleted_at IS NULL)
272      AND (ed.recipient_email IS NOT NULL OR ud.email IS NOT NULL)
273      AND et.deleted_at IS NULL
274      AND (ed.next_retry_at IS NULL OR ed.next_retry_at <= now())
275    ORDER BY coalesce(ed.next_retry_at, '-infinity'::timestamptz), ed.created_at
276    -- OF ed is required, not cosmetic: rows on the nullable side of an outer join cannot be locked.
277    FOR UPDATE OF ed SKIP LOCKED
278    LIMIT $1
279),
280claimed AS (
281    UPDATE email_deliveries ed
282    SET last_attempt_at = now(),
283        -- Crash-recovery lease for claimed rows; this is not retry backoff.
284        next_retry_at = now() + interval '5 minutes'
285    FROM due
286    WHERE ed.id = due.id
287    RETURNING
288        ed.id,
289        ed.user_id,
290        ed.recipient_email,
291        ed.placeholders,
292        ed.test_subject,
293        ed.test_content,
294        ed.email_template_id,
295        ed.retry_count,
296        ed.next_retry_at,
297        ed.retryable,
298        ed.first_failed_at,
299        ed.last_attempt_at
300)
301SELECT
302    c.id AS id,
303    c.user_id AS user_id,
304    COALESCE(c.recipient_email, ud.email) AS "to!",
305    COALESCE(c.test_subject, et.subject) AS subject,
306    COALESCE(c.test_content, et.content) AS body,
307    et.email_template_type AS "template_type",
308    email_template_language(et) AS language,
309    c.placeholders AS placeholders,
310    c.test_content IS NOT NULL AS "is_test!",
311    c.retry_count AS retry_count,
312    c.next_retry_at AS next_retry_at,
313    c.retryable AS retryable,
314    c.first_failed_at AS first_failed_at,
315    c.last_attempt_at AS last_attempt_at
316FROM claimed c
317JOIN email_templates et ON et.id = c.email_template_id
318LEFT JOIN user_details ud ON ud.user_id = c.user_id
319ORDER BY c.last_attempt_at ASC;
320        "#,
321        FETCH_LIMIT
322    )
323    .fetch_all(conn)
324    .await?;
325
326    Ok(emails)
327}
328
329pub async fn mark_as_sent(conn: &mut PgConnection, email_id: Uuid) -> ModelResult<()> {
330    sqlx::query!(
331        "
332update email_deliveries
333set sent = TRUE,
334    next_retry_at = NULL
335where id = $1;
336    ",
337        email_id
338    )
339    .execute(conn)
340    .await?;
341
342    Ok(())
343}
344
345pub async fn insert_email_delivery_error(
346    conn: &mut PgConnection,
347    error: EmailDeliveryErrorInsert,
348) -> ModelResult<Uuid> {
349    let id = Uuid::new_v4();
350    sqlx::query!(
351        r#"
352INSERT INTO email_delivery_errors (
353    id,
354    email_delivery_id,
355    attempt,
356    error_message,
357    error_code,
358    smtp_response,
359    smtp_response_code,
360    is_transient
361)
362VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
363        "#,
364        id,
365        error.email_delivery_id,
366        error.attempt,
367        error.error_message,
368        error.error_code,
369        error.smtp_response,
370        error.smtp_response_code,
371        error.is_transient
372    )
373    .execute(conn)
374    .await?;
375
376    Ok(id)
377}
378
379pub struct EmailDeliveryErrorInsert {
380    pub email_delivery_id: Uuid,
381    pub attempt: i32,
382    pub error_message: String,
383    pub error_code: Option<String>,
384    pub smtp_response: Option<String>,
385    pub smtp_response_code: Option<i32>,
386    pub is_transient: bool,
387}
388
389#[derive(Debug, Serialize, Deserialize, PartialEq, Clone)]
390pub struct EmailDeliveryError {
391    pub id: Uuid,
392    pub email_delivery_id: Uuid,
393    pub attempt: i32,
394    pub error_message: String,
395    pub error_code: Option<String>,
396    pub smtp_response: Option<String>,
397    pub smtp_response_code: Option<i32>,
398    pub is_transient: bool,
399    pub created_at: DateTime<Utc>,
400    pub updated_at: DateTime<Utc>,
401    pub deleted_at: Option<DateTime<Utc>>,
402}
403
404/// Which mails one account has been queued, newest first. There is no mail capture in this repo, so
405/// a system test asserting a message was composed reads the queue instead of an inbox.
406pub async fn get_recent_template_types_for_user_for_testing(
407    conn: &mut PgConnection,
408    user_id: Uuid,
409    limit: i64,
410) -> ModelResult<Vec<(EmailTemplateType, serde_json::Value)>> {
411    let rows = sqlx::query!(
412        r#"
413SELECT t.email_template_type AS "template_type",
414  COALESCE(d.placeholders, '{}'::jsonb) AS "placeholders!"
415FROM email_deliveries d
416  JOIN email_templates t ON t.id = d.email_template_id
417WHERE d.user_id = $1
418  AND d.deleted_at IS NULL
419ORDER BY d.created_at DESC
420LIMIT $2
421        "#,
422        user_id,
423        limit,
424    )
425    .fetch_all(conn)
426    .await?;
427    Ok(rows
428        .into_iter()
429        .map(|row| (row.template_type, row.placeholders))
430        .collect())
431}
432
433pub async fn increment_retry_and_schedule(
434    conn: &mut PgConnection,
435    email_id: Uuid,
436    next_retry_at: Option<DateTime<Utc>>,
437) -> ModelResult<()> {
438    sqlx::query!(
439        "
440UPDATE email_deliveries
441SET retry_count = retry_count + 1,
442    next_retry_at = $2,
443    first_failed_at = COALESCE(first_failed_at, NOW())
444where id = $1;
445    ",
446        email_id,
447        next_retry_at
448    )
449    .execute(conn)
450    .await?;
451
452    Ok(())
453}
454
455pub async fn increment_retry_and_mark_non_retryable(
456    conn: &mut PgConnection,
457    email_id: Uuid,
458) -> ModelResult<()> {
459    sqlx::query!(
460        "
461UPDATE email_deliveries
462SET retry_count = retry_count + 1,
463    first_failed_at = COALESCE(first_failed_at, NOW()),
464    retryable = FALSE,
465    next_retry_at = NULL
466WHERE id = $1;
467    ",
468        email_id
469    )
470    .execute(conn)
471    .await?;
472
473    Ok(())
474}
475
476/// What we can honestly say about an email we queued.
477///
478/// We only hand messages to an SMTP relay, so copy rendering this must never say "delivered" or
479/// "received".
480#[derive(Debug, Serialize, Deserialize, PartialEq, Eq, Clone, Copy, ToSchema)]
481#[serde(rename_all = "snake_case")]
482pub enum EmailSendStatus {
483    /// In our queue, not handed over yet.
484    Queued,
485    /// At least one attempt failed transiently; `next_retry_at` says when we try again.
486    Retrying,
487    /// Handed to the mail relay. Not a delivery confirmation.
488    Sent,
489    /// We could not hand it over at all, and will not try again.
490    SendFailed,
491}
492
493/// The shared payload for every surface that reports on a queued email.
494#[derive(Debug, Serialize, Deserialize, PartialEq, Clone, ToSchema)]
495pub struct EmailSendStatusReport {
496    pub email_send_status: EmailSendStatus,
497    pub sent_at: Option<DateTime<Utc>>,
498    pub last_attempt_at: Option<DateTime<Utc>>,
499    pub retry_count: i32,
500    pub next_retry_at: Option<DateTime<Utc>>,
501    pub failure_code: Option<String>,
502    pub failure_is_transient: Option<bool>,
503}
504
505/// The delivery-row facts [`derive_email_send_status`] needs, so the derivation itself is pure.
506#[derive(Debug, PartialEq, Clone)]
507pub struct EmailSendStatusFacts {
508    pub sent: bool,
509    pub retryable: bool,
510    pub retry_count: i32,
511    pub next_retry_at: Option<DateTime<Utc>>,
512    pub first_failed_at: Option<DateTime<Utc>>,
513    pub last_attempt_at: Option<DateTime<Utc>>,
514    /// From the newest `email_delivery_errors` row, if any.
515    pub failure_code: Option<String>,
516    pub failure_is_transient: Option<bool>,
517}
518
519/// True once a delivery attempt has failed for good: not retryable at all, or retryable but its
520/// retry window has expired. Shared with `credit_registration_account_linking_emails.rs` so the
521/// two cannot drift on what counts as a hard failure.
522pub fn is_hard_send_failure(
523    retryable: bool,
524    first_failed_at: Option<DateTime<Utc>>,
525    now: DateTime<Utc>,
526) -> bool {
527    !retryable
528        || first_failed_at.is_some_and(|first| (now - first).num_seconds() > RETRY_WINDOW_SECS)
529}
530
531/// The one derivation of send status. Every surface goes through it.
532pub fn derive_email_send_status(
533    facts: &EmailSendStatusFacts,
534    now: DateTime<Utc>,
535) -> EmailSendStatusReport {
536    let email_send_status = if facts.sent {
537        EmailSendStatus::Sent
538    } else if is_hard_send_failure(facts.retryable, facts.first_failed_at, now) {
539        EmailSendStatus::SendFailed
540    } else if facts.retry_count > 0 {
541        EmailSendStatus::Retrying
542    } else {
543        EmailSendStatus::Queued
544    };
545
546    EmailSendStatusReport {
547        email_send_status,
548        // There is no sent_at column; on a sent row the last attempt is the one that succeeded.
549        sent_at: if facts.sent {
550            facts.last_attempt_at
551        } else {
552            None
553        },
554        last_attempt_at: facts.last_attempt_at,
555        retry_count: facts.retry_count,
556        next_retry_at: match email_send_status {
557            EmailSendStatus::Retrying => facts.next_retry_at,
558            _ => None,
559        },
560        failure_code: facts.failure_code.clone(),
561        failure_is_transient: facts.failure_is_transient,
562    }
563}
564
565pub async fn get_send_statuses(
566    conn: &mut PgConnection,
567    email_delivery_ids: &[Uuid],
568) -> ModelResult<HashMap<Uuid, EmailSendStatusReport>> {
569    let rows = sqlx::query!(
570        r#"
571SELECT ed.id,
572  ed.sent,
573  ed.retryable,
574  ed.retry_count,
575  ed.next_retry_at,
576  ed.first_failed_at,
577  ed.last_attempt_at,
578  latest_error.error_code,
579  latest_error.is_transient
580FROM email_deliveries ed
581  LEFT JOIN LATERAL (
582    SELECT ede.error_code,
583      ede.is_transient
584    FROM email_delivery_errors ede
585    WHERE ede.email_delivery_id = ed.id
586      AND ede.deleted_at IS NULL
587    ORDER BY ede.attempt DESC,
588      ede.created_at DESC
589    LIMIT 1
590  ) latest_error ON TRUE
591WHERE ed.id = ANY($1::uuid [])
592  AND ed.deleted_at IS NULL
593        "#,
594        email_delivery_ids
595    )
596    .fetch_all(conn)
597    .await?;
598
599    let now = Utc::now();
600    let res = rows
601        .into_iter()
602        .map(|row| {
603            let facts = EmailSendStatusFacts {
604                sent: row.sent,
605                retryable: row.retryable,
606                retry_count: row.retry_count,
607                next_retry_at: row.next_retry_at,
608                first_failed_at: row.first_failed_at,
609                last_attempt_at: row.last_attempt_at,
610                failure_code: row.error_code,
611                failure_is_transient: row.is_transient,
612            };
613            (row.id, derive_email_send_status(&facts, now))
614        })
615        .collect();
616    Ok(res)
617}
618
619pub async fn get_send_status(
620    conn: &mut PgConnection,
621    email_delivery_id: Uuid,
622) -> ModelResult<Option<EmailSendStatusReport>> {
623    let mut statuses = get_send_statuses(conn, &[email_delivery_id]).await?;
624    Ok(statuses.remove(&email_delivery_id))
625}
626
627/// One delivery attempt, as summarized for a support-facing view of a user's mail history.
628///
629/// Deliberately excludes the message content and the recipient address (which may be purged by
630/// [`maybe_purge_expired_recipient_addresses`] anyway): this exists so support tooling can see
631/// whether mail reached a user, not what was in it.
632#[derive(Debug, Serialize, Deserialize, PartialEq, Clone, ToSchema)]
633pub struct UserEmailDeliverySummary {
634    pub email_template_type: EmailTemplateType,
635    pub created_at: DateTime<Utc>,
636    pub status: EmailSendStatus,
637    pub retry_count: i32,
638    pub last_attempt_at: Option<DateTime<Utc>>,
639    pub failure_code: Option<String>,
640    pub failure_is_transient: Option<bool>,
641}
642
643/// A user's most recent email deliveries, newest first, for support tooling.
644pub async fn get_recent_deliveries_for_user(
645    conn: &mut PgConnection,
646    user_id: Uuid,
647    limit: i64,
648) -> ModelResult<Vec<UserEmailDeliverySummary>> {
649    let rows = sqlx::query!(
650        r#"
651SELECT et.email_template_type,
652  ed.created_at,
653  ed.sent,
654  ed.retryable,
655  ed.retry_count,
656  ed.next_retry_at,
657  ed.first_failed_at,
658  ed.last_attempt_at,
659  latest_error.error_code,
660  latest_error.is_transient AS "is_transient?"
661FROM email_deliveries ed
662  JOIN email_templates et ON et.id = ed.email_template_id
663  LEFT JOIN LATERAL (
664    SELECT ede.error_code,
665      ede.is_transient
666    FROM email_delivery_errors ede
667    WHERE ede.email_delivery_id = ed.id
668      AND ede.deleted_at IS NULL
669    ORDER BY ede.attempt DESC,
670      ede.created_at DESC
671    LIMIT 1
672  ) latest_error ON TRUE
673WHERE ed.user_id = $1
674  AND ed.deleted_at IS NULL
675ORDER BY ed.created_at DESC
676LIMIT $2
677        "#,
678        user_id,
679        limit
680    )
681    .fetch_all(conn)
682    .await?;
683
684    let now = Utc::now();
685    let summaries = rows
686        .into_iter()
687        .map(|row| {
688            let facts = EmailSendStatusFacts {
689                sent: row.sent,
690                retryable: row.retryable,
691                retry_count: row.retry_count,
692                next_retry_at: row.next_retry_at,
693                first_failed_at: row.first_failed_at,
694                last_attempt_at: row.last_attempt_at,
695                failure_code: row.error_code,
696                failure_is_transient: row.is_transient,
697            };
698            let report = derive_email_send_status(&facts, now);
699            UserEmailDeliverySummary {
700                email_template_type: row.email_template_type,
701                created_at: row.created_at,
702                status: report.email_send_status,
703                retry_count: report.retry_count,
704                last_attempt_at: report.last_attempt_at,
705                failure_code: report.failure_code,
706                failure_is_transient: report.failure_is_transient,
707            }
708        })
709        .collect();
710    Ok(summaries)
711}
712
713/// Soft-deletes unsent, still-retryable email deliveries for a user. Call when soft-deleting the user so pending deliveries are not retried.
714pub async fn soft_delete_unsent_retryable_deliveries_for_user(
715    conn: &mut PgConnection,
716    user_id: Uuid,
717) -> ModelResult<()> {
718    sqlx::query!(
719        "
720UPDATE email_deliveries
721SET deleted_at = NOW()
722WHERE user_id = $1
723  AND deleted_at IS NULL
724  AND sent = FALSE
725  AND retryable = TRUE",
726        user_id
727    )
728    .execute(conn)
729    .await?;
730    Ok(())
731}
732
733/// Probabilistic purge instead of a cron, in the style of the token cleanups.
734///
735/// Only raw-address rows ever hold an address; a delivery addressed by `user_id` resolves the address
736/// from `user_details` at send time and stores nothing.
737pub async fn maybe_purge_expired_recipient_addresses(conn: &mut PgConnection) -> ModelResult<u64> {
738    if rand::rng().random_range(1..=PURGE_CHANCE_IN) != 1 {
739        return Ok(0);
740    }
741    info!("Purging retained recipient addresses past their retention window");
742    let result = sqlx::query!(
743        r#"
744UPDATE email_deliveries ed
745SET recipient_email = NULL,
746    placeholders = CASE
747      WHEN ed.placeholders IS NULL THEN NULL
748      ELSE ed.placeholders - 'EMAIL'
749    END,
750    -- Without an address the row can never be delivered, so retire it here instead of leaving the
751    -- sender to claim it and fail. The CHECK constraint also requires this.
752    retryable = CASE WHEN ed.sent THEN ed.retryable ELSE FALSE END,
753    next_retry_at = CASE WHEN ed.sent THEN ed.next_retry_at ELSE NULL END,
754    deleted_at = CASE
755      WHEN ed.sent OR ed.deleted_at IS NOT NULL THEN ed.deleted_at
756      ELSE now()
757    END
758WHERE ed.recipient_email IS NOT NULL
759  AND ed.created_at < now() - $1::text::interval
760  -- The sender stamps last_attempt_at when it claims a row and holds a five minute lease, so an hour
761  -- of quiet means nothing is mid-send.
762  AND (
763    ed.sent
764    OR ed.last_attempt_at IS NULL
765    OR ed.last_attempt_at < now() - interval '1 hour'
766  )
767        "#,
768        RECIPIENT_ADDRESS_RETENTION
769    )
770    .execute(conn)
771    .await?;
772    Ok(result.rows_affected())
773}
774
775#[cfg(test)]
776mod tests {
777    use chrono::Duration;
778
779    use super::*;
780
781    fn queued_facts() -> EmailSendStatusFacts {
782        EmailSendStatusFacts {
783            sent: false,
784            retryable: true,
785            retry_count: 0,
786            next_retry_at: None,
787            first_failed_at: None,
788            last_attempt_at: None,
789            failure_code: None,
790            failure_is_transient: None,
791        }
792    }
793
794    #[test]
795    fn queued_when_nothing_has_been_attempted() {
796        let now = Utc::now();
797        let report = derive_email_send_status(&queued_facts(), now);
798        assert_eq!(report.email_send_status, EmailSendStatus::Queued);
799        assert_eq!(report.sent_at, None);
800        assert_eq!(report.next_retry_at, None);
801    }
802
803    #[test]
804    fn retrying_after_a_transient_failure_reports_when_we_try_again() {
805        let now = Utc::now();
806        let next_retry_at = now + Duration::minutes(5);
807        let facts = EmailSendStatusFacts {
808            retry_count: 1,
809            next_retry_at: Some(next_retry_at),
810            first_failed_at: Some(now - Duration::minutes(1)),
811            last_attempt_at: Some(now - Duration::minutes(1)),
812            failure_code: Some("transient".to_string()),
813            failure_is_transient: Some(true),
814            ..queued_facts()
815        };
816        let report = derive_email_send_status(&facts, now);
817        assert_eq!(report.email_send_status, EmailSendStatus::Retrying);
818        assert_eq!(report.next_retry_at, Some(next_retry_at));
819        assert_eq!(report.failure_code.as_deref(), Some("transient"));
820    }
821
822    #[test]
823    fn sent_reports_the_successful_attempt_as_sent_at() {
824        let now = Utc::now();
825        let handed_over_at = now - Duration::minutes(2);
826        let facts = EmailSendStatusFacts {
827            sent: true,
828            last_attempt_at: Some(handed_over_at),
829            ..queued_facts()
830        };
831        let report = derive_email_send_status(&facts, now);
832        assert_eq!(report.email_send_status, EmailSendStatus::Sent);
833        assert_eq!(report.sent_at, Some(handed_over_at));
834    }
835
836    #[test]
837    fn send_failed_when_the_delivery_is_no_longer_retryable() {
838        let now = Utc::now();
839        let facts = EmailSendStatusFacts {
840            retryable: false,
841            retry_count: 1,
842            first_failed_at: Some(now - Duration::minutes(1)),
843            last_attempt_at: Some(now - Duration::minutes(1)),
844            failure_code: Some("permanent".to_string()),
845            failure_is_transient: Some(false),
846            ..queued_facts()
847        };
848        let report = derive_email_send_status(&facts, now);
849        assert_eq!(report.email_send_status, EmailSendStatus::SendFailed);
850        assert_eq!(report.next_retry_at, None);
851        assert_eq!(report.failure_is_transient, Some(false));
852    }
853
854    #[test]
855    fn send_failed_when_the_retry_window_has_expired_even_though_the_row_still_says_retryable() {
856        let now = Utc::now();
857        let facts = EmailSendStatusFacts {
858            retry_count: 9,
859            next_retry_at: Some(now + Duration::hours(1)),
860            first_failed_at: Some(now - Duration::seconds(RETRY_WINDOW_SECS + 1)),
861            last_attempt_at: Some(now - Duration::hours(1)),
862            failure_code: Some("transient".to_string()),
863            failure_is_transient: Some(true),
864            ..queued_facts()
865        };
866        let report = derive_email_send_status(&facts, now);
867        assert_eq!(report.email_send_status, EmailSendStatus::SendFailed);
868        assert_eq!(report.next_retry_at, None);
869    }
870
871    #[test]
872    fn a_sent_delivery_stays_sent_even_if_earlier_attempts_failed() {
873        let now = Utc::now();
874        let facts = EmailSendStatusFacts {
875            sent: true,
876            retryable: false,
877            retry_count: 2,
878            first_failed_at: Some(now - Duration::seconds(RETRY_WINDOW_SECS + 1)),
879            last_attempt_at: Some(now),
880            failure_code: Some("transient".to_string()),
881            failure_is_transient: Some(true),
882            ..queued_facts()
883        };
884        assert_eq!(
885            derive_email_send_status(&facts, now).email_send_status,
886            EmailSendStatus::Sent
887        );
888    }
889}