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

aster_forge_mail/
outbox.rs
