rustango 0.60.7

//! Send email off the request path via the [`crate::jobs`] queue.
//!
//! Render the message in the request, push it on the queue, and let a
//! worker talk to SMTP, SES or Mailgun. SMTP is slow and flaky, so
//! this keeps handler latency steady and gets retry with backoff from
//! the queue.
//!
//! ## Quick start
//!
//! ```ignore
//! use rustango::email_jobs::{EmailJobConfig, register_email_job, dispatch_email};
//! use rustango::jobs::JobQueue;
//! use rustango::email::BoxedMailer;
//! use std::sync::Arc;
//!
//! // Once at startup — register the worker handler with the queue.
//! let mailer: BoxedMailer = Arc::new(SmtpMailer::new(...));
//! register_email_job(&queue, EmailJobConfig::new(mailer.clone())).await;
//! queue.start().await;
//!
//! // From a handler:
//! let email = renderer.render("welcome", &ctx)?
//!     .from("noreply@example.com")
//!     .to("alice@example.com");
//! dispatch_email(&queue, &email).await?;
//! ```
//!
//! ## Behavior
//!
//! - The whole [`Email`] goes into the job payload, so the worker
//!   rebuilds it before sending.
//! - A transport failure becomes [`crate::jobs::JobError::Retryable`], so
//!   the queue backs off and retries; a rejected message is
//!   [`crate::jobs::JobError::Fatal`]. A job that runs out of attempts
//!   goes to the dead-letter callback.
//! - Each queue's handler holds the [`crate::email::Mailer`] it was
//!   registered with, so two queues can send through two mailers.
//!
//! [`Email`]: crate::email::Email

use std::sync::{OnceLock, RwLock};

use serde::{Deserialize, Serialize};

use crate::email::{BoxedMailer, Email};
use crate::jobs::{Job, JobError, JobQueue};

/// Fallback mailer for a custom [`JobQueue`] that does not override
/// `register_with`; the built-in queues keep the mailer in the handler.
fn mailer_registry() -> &'static RwLock<std::collections::HashMap<&'static str, BoxedMailer>> {
    static REG: OnceLock<RwLock<std::collections::HashMap<&'static str, BoxedMailer>>> =
        OnceLock::new();
    REG.get_or_init(|| RwLock::new(std::collections::HashMap::new()))
}

/// Config registered together with the job.
#[derive(Clone)]
pub struct EmailJobConfig {
    /// The mailer the worker sends with.
    pub mailer: BoxedMailer,
}

impl EmailJobConfig {
    #[must_use]
    pub fn new(mailer: BoxedMailer) -> Self {
        Self { mailer }
    }
}

/// The job payload: a copy of the [`Email`].
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct EmailJob {
    pub email: Email,
}

#[async_trait::async_trait]
impl Job for EmailJob {
    const NAME: &'static str = "rustango.send_email";
    /// 8 runs, 5s doubling between them: about ten minutes, enough to
    /// ride out a relay restart.
    const MAX_ATTEMPTS: u32 = 8;

    fn retry_backoff(failed_attempt: u32) -> std::time::Duration {
        crate::jobs::exponential_backoff(std::time::Duration::from_secs(5), failed_attempt)
    }

    async fn run(&self) -> Result<(), JobError> {
        let mailer = mailer_registry()
            .read()
            .unwrap_or_else(|e| e.into_inner())
            .get(Self::NAME)
            .cloned()
            .ok_or_else(|| {
                JobError::Queue(
                    "EmailJob: no mailer registered (call register_email_job at startup)".into(),
                )
            })?;
        send(&mailer, &self.email).await
    }
}

async fn send(mailer: &BoxedMailer, email: &Email) -> Result<(), JobError> {
    mailer.send(email).await.map_err(|e| {
        if e.is_retryable() {
            JobError::Retryable(format!("mailer: {e}"))
        } else {
            JobError::Fatal(format!("mailer: {e}"))
        }
    })
}

/// Register the email job and its mailer on `queue`, once at startup.
/// Calling it again on the same queue replaces the mailer.
///
/// Each in-memory queue keeps its own mailer (#2334). Database queues
/// share one `rustango_jobs` table, so any of them may send any queued
/// mail: use one mailer per jobs table (#2338).
pub async fn register_email_job<Q: JobQueue>(queue: &Q, cfg: EmailJobConfig) {
    #[cfg(feature = "jobs-postgres")]
    warn_on_second_db_mailer(queue, &cfg.mailer);
    // Only for a custom queue that does not override `register_with`.
    mailer_registry()
        .write()
        .unwrap_or_else(|e| e.into_inner())
        .insert(EmailJob::NAME, cfg.mailer.clone());
    let mailer = cfg.mailer;
    queue
        .register_with(move |job: EmailJob| {
            let mailer = mailer.clone();
            async move { send(&mailer, &job.email).await }
        })
        .await;
}

/// Warn once per database when two of its queues get different mailers.
#[cfg(feature = "jobs-postgres")]
fn warn_on_second_db_mailer<Q: JobQueue>(queue: &Q, mailer: &BoxedMailer) {
    let any: &dyn std::any::Any = queue;
    let Some(q) = any.downcast_ref::<crate::jobs::pg::PgJobQueue>() else {
        return;
    };
    let (db, queue_id) = q.identity();
    if mixed_db_mailers(db, queue_id, mailer) {
        tracing::warn!(
            target: "rustango::email_jobs",
            "register_email_job: two queues on one jobs table have different mailers; \
             either may send any queued mail (#2338)"
        );
    }
}

/// Record `mailer` for `queue_id` on `db`. `true` the first time the
/// queues of `db` disagree on their mailer.
#[cfg(feature = "jobs-postgres")]
fn mixed_db_mailers(db: u64, queue_id: usize, mailer: &BoxedMailer) -> bool {
    use std::collections::HashMap;
    type Seen = HashMap<u64, (HashMap<usize, BoxedMailer>, bool)>;
    static SEEN: OnceLock<std::sync::Mutex<Seen>> = OnceLock::new();
    let mut seen = SEEN
        .get_or_init(Default::default)
        .lock()
        .unwrap_or_else(|e| e.into_inner());
    let (queues, warned) = seen.entry(db).or_default();
    queues.insert(queue_id, mailer.clone());
    let mixed = queues.values().any(|m| !std::sync::Arc::ptr_eq(m, mailer));
    if mixed && !*warned {
        *warned = true;
        return true;
    }
    false
}

/// Queue an email and return at once; a worker delivers it.
///
/// # Errors
/// [`JobError::Fatal`] when [`Email::validate`] rejects the message;
/// [`JobError::Queue`] when the enqueue fails, for example the
/// database is down or the payload will not serialize.
pub async fn dispatch_email<Q: JobQueue>(queue: &Q, email: &Email) -> Result<(), JobError> {
    email
        .validate()
        .map_err(|e| JobError::Fatal(format!("email: {e}")))?;
    queue
        .dispatch(&EmailJob {
            email: email.clone(),
        })
        .await
}

/// Clear the mailer registry between tests.
#[cfg(test)]
pub fn reset_mailer_registry() {
    if let Ok(mut g) = mailer_registry().write() {
        g.clear();
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::email::{InMemoryMailer, NullMailer};
    use crate::jobs::InMemoryJobQueue;
    use std::sync::Arc as StdArc;
    use std::time::Duration;
    use tokio::sync::Mutex;

    /// These tests share the global mailer registry, so they must run
    /// one at a time.
    fn lock() -> &'static Mutex<()> {
        static M: std::sync::OnceLock<Mutex<()>> = std::sync::OnceLock::new();
        M.get_or_init(|| Mutex::new(()))
    }

    fn email() -> Email {
        Email::new()
            .from("noreply@x.com")
            .to("alice@x.com")
            .subject("Hi")
            .body("hello")
    }

    #[tokio::test]
    async fn dispatch_then_worker_sends() {
        let _g = lock().lock().await;
        reset_mailer_registry();
        let mailer = StdArc::new(InMemoryMailer::new());
        let q = InMemoryJobQueue::with_workers(1);
        register_email_job(&q, EmailJobConfig::new(mailer.clone())).await;
        q.start().await;

        dispatch_email(&q, &email()).await.unwrap();

        for _ in 0..50 {
            if mailer.count() > 0 {
                break;
            }
            tokio::time::sleep(Duration::from_millis(50)).await;
        }
        assert_eq!(mailer.count(), 1);
        let sent = mailer.sent();
        assert_eq!(sent[0].subject, "Hi");
        assert_eq!(sent[0].to, vec!["alice@x.com"]);

        q.shutdown().await;
    }

    #[tokio::test]
    async fn no_mailer_registered_returns_queue_error() {
        let _g = lock().lock().await;
        reset_mailer_registry();
        // Register the job so dispatch works, but no mailer.
        let q = InMemoryJobQueue::with_workers(1);
        q.register::<EmailJob>().await;

        // Count dead letters to see the worker reject the job.
        let dl_count = StdArc::new(std::sync::atomic::AtomicUsize::new(0));
        let dl = dl_count.clone();
        q.on_dead_letter(move |_dl| {
            let dl = dl.clone();
            async move {
                dl.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            }
        })
        .await;
        q.start().await;

        dispatch_email(&q, &email()).await.unwrap();

        // JobError::Queue is not retried, so it dead-letters at once.
        for _ in 0..40 {
            if dl_count.load(std::sync::atomic::Ordering::SeqCst) > 0 {
                break;
            }
            tokio::time::sleep(Duration::from_millis(50)).await;
        }
        assert_eq!(dl_count.load(std::sync::atomic::Ordering::SeqCst), 1);
        q.shutdown().await;
    }

    #[tokio::test]
    async fn re_register_swaps_the_mailer() {
        let _g = lock().lock().await;
        reset_mailer_registry();
        let m1 = StdArc::new(InMemoryMailer::new());
        let m2 = StdArc::new(InMemoryMailer::new());
        let q = InMemoryJobQueue::with_workers(1);

        register_email_job(&q, EmailJobConfig::new(m1.clone())).await;
        register_email_job(&q, EmailJobConfig::new(m2.clone())).await;
        q.start().await;

        dispatch_email(&q, &email()).await.unwrap();
        for _ in 0..50 {
            if m2.count() > 0 {
                break;
            }
            tokio::time::sleep(Duration::from_millis(50)).await;
        }
        assert_eq!(m1.count(), 0, "old mailer should not receive");
        assert_eq!(m2.count(), 1, "new mailer should receive");

        q.shutdown().await;
    }

    /// Two queues keep their own mailers; the second registration must
    /// not reroute the first queue's mail (#2334).
    #[tokio::test]
    async fn two_queues_send_through_their_own_mailers() {
        let _g = lock().lock().await;
        reset_mailer_registry();
        let (m1, m2) = (
            StdArc::new(InMemoryMailer::new()),
            StdArc::new(InMemoryMailer::new()),
        );
        let (q1, q2) = (
            InMemoryJobQueue::with_workers(1),
            InMemoryJobQueue::with_workers(1),
        );
        register_email_job(&q1, EmailJobConfig::new(m1.clone())).await;
        register_email_job(&q2, EmailJobConfig::new(m2.clone())).await;
        q1.start().await;
        q2.start().await;

        dispatch_email(&q1, &email()).await.unwrap();
        for _ in 0..50 {
            if m1.count() + m2.count() > 0 {
                break;
            }
            tokio::time::sleep(Duration::from_millis(50)).await;
        }
        assert_eq!((m1.count(), m2.count()), (1, 0), "sent by the wrong mailer");
        q1.shutdown().await;
        q2.shutdown().await;
    }

    /// The warning fires once per database, only when two of its queues
    /// have different mailers.
    #[cfg(feature = "jobs-postgres")]
    #[test]
    fn mixed_db_mailers_warns_once_per_database() {
        let (m1, m2): (BoxedMailer, BoxedMailer) =
            (StdArc::new(NullMailer), StdArc::new(NullMailer));
        // Distinct `db` keys keep this test apart from any other.
        let db = 0xD8_2337;
        assert!(!mixed_db_mailers(db, 1, &m1), "first queue");
        assert!(!mixed_db_mailers(db, 1, &m2), "same queue re-registered");
        assert!(!mixed_db_mailers(db + 1, 2, &m1), "own database");
        assert!(!mixed_db_mailers(db, 2, &m2), "same mailer on the database");
        assert!(mixed_db_mailers(db, 3, &m1), "two mailers on one table");
        assert!(!mixed_db_mailers(db, 4, &m1), "warned once");
    }

    /// Retries span minutes, so a short relay outage does not dead-letter
    /// the mail (#2332).
    #[test]
    fn email_retries_span_several_minutes() {
        let total: Duration = (0..EmailJob::MAX_ATTEMPTS - 1)
            .map(EmailJob::retry_backoff)
            .sum();
        assert!(
            total >= Duration::from_secs(300),
            "gives up after {total:?}"
        );
    }

    /// Rejects every message and counts the calls.
    struct Rejecting(std::sync::atomic::AtomicUsize);

    #[async_trait::async_trait]
    impl crate::email::Mailer for Rejecting {
        async fn send(&self, _: &Email) -> Result<(), crate::email::MailError> {
            self.0.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            Err(crate::email::MailError::BadHeader("x".into()))
        }
    }

    /// A message the mailer rejects will never send; retrying it only
    /// delays the dead letter (#1923).
    #[tokio::test]
    async fn a_rejected_message_dead_letters_without_retry() {
        let _g = lock().lock().await;
        reset_mailer_registry();
        let mailer = StdArc::new(Rejecting(std::sync::atomic::AtomicUsize::new(0)));
        let q = InMemoryJobQueue::with_workers(1);
        register_email_job(&q, EmailJobConfig::new(mailer.clone())).await;
        let dead = StdArc::new(std::sync::atomic::AtomicUsize::new(0));
        let d = dead.clone();
        q.on_dead_letter(move |_| {
            let d = d.clone();
            async move {
                d.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            }
        })
        .await;
        q.start().await;

        dispatch_email(&q, &email()).await.unwrap();
        for _ in 0..40 {
            if dead.load(std::sync::atomic::Ordering::SeqCst) > 0 {
                break;
            }
            tokio::time::sleep(Duration::from_millis(50)).await;
        }
        assert_eq!(
            dead.load(std::sync::atomic::Ordering::SeqCst),
            1,
            "not dead-lettered"
        );
        assert_eq!(
            mailer.0.load(std::sync::atomic::Ordering::SeqCst),
            1,
            "retried"
        );
        q.shutdown().await;
    }

    /// An invalid message is refused at dispatch, not queued (#1923).
    #[tokio::test]
    async fn dispatch_refuses_an_invalid_message() {
        let _g = lock().lock().await;
        reset_mailer_registry();
        let q = InMemoryJobQueue::with_workers(1);
        q.register::<EmailJob>().await;
        let bad = email().subject("");
        assert!(matches!(
            dispatch_email(&q, &bad).await,
            Err(JobError::Fatal(_))
        ));
        assert_eq!(q.pending_count().await, 0);
    }

    #[tokio::test]
    async fn null_mailer_succeeds_silently() {
        let _g = lock().lock().await;
        reset_mailer_registry();
        let mailer: BoxedMailer = StdArc::new(NullMailer);
        let q = InMemoryJobQueue::with_workers(1);
        register_email_job(&q, EmailJobConfig::new(mailer)).await;
        q.start().await;

        // Nothing to observe: it must not panic or dead-letter.
        dispatch_email(&q, &email()).await.unwrap();
        tokio::time::sleep(Duration::from_millis(100)).await;
        // Back to 0 once the job is done.
        assert_eq!(q.pending_count().await, 0);
        q.shutdown().await;
    }
}