aster_forge_mail/
outbox.rs

1//! Product-neutral mail outbox dispatch helpers.
2
3use std::future::Future;
4use std::time::Duration;
5
6#[cfg(feature = "persistence")]
7use sea_orm::entity::prelude::*;
8use serde::{Deserialize, Serialize};
9#[cfg(all(debug_assertions, feature = "openapi"))]
10use utoipa::ToSchema;
11
12/// Default maximum stored delivery error length.
13pub const DEFAULT_ERROR_MAX_LEN: usize = 1024;
14
15/// Default retry delays for persisting "sent" after successful delivery.
16///
17/// The first attempt is immediate. Later attempts provide a short best-effort
18/// window for transient database failures after SMTP has already accepted the
19/// message.
20pub const DEFAULT_MARK_SENT_RETRY_DELAYS_MS: &[u64] = &[0, 100, 500, 2_000, 5_000];
21
22/// Built-in Aster mail template code.
23///
24/// These are the shared account/auth mail templates used by current Aster
25/// services. Products may still maintain their own template payload enum and
26/// renderer; this type only standardizes the persisted template code for the
27/// common catalog.
28#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
29#[cfg_attr(feature = "persistence", derive(EnumIter, DeriveActiveEnum))]
30#[cfg_attr(all(debug_assertions, feature = "openapi"), derive(ToSchema))]
31#[cfg_attr(
32    feature = "persistence",
33    sea_orm(rs_type = "String", db_type = "String(StringLen::N(64))")
34)]
35#[serde(rename_all = "snake_case")]
36pub enum MailTemplateCode {
37    /// Account registration activation message.
38    #[cfg_attr(feature = "persistence", sea_orm(string_value = "register_activation"))]
39    RegisterActivation,
40    /// Confirmation message for changing a contact address.
41    #[cfg_attr(
42        feature = "persistence",
43        sea_orm(string_value = "contact_change_confirmation")
44    )]
45    ContactChangeConfirmation,
46    /// Password reset message.
47    #[cfg_attr(feature = "persistence", sea_orm(string_value = "password_reset"))]
48    PasswordReset,
49    /// Password reset notice message.
50    #[cfg_attr(
51        feature = "persistence",
52        sea_orm(string_value = "password_reset_notice")
53    )]
54    PasswordResetNotice,
55    /// Contact change notice message.
56    #[cfg_attr(
57        feature = "persistence",
58        sea_orm(string_value = "contact_change_notice")
59    )]
60    ContactChangeNotice,
61    /// External auth email verification message.
62    #[cfg_attr(
63        feature = "persistence",
64        sea_orm(string_value = "external_auth_email_verification")
65    )]
66    ExternalAuthEmailVerification,
67    /// Login email code message.
68    #[cfg_attr(feature = "persistence", sea_orm(string_value = "login_email_code"))]
69    LoginEmailCode,
70    /// User invitation message.
71    #[cfg_attr(feature = "persistence", sea_orm(string_value = "user_invitation"))]
72    UserInvitation,
73}
74
75impl MailTemplateCode {
76    /// Returns the stable persisted template code.
77    #[must_use]
78    pub const fn as_str(self) -> &'static str {
79        match self {
80            Self::RegisterActivation => "register_activation",
81            Self::ContactChangeConfirmation => "contact_change_confirmation",
82            Self::PasswordReset => "password_reset",
83            Self::PasswordResetNotice => "password_reset_notice",
84            Self::ContactChangeNotice => "contact_change_notice",
85            Self::ExternalAuthEmailVerification => "external_auth_email_verification",
86            Self::LoginEmailCode => "login_email_code",
87            Self::UserInvitation => "user_invitation",
88        }
89    }
90}
91
92/// Raw JSON payload stored in `mail_outbox.payload_json`.
93#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
94#[cfg_attr(feature = "persistence", derive(DeriveValueType))]
95pub struct StoredMailPayload(pub String);
96
97impl StoredMailPayload {
98    /// Payload value persisted after terminal delivery, so sensitive template
99    /// variables do not remain in the outbox row.
100    pub const CLEARED_JSON: &str = "{}";
101
102    /// Creates the cleared payload marker.
103    #[must_use]
104    pub fn cleared() -> Self {
105        Self(Self::CLEARED_JSON.to_string())
106    }
107}
108
109impl AsRef<str> for StoredMailPayload {
110    fn as_ref(&self) -> &str {
111        &self.0
112    }
113}
114
115impl From<String> for StoredMailPayload {
116    fn from(value: String) -> Self {
117        Self(value)
118    }
119}
120
121impl From<StoredMailPayload> for String {
122    fn from(value: StoredMailPayload) -> Self {
123        value.0
124    }
125}
126
127/// Persistent mail outbox row status.
128#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
129#[cfg_attr(feature = "persistence", derive(EnumIter, DeriveActiveEnum))]
130#[cfg_attr(all(debug_assertions, feature = "openapi"), derive(ToSchema))]
131#[cfg_attr(
132    feature = "persistence",
133    sea_orm(rs_type = "String", db_type = "String(StringLen::N(16))")
134)]
135#[serde(rename_all = "snake_case")]
136pub enum MailOutboxStatus {
137    /// Row is waiting for first delivery attempt.
138    #[cfg_attr(feature = "persistence", sea_orm(string_value = "pending"))]
139    Pending,
140    /// Row is claimed by a dispatcher.
141    #[cfg_attr(feature = "persistence", sea_orm(string_value = "processing"))]
142    Processing,
143    /// Row is waiting for another delivery attempt.
144    #[cfg_attr(feature = "persistence", sea_orm(string_value = "retry"))]
145    Retry,
146    /// Row was delivered and marked sent.
147    #[cfg_attr(feature = "persistence", sea_orm(string_value = "sent"))]
148    Sent,
149    /// Row exhausted retry policy.
150    #[cfg_attr(feature = "persistence", sea_orm(string_value = "failed"))]
151    Failed,
152}
153
154impl MailOutboxStatus {
155    /// Returns whether the status is terminal.
156    #[must_use]
157    pub const fn is_terminal(self) -> bool {
158        matches!(self, Self::Sent | Self::Failed)
159    }
160}
161
162/// Aggregate counters returned by an outbox dispatch or drain pass.
163#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
164pub struct DispatchStats {
165    /// Rows claimed for processing.
166    pub claimed: usize,
167    /// Rows delivered and marked sent.
168    pub sent: usize,
169    /// Rows scheduled for retry.
170    pub retried: usize,
171    /// Rows permanently failed.
172    pub failed: usize,
173}
174
175impl DispatchStats {
176    /// Adds another counter set into this one.
177    pub fn merge(&mut self, other: Self) {
178        self.claimed += other.claimed;
179        self.sent += other.sent;
180        self.retried += other.retried;
181        self.failed += other.failed;
182    }
183
184    /// Returns whether the dispatch pass did any visible work.
185    #[must_use]
186    pub const fn is_empty(self) -> bool {
187        self.claimed == 0 && self.sent == 0 && self.retried == 0 && self.failed == 0
188    }
189}
190
191/// Lightweight row metadata captured before a dispatcher consumes the concrete row.
192///
193/// Dispatch callbacks often need stable audit/log fields after the delivery callback has consumed
194/// the database row. Keeping those fields in this small context avoids cloning the full row, which
195/// may contain a large template payload or delivery error.
196#[derive(Debug, Clone, PartialEq, Eq)]
197pub struct MailOutboxDispatchContext {
198    /// Stable row id.
199    pub id: i64,
200    /// Current delivery attempt count stored on the row before this dispatch attempt.
201    pub attempt_count: i32,
202    /// Stable product template code used for logs and audits.
203    pub template_code: String,
204    /// Recipient address used for logs and audits.
205    pub to_address: String,
206    /// Optional recipient display name used for audit records.
207    pub to_name: Option<String>,
208}
209
210impl MailOutboxDispatchContext {
211    /// Returns the attempt count after the current delivery attempt.
212    #[must_use]
213    pub const fn delivery_attempt_count(&self) -> i32 {
214        self.attempt_count + 1
215    }
216}
217
218/// Minimal row metadata needed by the shared outbox dispatcher.
219///
220/// Product crates keep their concrete database model and implement this trait on it so Forge can
221/// log and apply retry policy without knowing `SeaORM` entities or product-specific columns.
222pub trait MailOutboxDispatchRow {
223    /// Stable row id.
224    fn id(&self) -> i64;
225
226    /// Current delivery attempt count stored on the row before this dispatch attempt.
227    fn attempt_count(&self) -> i32;
228
229    /// Stable product template code used for logs and audits.
230    fn template_code(&self) -> &str;
231
232    /// Recipient address used for logs and audits.
233    fn to_address(&self) -> &str;
234
235    /// Optional recipient display name used for audit records.
236    fn to_name(&self) -> Option<&str> {
237        None
238    }
239
240    /// Captures the small metadata set that remains useful after the concrete row is consumed.
241    fn dispatch_context(&self) -> MailOutboxDispatchContext {
242        MailOutboxDispatchContext {
243            id: self.id(),
244            attempt_count: self.attempt_count(),
245            template_code: self.template_code().to_string(),
246            to_address: self.to_address().to_string(),
247            to_name: self.to_name().map(str::to_string),
248        }
249    }
250}
251
252/// Configuration for the shared mail outbox dispatcher.
253#[derive(Debug, Clone, PartialEq, Eq)]
254pub struct MailOutboxDispatchConfig {
255    /// Maximum rows loaded in one dispatch pass.
256    pub batch_size: u64,
257    /// Stale processing window in seconds. Product callbacks decide how this maps to timestamps.
258    pub processing_stale_secs: i64,
259    /// Maximum dispatch rounds during a drain pass.
260    pub drain_max_rounds: usize,
261    /// Delivery retry policy.
262    pub retry_policy: MailOutboxRetryPolicy,
263    /// Retry delays for marking a row as sent after delivery success.
264    pub mark_sent_retry_delays_ms: &'static [u64],
265}
266
267impl MailOutboxDispatchConfig {
268    /// Creates a dispatch config.
269    #[must_use]
270    pub const fn new(
271        batch_size: u64,
272        processing_stale_secs: i64,
273        drain_max_rounds: usize,
274        retry_policy: MailOutboxRetryPolicy,
275    ) -> Self {
276        Self {
277            batch_size,
278            processing_stale_secs,
279            drain_max_rounds,
280            retry_policy,
281            mark_sent_retry_delays_ms: DEFAULT_MARK_SENT_RETRY_DELAYS_MS,
282        }
283    }
284}
285
286/// Retry and truncation policy for an outbox dispatcher.
287#[derive(Debug, Clone, PartialEq, Eq)]
288pub struct MailOutboxRetryPolicy {
289    /// Maximum number of delivery attempts before permanent failure.
290    pub max_attempts: i32,
291    /// Maximum stored error string length, counted in Unicode scalar values.
292    pub max_error_len: usize,
293}
294
295/// Decision returned after a mail delivery attempt fails.
296#[derive(Debug, Clone, PartialEq, Eq)]
297pub enum MailOutboxDeliveryFailureDecision {
298    /// The row exhausted delivery attempts and should be moved to a terminal failure state.
299    PermanentFailure {
300        /// Attempt count that should be persisted on the outbox row.
301        attempt_count: i32,
302        /// Truncated delivery error safe for storage.
303        error_message: String,
304    },
305    /// The row should be scheduled for another delivery attempt.
306    Retry {
307        /// Attempt count that should be persisted on the outbox row.
308        attempt_count: i32,
309        /// Delay before the next delivery attempt, in seconds.
310        retry_delay_secs: i64,
311        /// Truncated delivery error safe for storage.
312        error_message: String,
313    },
314}
315
316impl MailOutboxRetryPolicy {
317    /// Creates a retry policy.
318    #[must_use]
319    pub const fn new(max_attempts: i32, max_error_len: usize) -> Self {
320        Self {
321            max_attempts,
322            max_error_len,
323        }
324    }
325
326    /// Returns whether `attempt_count` should permanently fail.
327    #[must_use]
328    pub const fn should_permanently_fail(&self, attempt_count: i32) -> bool {
329        attempt_count >= self.max_attempts
330    }
331
332    /// Returns the delay before the next delivery retry.
333    #[must_use]
334    pub const fn retry_delay_secs(&self, attempt_count: i32) -> i64 {
335        retry_delay_secs(attempt_count)
336    }
337
338    /// Truncates a delivery error according to this policy.
339    #[must_use]
340    pub fn truncate_error(&self, error: &str) -> String {
341        truncate_error(error, self.max_error_len)
342    }
343
344    /// Classifies a failed delivery attempt.
345    ///
346    /// `attempt_count` should be the post-delivery attempt count that the
347    /// product crate will persist on the outbox row.
348    pub fn delivery_failure_decision(
349        &self,
350        attempt_count: i32,
351        error: impl AsRef<str>,
352    ) -> MailOutboxDeliveryFailureDecision {
353        let error_message = self.truncate_error(error.as_ref());
354        if self.should_permanently_fail(attempt_count) {
355            MailOutboxDeliveryFailureDecision::PermanentFailure {
356                attempt_count,
357                error_message,
358            }
359        } else {
360            MailOutboxDeliveryFailureDecision::Retry {
361                attempt_count,
362                retry_delay_secs: self.retry_delay_secs(attempt_count),
363                error_message,
364            }
365        }
366    }
367}
368
369/// Returns the default mail delivery retry delay for an attempt count.
370#[must_use]
371pub const fn retry_delay_secs(attempt_count: i32) -> i64 {
372    match attempt_count {
373        1 => 5,
374        2 => 15,
375        3 => 60,
376        4 => 300,
377        5 => 900,
378        _ => 1800,
379    }
380}
381
382/// Truncates an error string without splitting UTF-8 code points.
383#[must_use]
384pub fn truncate_error(error: &str, max_len: usize) -> String {
385    error.chars().take(max_len).collect()
386}
387
388/// Runs one mail outbox dispatch pass using product-provided persistence callbacks.
389///
390/// Forge owns the control flow and retry classification. Product crates own time calculation,
391/// repository calls, template rendering, mail auditing, and error types.
392#[expect(
393    clippy::too_many_arguments,
394    clippy::too_many_lines,
395    reason = "The dispatcher keeps claim, delivery, outcome persistence, retry, and audit callbacks in one fenced dispatch transaction."
396)]
397///
398/// # Errors
399///
400/// Returns the product callback error when listing, claiming, delivery, or outcome persistence fails.
401pub async fn dispatch_mail_outbox<
402    R,
403    E,
404    List,
405    ListFut,
406    Claim,
407    ClaimFut,
408    Deliver,
409    DeliverFut,
410    MarkSent,
411    MarkSentFut,
412    MarkRetry,
413    MarkRetryFut,
414    MarkFailed,
415    MarkFailedFut,
416    OnSent,
417    OnSentFut,
418    OnFailed,
419    OnFailedFut,
420>(
421    config: &MailOutboxDispatchConfig,
422    mut list_claimable: List,
423    mut try_claim: Claim,
424    mut deliver: Deliver,
425    mut mark_sent: MarkSent,
426    mut mark_retry: MarkRetry,
427    mut mark_failed: MarkFailed,
428    mut on_sent: OnSent,
429    mut on_failed: OnFailed,
430) -> Result<DispatchStats, E>
431where
432    R: MailOutboxDispatchRow,
433    E: std::fmt::Display,
434    List: FnMut(u64, i64) -> ListFut,
435    ListFut: Future<Output = Result<Vec<R>, E>>,
436    Claim: FnMut(i64) -> ClaimFut,
437    ClaimFut: Future<Output = Result<bool, E>>,
438    Deliver: FnMut(R) -> DeliverFut,
439    DeliverFut: Future<Output = Result<String, E>>,
440    MarkSent: FnMut(i64, usize) -> MarkSentFut,
441    MarkSentFut: Future<Output = Result<bool, E>>,
442    MarkRetry: FnMut(i64, i32, i64, String) -> MarkRetryFut,
443    MarkRetryFut: Future<Output = Result<bool, E>>,
444    MarkFailed: FnMut(i64, i32, String) -> MarkFailedFut,
445    MarkFailedFut: Future<Output = Result<bool, E>>,
446    OnSent: FnMut(MailOutboxDispatchContext, i32, String) -> OnSentFut,
447    OnSentFut: Future<Output = ()>,
448    OnFailed: FnMut(MailOutboxDispatchContext, i32, String) -> OnFailedFut,
449    OnFailedFut: Future<Output = ()>,
450{
451    let rows = list_claimable(config.batch_size, config.processing_stale_secs).await?;
452    let mut stats = DispatchStats::default();
453    tracing::debug!(
454        batch_size = config.batch_size,
455        due_count = rows.len(),
456        stale_secs = config.processing_stale_secs,
457        "dispatching due mail outbox rows"
458    );
459
460    for row in rows {
461        let context = row.dispatch_context();
462        if !try_claim(context.id).await? {
463            tracing::debug!(
464                mail_outbox_id = context.id,
465                template_code = %context.template_code,
466                "mail outbox claim skipped because row was already claimed"
467            );
468            continue;
469        }
470
471        stats.claimed += 1;
472        tracing::debug!(
473            mail_outbox_id = context.id,
474            template_code = %context.template_code,
475            attempt_count = context.attempt_count,
476            "claimed mail outbox row"
477        );
478
479        match deliver(row).await {
480            Ok(subject) => {
481                tracing::debug!(
482                    mail_outbox_id = context.id,
483                    template_code = %context.template_code,
484                    "mail outbox delivery succeeded"
485                );
486                match retry_mark_sent(context.id, config.mark_sent_retry_delays_ms, &mut mark_sent)
487                    .await
488                {
489                    Ok(true) => {
490                        stats.sent += 1;
491                        let attempt_count = context.delivery_attempt_count();
492                        on_sent(context, attempt_count, subject).await;
493                    }
494                    Ok(false) => {
495                        tracing::warn!(
496                            mail_outbox_id = context.id,
497                            template_code = %context.template_code,
498                            to = %context.to_address,
499                            "mark_sent affected 0 rows after successful delivery; state will be rechecked"
500                        );
501                    }
502                    Err(error) => {
503                        tracing::error!(
504                            mail_outbox_id = context.id,
505                            template_code = %context.template_code,
506                            to = %context.to_address,
507                            stale_secs = config.processing_stale_secs,
508                            error = %error,
509                            "CRITICAL: mail delivery succeeded but mark_sent failed after all retries; \
510                             row remains Processing and may be re-claimed, causing duplicate delivery"
511                        );
512                    }
513                }
514            }
515            Err(error) => {
516                let attempt_count = context.delivery_attempt_count();
517                match config
518                    .retry_policy
519                    .delivery_failure_decision(attempt_count, error.to_string())
520                {
521                    MailOutboxDeliveryFailureDecision::PermanentFailure {
522                        attempt_count,
523                        error_message,
524                    } => {
525                        if mark_failed(context.id, attempt_count, error_message.clone()).await? {
526                            stats.failed += 1;
527                            on_failed(context.clone(), attempt_count, error_message.clone()).await;
528                        }
529                        tracing::warn!(
530                            mail_outbox_id = context.id,
531                            template_code = %context.template_code,
532                            to = %context.to_address,
533                            attempt_count,
534                            error = %error_message,
535                            "mail outbox delivery permanently failed"
536                        );
537                    }
538                    MailOutboxDeliveryFailureDecision::Retry {
539                        attempt_count,
540                        retry_delay_secs,
541                        error_message,
542                    } => {
543                        if mark_retry(
544                            context.id,
545                            attempt_count,
546                            retry_delay_secs,
547                            error_message.clone(),
548                        )
549                        .await?
550                        {
551                            stats.retried += 1;
552                        }
553                        tracing::warn!(
554                            mail_outbox_id = context.id,
555                            template_code = %context.template_code,
556                            to = %context.to_address,
557                            attempt_count,
558                            retry_delay_secs,
559                            error = %error_message,
560                            "mail outbox delivery failed; scheduled retry"
561                        );
562                    }
563                }
564            }
565        }
566    }
567
568    tracing::debug!(
569        claimed = stats.claimed,
570        sent = stats.sent,
571        retried = stats.retried,
572        failed = stats.failed,
573        "finished dispatching due mail outbox rows"
574    );
575    Ok(stats)
576}
577
578/// Runs mail outbox dispatch passes until no rows are claimed or the configured drain limit is hit.
579///
580/// # Errors
581///
582/// Returns the dispatcher error from a drain round that does not complete successfully.
583pub async fn drain_mail_outbox<E, Dispatch, DispatchFut>(
584    config: &MailOutboxDispatchConfig,
585    mut dispatch: Dispatch,
586) -> Result<DispatchStats, E>
587where
588    E: std::fmt::Display,
589    Dispatch: FnMut() -> DispatchFut,
590    DispatchFut: Future<Output = Result<DispatchStats, E>>,
591{
592    let mut total = DispatchStats::default();
593    tracing::debug!("draining mail outbox");
594
595    for _ in 0..config.drain_max_rounds {
596        let stats = dispatch().await?;
597        let claimed = stats.claimed;
598        total.merge(stats);
599        if claimed == 0 {
600            tracing::debug!("mail outbox drain finished because no rows were claimed");
601            break;
602        }
603    }
604
605    tracing::debug!(
606        claimed = total.claimed,
607        sent = total.sent,
608        retried = total.retried,
609        failed = total.failed,
610        "mail outbox drain completed"
611    );
612    Ok(total)
613}
614
615/// Retries a product-provided `mark_sent` operation after SMTP success.
616///
617/// This helper exists to narrow the duplicate-delivery window where SMTP has
618/// accepted a message but the database row still says `Processing`. The caller
619/// provides the actual persistence function so repositories, transactions,
620/// timestamps, and product errors stay in the product crate.
621///
622/// # Errors
623///
624/// Returns the final persistence error after every configured mark-sent retry fails.
625pub async fn retry_mark_sent<F, Fut, E>(
626    id: i64,
627    retry_delays_ms: &[u64],
628    mut mark_sent: F,
629) -> Result<bool, E>
630where
631    F: FnMut(i64, usize) -> Fut,
632    Fut: Future<Output = Result<bool, E>>,
633    E: std::fmt::Display,
634{
635    let mut last_err = None;
636    for (index, delay_ms) in retry_delays_ms.iter().enumerate() {
637        if *delay_ms > 0 {
638            tokio::time::sleep(Duration::from_millis(*delay_ms)).await;
639        }
640
641        let attempt = index + 1;
642        match mark_sent(id, attempt).await {
643            Ok(updated) => {
644                tracing::debug!(
645                    mail_outbox_id = id,
646                    attempt,
647                    updated,
648                    "marked mail outbox row as sent"
649                );
650                return Ok(updated);
651            }
652            Err(error) => {
653                tracing::warn!(
654                    mail_outbox_id = id,
655                    attempt,
656                    "mark_sent failed, will retry: {error}"
657                );
658                last_err = Some(error);
659            }
660        }
661    }
662
663    match last_err {
664        Some(error) => Err(error),
665        None => mark_sent(id, retry_delays_ms.len() + 1).await,
666    }
667}
668
669#[cfg(test)]
670mod tests {
671    use super::{
672        DEFAULT_ERROR_MAX_LEN, DispatchStats, MailOutboxDispatchConfig, MailOutboxDispatchContext,
673        MailOutboxDispatchRow, MailOutboxRetryPolicy, MailOutboxStatus, MailTemplateCode,
674        StoredMailPayload, dispatch_mail_outbox, retry_delay_secs, retry_mark_sent, truncate_error,
675    };
676    use std::sync::{
677        Arc,
678        atomic::{AtomicUsize, Ordering},
679    };
680
681    #[test]
682    fn dispatch_stats_merge_adds_all_counters() {
683        let mut stats = DispatchStats {
684            claimed: 1,
685            sent: 2,
686            retried: 3,
687            failed: 4,
688        };
689        stats.merge(DispatchStats {
690            claimed: 10,
691            sent: 20,
692            retried: 30,
693            failed: 40,
694        });
695
696        assert_eq!(
697            stats,
698            DispatchStats {
699                claimed: 11,
700                sent: 22,
701                retried: 33,
702                failed: 44,
703            }
704        );
705        assert!(!stats.is_empty());
706        assert!(DispatchStats::default().is_empty());
707    }
708
709    #[test]
710    fn retry_policy_matches_default_mail_backoff() {
711        let policy = MailOutboxRetryPolicy::new(6, DEFAULT_ERROR_MAX_LEN);
712
713        assert!(!policy.should_permanently_fail(5));
714        assert!(policy.should_permanently_fail(6));
715        assert_eq!(policy.retry_delay_secs(1), 5);
716        assert_eq!(policy.retry_delay_secs(2), 15);
717        assert_eq!(policy.retry_delay_secs(3), 60);
718        assert_eq!(policy.retry_delay_secs(4), 300);
719        assert_eq!(policy.retry_delay_secs(5), 900);
720        assert_eq!(retry_delay_secs(99), 1800);
721    }
722
723    #[test]
724    fn retry_policy_classifies_delivery_failures() {
725        let policy = MailOutboxRetryPolicy::new(2, 3);
726
727        assert_eq!(
728            policy.delivery_failure_decision(1, "abcdef"),
729            super::MailOutboxDeliveryFailureDecision::Retry {
730                attempt_count: 1,
731                retry_delay_secs: 5,
732                error_message: "abc".to_string(),
733            }
734        );
735        assert_eq!(
736            policy.delivery_failure_decision(2, "abcdef"),
737            super::MailOutboxDeliveryFailureDecision::PermanentFailure {
738                attempt_count: 2,
739                error_message: "abc".to_string(),
740            }
741        );
742    }
743
744    #[test]
745    fn truncate_error_preserves_utf8_boundaries() {
746        let value = "界".repeat(4);
747        assert_eq!(truncate_error(&value, 3), "界界界");
748    }
749
750    #[test]
751    fn mail_template_code_exposes_stable_storage_names() {
752        assert_eq!(
753            MailTemplateCode::RegisterActivation.as_str(),
754            "register_activation"
755        );
756        assert_eq!(
757            MailTemplateCode::ContactChangeConfirmation.as_str(),
758            "contact_change_confirmation"
759        );
760        assert_eq!(MailTemplateCode::PasswordReset.as_str(), "password_reset");
761        assert_eq!(
762            MailTemplateCode::PasswordResetNotice.as_str(),
763            "password_reset_notice"
764        );
765        assert_eq!(
766            MailTemplateCode::ContactChangeNotice.as_str(),
767            "contact_change_notice"
768        );
769        assert_eq!(
770            MailTemplateCode::ExternalAuthEmailVerification.as_str(),
771            "external_auth_email_verification"
772        );
773        assert_eq!(
774            MailTemplateCode::LoginEmailCode.as_str(),
775            "login_email_code"
776        );
777        assert_eq!(MailTemplateCode::UserInvitation.as_str(), "user_invitation");
778    }
779
780    #[test]
781    fn mail_template_code_storage_names_fit_shared_schema() {
782        let codes = [
783            MailTemplateCode::RegisterActivation,
784            MailTemplateCode::ContactChangeConfirmation,
785            MailTemplateCode::PasswordReset,
786            MailTemplateCode::PasswordResetNotice,
787            MailTemplateCode::ContactChangeNotice,
788            MailTemplateCode::ExternalAuthEmailVerification,
789            MailTemplateCode::LoginEmailCode,
790            MailTemplateCode::UserInvitation,
791        ];
792
793        for code in codes {
794            assert!(
795                code.as_str().len() <= 64,
796                "mail template code `{}` exceeds shared schema length",
797                code.as_str()
798            );
799        }
800    }
801
802    #[test]
803    fn stored_mail_payload_helpers_preserve_raw_json() {
804        let payload = StoredMailPayload::from("{\"token\":\"abc\"}".to_string());
805        assert_eq!(payload.as_ref(), "{\"token\":\"abc\"}");
806
807        let raw: String = payload.into();
808        assert_eq!(raw, "{\"token\":\"abc\"}");
809        assert_eq!(StoredMailPayload::cleared().as_ref(), "{}");
810    }
811
812    #[test]
813    fn mail_outbox_status_terminal_states_are_explicit() {
814        assert!(!MailOutboxStatus::Pending.is_terminal());
815        assert!(!MailOutboxStatus::Processing.is_terminal());
816        assert!(!MailOutboxStatus::Retry.is_terminal());
817        assert!(MailOutboxStatus::Sent.is_terminal());
818        assert!(MailOutboxStatus::Failed.is_terminal());
819    }
820
821    #[derive(Debug)]
822    struct NonCloneDispatchRow {
823        id: i64,
824        attempt_count: i32,
825        template_code: String,
826        to_address: String,
827        to_name: Option<String>,
828        payload_json: String,
829    }
830
831    impl MailOutboxDispatchRow for NonCloneDispatchRow {
832        fn id(&self) -> i64 {
833            self.id
834        }
835
836        fn attempt_count(&self) -> i32 {
837            self.attempt_count
838        }
839
840        fn template_code(&self) -> &str {
841            &self.template_code
842        }
843
844        fn to_address(&self) -> &str {
845            &self.to_address
846        }
847
848        fn to_name(&self) -> Option<&str> {
849            self.to_name.as_deref()
850        }
851    }
852
853    #[tokio::test]
854    async fn dispatch_mail_outbox_does_not_require_cloning_rows() {
855        let delivered_payload_len = Arc::new(AtomicUsize::new(0));
856        let delivered_payload_len_for_deliver = delivered_payload_len.clone();
857        let sent_context = Arc::new(std::sync::Mutex::new(None::<MailOutboxDispatchContext>));
858        let sent_context_for_hook = sent_context.clone();
859        let config = MailOutboxDispatchConfig::new(
860            20,
861            60,
862            1,
863            MailOutboxRetryPolicy::new(3, DEFAULT_ERROR_MAX_LEN),
864        );
865
866        let stats = dispatch_mail_outbox(
867            &config,
868            |_batch_size, _stale_secs| async {
869                Ok::<_, String>(vec![NonCloneDispatchRow {
870                    id: 7,
871                    attempt_count: 0,
872                    template_code: "login_email_code".to_string(),
873                    to_address: "operator@example.com".to_string(),
874                    to_name: Some("Operator".to_string()),
875                    payload_json: "x".repeat(4096),
876                }])
877            },
878            |id| async move { Ok::<_, String>(id == 7) },
879            move |row| {
880                let delivered_payload_len = delivered_payload_len_for_deliver.clone();
881                async move {
882                    delivered_payload_len.store(row.payload_json.len(), Ordering::SeqCst);
883                    Ok::<_, String>("Subject".to_string())
884                }
885            },
886            |_id, _attempt| async { Ok::<_, String>(true) },
887            |_id, _attempt_count, _retry_delay_secs, _error_message| async {
888                Ok::<_, String>(true)
889            },
890            |_id, _attempt_count, _error_message| async { Ok::<_, String>(true) },
891            move |context, _attempt_count, _subject| {
892                let sent_context = sent_context_for_hook.clone();
893                async move {
894                    *sent_context
895                        .lock()
896                        .expect("sent context mutex should not be poisoned") = Some(context);
897                }
898            },
899            |_context, _attempt_count, _error_message| async {},
900        )
901        .await
902        .expect("dispatch should succeed");
903
904        assert_eq!(
905            stats,
906            DispatchStats {
907                claimed: 1,
908                sent: 1,
909                retried: 0,
910                failed: 0,
911            }
912        );
913        assert_eq!(delivered_payload_len.load(Ordering::SeqCst), 4096);
914        let context = sent_context
915            .lock()
916            .expect("sent context mutex should not be poisoned")
917            .clone()
918            .expect("sent hook should receive context");
919        assert_eq!(context.id, 7);
920        assert_eq!(context.template_code, "login_email_code");
921        assert_eq!(context.to_address, "operator@example.com");
922        assert_eq!(context.to_name.as_deref(), Some("Operator"));
923    }
924
925    #[tokio::test]
926    async fn retry_mark_sent_retries_until_success() {
927        let attempts = Arc::new(AtomicUsize::new(0));
928        let attempts_for_closure = attempts.clone();
929
930        let updated = retry_mark_sent(42, &[0, 0, 0], move |_id, _attempt| {
931            let attempts = attempts_for_closure.clone();
932            async move {
933                let current = attempts.fetch_add(1, Ordering::SeqCst) + 1;
934                if current < 3 {
935                    Err("temporary db error")
936                } else {
937                    Ok(true)
938                }
939            }
940        })
941        .await
942        .expect("mark_sent should eventually succeed");
943
944        assert!(updated);
945        assert_eq!(attempts.load(Ordering::SeqCst), 3);
946    }
947
948    #[tokio::test]
949    async fn retry_mark_sent_without_delays_runs_one_attempt() {
950        let attempts = Arc::new(AtomicUsize::new(0));
951        let attempts_for_closure = attempts.clone();
952
953        let error = retry_mark_sent(42, &[], move |_id, _attempt| {
954            let attempts = attempts_for_closure.clone();
955            async move {
956                attempts.fetch_add(1, Ordering::SeqCst);
957                Err::<bool, _>("db down")
958            }
959        })
960        .await
961        .expect_err("mark_sent should fail");
962
963        assert_eq!(error, "db down");
964        assert_eq!(attempts.load(Ordering::SeqCst), 1);
965    }
966}
aster_forge_mail/outbox.rs

aster_forge_mail/
outbox.rs
