aster_forge_db/
mail_outbox.rs

1//! Database-backed mail outbox table and dispatch store.
2//!
3//! Aster products share the same mail outbox persistence mechanics: enqueue a
4//! rendered-template payload, claim due rows, move failed deliveries to retry or
5//! failed, clear sensitive payload JSON after terminal states, and count active
6//! rows. Products still own template rendering, audit records, and the business
7//! context that creates outbox rows.
8
9use std::future::Future;
10
11use chrono::{DateTime, Duration, Utc};
12use sea_orm::entity::prelude::*;
13use sea_orm::sea_query::{
14    Alias, ColumnDef, Index, IndexCreateStatement, Table, TableCreateStatement, TableDropStatement,
15};
16use sea_orm::{
17    ActiveEnum, ActiveModelTrait, ColumnTrait, Condition, ConnectionTrait, DatabaseBackend,
18    DatabaseConnection, EntityTrait, PaginatorTrait, QueryFilter, QueryOrder, QuerySelect, Set,
19    sea_query::Expr,
20};
21
22use crate::DbError;
23use aster_forge_mail::{
24    DispatchStats, MailOutboxDispatchConfig, MailOutboxDispatchContext, MailOutboxDispatchRow,
25    MailOutboxStatus, MailTemplateCode, StoredMailPayload,
26};
27
28/// Mail outbox table name.
29pub const MAIL_OUTBOX_TABLE: &str = "mail_outbox";
30/// Stable row id column.
31pub const MAIL_OUTBOX_ID_COLUMN: &str = "id";
32/// Template code column.
33pub const MAIL_OUTBOX_TEMPLATE_CODE_COLUMN: &str = "template_code";
34/// Recipient email address column.
35pub const MAIL_OUTBOX_TO_ADDRESS_COLUMN: &str = "to_address";
36/// Optional recipient display name column.
37pub const MAIL_OUTBOX_TO_NAME_COLUMN: &str = "to_name";
38/// Stored template payload JSON column.
39pub const MAIL_OUTBOX_PAYLOAD_JSON_COLUMN: &str = "payload_json";
40/// Dispatch status column.
41pub const MAIL_OUTBOX_STATUS_COLUMN: &str = "status";
42/// Delivery attempt count column.
43pub const MAIL_OUTBOX_ATTEMPT_COUNT_COLUMN: &str = "attempt_count";
44/// Next delivery attempt timestamp column.
45pub const MAIL_OUTBOX_NEXT_ATTEMPT_AT_COLUMN: &str = "next_attempt_at";
46/// Processing claim timestamp column.
47pub const MAIL_OUTBOX_PROCESSING_STARTED_AT_COLUMN: &str = "processing_started_at";
48/// Sent timestamp column.
49pub const MAIL_OUTBOX_SENT_AT_COLUMN: &str = "sent_at";
50/// Last delivery error column.
51pub const MAIL_OUTBOX_LAST_ERROR_COLUMN: &str = "last_error";
52/// Row creation timestamp column.
53pub const MAIL_OUTBOX_CREATED_AT_COLUMN: &str = "created_at";
54/// Row update timestamp column.
55pub const MAIL_OUTBOX_UPDATED_AT_COLUMN: &str = "updated_at";
56/// Index name for due-row dispatch scans.
57pub const MAIL_OUTBOX_DUE_INDEX: &str = "idx_mail_outbox_due";
58/// Index name for stale-processing recovery scans.
59pub const MAIL_OUTBOX_PROCESSING_INDEX: &str = "idx_mail_outbox_processing";
60/// Index name for sent-row retention scans.
61pub const MAIL_OUTBOX_SENT_AT_INDEX: &str = "idx_mail_outbox_sent_at";
62
63const MAIL_TEMPLATE_CODE_MAX_LEN: u32 = 64;
64const MAIL_TEMPLATE_CODE_MAX_BYTES: usize = 64;
65const MAIL_OUTBOX_TO_ADDRESS_MAX_LEN: usize = 255;
66const MAIL_OUTBOX_TO_NAME_MAX_LEN: usize = 255;
67
68/// Builds the shared `mail_outbox` table creation statement.
69pub fn create_mail_outbox_table(backend: DatabaseBackend) -> TableCreateStatement {
70    Table::create()
71        .table(mail_outbox_table())
72        .if_not_exists()
73        .col(
74            ColumnDef::new(mail_outbox_id())
75                .big_integer()
76                .not_null()
77                .auto_increment()
78                .primary_key(),
79        )
80        .col(
81            ColumnDef::new(mail_outbox_template_code())
82                .string_len(MAIL_TEMPLATE_CODE_MAX_LEN)
83                .not_null(),
84        )
85        .col(
86            ColumnDef::new(mail_outbox_to_address())
87                .string_len(255)
88                .not_null(),
89        )
90        .col(ColumnDef::new(mail_outbox_to_name()).string_len(255).null())
91        .col(ColumnDef::new(mail_outbox_payload_json()).text().not_null())
92        .col(
93            ColumnDef::new(mail_outbox_status())
94                .string_len(16)
95                .not_null(),
96        )
97        .col(
98            ColumnDef::new(mail_outbox_attempt_count())
99                .integer()
100                .not_null()
101                .default(0),
102        )
103        .col(utc_datetime_column(backend, mail_outbox_next_attempt_at()).not_null())
104        .col(utc_datetime_column(backend, mail_outbox_processing_started_at()).null())
105        .col(utc_datetime_column(backend, mail_outbox_sent_at()).null())
106        .col(ColumnDef::new(mail_outbox_last_error()).text().null())
107        .col(utc_datetime_column(backend, mail_outbox_created_at()).not_null())
108        .col(utc_datetime_column(backend, mail_outbox_updated_at()).not_null())
109        .to_owned()
110}
111
112/// Builds the shared `mail_outbox` table drop statement.
113pub fn drop_mail_outbox_table() -> TableDropStatement {
114    Table::drop()
115        .table(mail_outbox_table())
116        .if_exists()
117        .to_owned()
118}
119
120/// Builds the due-row index used by dispatch polling.
121pub fn create_mail_outbox_due_index() -> IndexCreateStatement {
122    Index::create()
123        .name(MAIL_OUTBOX_DUE_INDEX)
124        .table(mail_outbox_table())
125        .col(mail_outbox_status())
126        .col(mail_outbox_next_attempt_at())
127        .col(mail_outbox_created_at())
128        .if_not_exists()
129        .to_owned()
130}
131
132/// Builds the processing-stale index used by dispatch recovery.
133pub fn create_mail_outbox_processing_index() -> IndexCreateStatement {
134    Index::create()
135        .name(MAIL_OUTBOX_PROCESSING_INDEX)
136        .table(mail_outbox_table())
137        .col(mail_outbox_status())
138        .col(mail_outbox_processing_started_at())
139        .col(mail_outbox_created_at())
140        .if_not_exists()
141        .to_owned()
142}
143
144/// Builds the sent timestamp index used by retention and admin queries.
145pub fn create_mail_outbox_sent_at_index() -> IndexCreateStatement {
146    Index::create()
147        .name(MAIL_OUTBOX_SENT_AT_INDEX)
148        .table(mail_outbox_table())
149        .col(mail_outbox_sent_at())
150        .if_not_exists()
151        .to_owned()
152}
153
154fn mail_outbox_table() -> Alias {
155    Alias::new(MAIL_OUTBOX_TABLE)
156}
157
158fn mail_outbox_id() -> Alias {
159    Alias::new(MAIL_OUTBOX_ID_COLUMN)
160}
161
162fn mail_outbox_template_code() -> Alias {
163    Alias::new(MAIL_OUTBOX_TEMPLATE_CODE_COLUMN)
164}
165
166fn mail_outbox_to_address() -> Alias {
167    Alias::new(MAIL_OUTBOX_TO_ADDRESS_COLUMN)
168}
169
170fn mail_outbox_to_name() -> Alias {
171    Alias::new(MAIL_OUTBOX_TO_NAME_COLUMN)
172}
173
174fn mail_outbox_payload_json() -> Alias {
175    Alias::new(MAIL_OUTBOX_PAYLOAD_JSON_COLUMN)
176}
177
178fn mail_outbox_status() -> Alias {
179    Alias::new(MAIL_OUTBOX_STATUS_COLUMN)
180}
181
182fn mail_outbox_attempt_count() -> Alias {
183    Alias::new(MAIL_OUTBOX_ATTEMPT_COUNT_COLUMN)
184}
185
186fn mail_outbox_next_attempt_at() -> Alias {
187    Alias::new(MAIL_OUTBOX_NEXT_ATTEMPT_AT_COLUMN)
188}
189
190fn mail_outbox_processing_started_at() -> Alias {
191    Alias::new(MAIL_OUTBOX_PROCESSING_STARTED_AT_COLUMN)
192}
193
194fn mail_outbox_sent_at() -> Alias {
195    Alias::new(MAIL_OUTBOX_SENT_AT_COLUMN)
196}
197
198fn mail_outbox_last_error() -> Alias {
199    Alias::new(MAIL_OUTBOX_LAST_ERROR_COLUMN)
200}
201
202fn mail_outbox_created_at() -> Alias {
203    Alias::new(MAIL_OUTBOX_CREATED_AT_COLUMN)
204}
205
206fn mail_outbox_updated_at() -> Alias {
207    Alias::new(MAIL_OUTBOX_UPDATED_AT_COLUMN)
208}
209
210fn utc_datetime_column(backend: DatabaseBackend, column: Alias) -> ColumnDef {
211    let mut definition = ColumnDef::new(column);
212    match backend {
213        DatabaseBackend::MySql => {
214            definition.custom(Alias::new("datetime(6)"));
215        }
216        _ => {
217            definition.timestamp_with_time_zone();
218        }
219    }
220    definition
221}
222
223/// SeaORM model for `mail_outbox`.
224#[derive(Clone, Debug, PartialEq, DeriveEntityModel)]
225#[sea_orm(table_name = "mail_outbox")]
226pub struct Model {
227    /// Stable row id.
228    #[sea_orm(primary_key)]
229    pub id: i64,
230    /// Shared Aster mail template code.
231    pub template_code: MailTemplateCode,
232    /// Recipient email address.
233    pub to_address: String,
234    /// Optional recipient display name.
235    pub to_name: Option<String>,
236    /// Stored template payload JSON.
237    pub payload_json: StoredMailPayload,
238    /// Dispatch status.
239    pub status: MailOutboxStatus,
240    /// Delivery attempt count.
241    pub attempt_count: i32,
242    /// Next delivery attempt timestamp.
243    pub next_attempt_at: DateTimeUtc,
244    /// Processing claim timestamp.
245    pub processing_started_at: Option<DateTimeUtc>,
246    /// Sent timestamp.
247    pub sent_at: Option<DateTimeUtc>,
248    /// Last delivery error.
249    pub last_error: Option<String>,
250    /// Row creation timestamp.
251    pub created_at: DateTimeUtc,
252    /// Row update timestamp.
253    pub updated_at: DateTimeUtc,
254}
255
256impl MailOutboxDispatchRow for Model {
257    fn id(&self) -> i64 {
258        self.id
259    }
260
261    fn attempt_count(&self) -> i32 {
262        self.attempt_count
263    }
264
265    fn template_code(&self) -> &str {
266        self.template_code.as_str()
267    }
268
269    fn to_address(&self) -> &str {
270        &self.to_address
271    }
272
273    fn to_name(&self) -> Option<&str> {
274        self.to_name.as_deref()
275    }
276}
277
278#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
279pub enum Relation {}
280
281impl ActiveModelBehavior for ActiveModel {}
282
283/// Product request to enqueue one mail outbox row.
284#[derive(Debug, Clone, PartialEq, Eq)]
285pub struct MailOutboxCreate {
286    /// Shared Aster mail template code.
287    pub template_code: MailTemplateCode,
288    /// Recipient email address.
289    pub to_address: String,
290    /// Optional recipient display name.
291    pub to_name: Option<String>,
292    /// Stored template payload JSON.
293    pub payload_json: StoredMailPayload,
294    /// Initial next attempt timestamp.
295    pub next_attempt_at: DateTime<Utc>,
296    /// Row creation/update timestamp.
297    pub now: DateTime<Utc>,
298}
299
300/// SeaORM-backed mail outbox store.
301#[derive(Clone)]
302pub struct MailOutboxDbStore {
303    db: DatabaseConnection,
304}
305
306impl MailOutboxDbStore {
307    /// Creates a mail outbox store from a SeaORM database connection.
308    pub const fn new(db: DatabaseConnection) -> Self {
309        Self { db }
310    }
311
312    /// Enqueues one pending mail outbox row.
313    pub async fn create(&self, request: MailOutboxCreate) -> crate::Result<Model> {
314        create_mail_outbox_row(&self.db, request).await
315    }
316
317    /// Lists rows that are due or stale enough to be reclaimed.
318    pub async fn list_claimable(
319        &self,
320        now: DateTime<Utc>,
321        stale_before: DateTime<Utc>,
322        limit: u64,
323    ) -> crate::Result<Vec<Model>> {
324        list_claimable(&self.db, now, stale_before, limit).await
325    }
326
327    /// Attempts to claim one row for processing.
328    pub async fn try_claim(
329        &self,
330        id: i64,
331        now: DateTime<Utc>,
332        stale_before: DateTime<Utc>,
333    ) -> crate::Result<bool> {
334        try_claim(&self.db, id, now, stale_before).await
335    }
336
337    /// Marks a processing row as sent and clears its sensitive payload.
338    pub async fn mark_sent(&self, id: i64, sent_at: DateTime<Utc>) -> crate::Result<bool> {
339        mark_sent(&self.db, id, sent_at).await
340    }
341
342    /// Marks a processing row for retry.
343    pub async fn mark_retry(
344        &self,
345        id: i64,
346        attempt_count: i32,
347        next_attempt_at: DateTime<Utc>,
348        last_error: &str,
349    ) -> crate::Result<bool> {
350        mark_retry(&self.db, id, attempt_count, next_attempt_at, last_error).await
351    }
352
353    /// Marks a processing row as permanently failed and clears its sensitive payload.
354    pub async fn mark_failed(
355        &self,
356        id: i64,
357        attempt_count: i32,
358        failed_at: DateTime<Utc>,
359        last_error: &str,
360    ) -> crate::Result<bool> {
361        mark_failed(&self.db, id, attempt_count, failed_at, last_error).await
362    }
363
364    /// Counts pending or retry rows.
365    pub async fn count_active(&self) -> crate::Result<u64> {
366        count_active(&self.db).await
367    }
368
369    /// Runs one shared dispatch pass using the standard database-backed outbox mechanics.
370    ///
371    /// Forge owns list/claim/mark/retry/failure persistence for the shared `mail_outbox` table.
372    /// Products only provide rendering/delivery and audit hooks, which keeps every Aster service on
373    /// the same state machine without copying payload-heavy rows across persistence callbacks.
374    pub async fn dispatch_due<E, Deliver, DeliverFut, OnSent, OnSentFut, OnFailed, OnFailedFut>(
375        &self,
376        config: &MailOutboxDispatchConfig,
377        deliver: Deliver,
378        on_sent: OnSent,
379        on_failed: OnFailed,
380    ) -> std::result::Result<DispatchStats, E>
381    where
382        E: From<DbError> + std::fmt::Display,
383        Deliver: FnMut(Model) -> DeliverFut,
384        DeliverFut: Future<Output = std::result::Result<String, E>>,
385        OnSent: FnMut(MailOutboxDispatchContext, i32, String) -> OnSentFut,
386        OnSentFut: Future<Output = ()>,
387        OnFailed: FnMut(MailOutboxDispatchContext, i32, String) -> OnFailedFut,
388        OnFailedFut: Future<Output = ()>,
389    {
390        aster_forge_mail::dispatch_mail_outbox(
391            config,
392            |batch_size, stale_secs| async move {
393                let now = Utc::now();
394                let stale_before = now - Duration::seconds(stale_secs);
395                self.list_claimable(now, stale_before, batch_size)
396                    .await
397                    .map_err(E::from)
398            },
399            |id| async move {
400                let now = Utc::now();
401                let stale_before = now - Duration::seconds(config.processing_stale_secs);
402                self.try_claim(id, now, stale_before).await.map_err(E::from)
403            },
404            deliver,
405            |id, _attempt| async move { self.mark_sent(id, Utc::now()).await.map_err(E::from) },
406            |id, attempt_count, retry_delay_secs, error_message| async move {
407                let retry_at = Utc::now() + Duration::seconds(retry_delay_secs);
408                self.mark_retry(id, attempt_count, retry_at, &error_message)
409                    .await
410                    .map_err(E::from)
411            },
412            |id, attempt_count, error_message| async move {
413                self.mark_failed(id, attempt_count, Utc::now(), &error_message)
414                    .await
415                    .map_err(E::from)
416            },
417            on_sent,
418            on_failed,
419        )
420        .await
421    }
422}
423
424/// Enqueues one pending mail outbox row using any SeaORM connection or transaction.
425pub async fn create_mail_outbox_row<C>(db: &C, request: MailOutboxCreate) -> crate::Result<Model>
426where
427    C: ConnectionTrait,
428{
429    validate_create(&request)?;
430    ActiveModel {
431        template_code: Set(request.template_code),
432        to_address: Set(request.to_address),
433        to_name: Set(request.to_name),
434        payload_json: Set(request.payload_json),
435        status: Set(MailOutboxStatus::Pending),
436        attempt_count: Set(0),
437        next_attempt_at: Set(request.next_attempt_at),
438        processing_started_at: Set(None),
439        sent_at: Set(None),
440        last_error: Set(None),
441        created_at: Set(request.now),
442        updated_at: Set(request.now),
443        ..Default::default()
444    }
445    .insert(db)
446    .await
447    .map_err(DbError::from)
448}
449
450async fn list_claimable<C>(
451    db: &C,
452    now: DateTime<Utc>,
453    stale_before: DateTime<Utc>,
454    limit: u64,
455) -> crate::Result<Vec<Model>>
456where
457    C: ConnectionTrait,
458{
459    Entity::find()
460        .filter(claimable_condition(now, stale_before))
461        .order_by_asc(Column::CreatedAt)
462        .limit(limit)
463        .all(db)
464        .await
465        .map_err(DbError::from)
466}
467
468async fn try_claim<C>(
469    db: &C,
470    id: i64,
471    now: DateTime<Utc>,
472    stale_before: DateTime<Utc>,
473) -> crate::Result<bool>
474where
475    C: ConnectionTrait,
476{
477    let result = Entity::update_many()
478        .col_expr(
479            Column::Status,
480            Expr::value(MailOutboxStatus::Processing.to_value()),
481        )
482        .col_expr(Column::ProcessingStartedAt, Expr::value(Some(now)))
483        .col_expr(Column::UpdatedAt, Expr::value(now))
484        .filter(Column::Id.eq(id))
485        .filter(claimable_condition(now, stale_before))
486        .exec(db)
487        .await
488        .map_err(DbError::from)?;
489    Ok(result.rows_affected == 1)
490}
491
492async fn mark_sent<C>(db: &C, id: i64, sent_at: DateTime<Utc>) -> crate::Result<bool>
493where
494    C: ConnectionTrait,
495{
496    let result = Entity::update_many()
497        .col_expr(
498            Column::Status,
499            Expr::value(MailOutboxStatus::Sent.to_value()),
500        )
501        .col_expr(Column::SentAt, Expr::value(Some(sent_at)))
502        .col_expr(
503            Column::ProcessingStartedAt,
504            Expr::value(Option::<DateTime<Utc>>::None),
505        )
506        .col_expr(Column::LastError, Expr::value(Option::<String>::None))
507        .col_expr(
508            Column::PayloadJson,
509            Expr::value(StoredMailPayload::CLEARED_JSON),
510        )
511        .col_expr(Column::UpdatedAt, Expr::value(sent_at))
512        .filter(Column::Id.eq(id))
513        .filter(Column::Status.eq(MailOutboxStatus::Processing))
514        .exec(db)
515        .await
516        .map_err(DbError::from)?;
517    Ok(result.rows_affected == 1)
518}
519
520async fn mark_retry<C>(
521    db: &C,
522    id: i64,
523    attempt_count: i32,
524    next_attempt_at: DateTime<Utc>,
525    last_error: &str,
526) -> crate::Result<bool>
527where
528    C: ConnectionTrait,
529{
530    let result = Entity::update_many()
531        .col_expr(
532            Column::Status,
533            Expr::value(MailOutboxStatus::Retry.to_value()),
534        )
535        .col_expr(Column::AttemptCount, Expr::value(attempt_count))
536        .col_expr(Column::NextAttemptAt, Expr::value(next_attempt_at))
537        .col_expr(
538            Column::ProcessingStartedAt,
539            Expr::value(Option::<DateTime<Utc>>::None),
540        )
541        .col_expr(Column::LastError, Expr::value(Some(last_error)))
542        .col_expr(Column::UpdatedAt, Expr::value(Utc::now()))
543        .filter(Column::Id.eq(id))
544        .filter(Column::Status.eq(MailOutboxStatus::Processing))
545        .exec(db)
546        .await
547        .map_err(DbError::from)?;
548    Ok(result.rows_affected == 1)
549}
550
551async fn mark_failed<C>(
552    db: &C,
553    id: i64,
554    attempt_count: i32,
555    failed_at: DateTime<Utc>,
556    last_error: &str,
557) -> crate::Result<bool>
558where
559    C: ConnectionTrait,
560{
561    let result = Entity::update_many()
562        .col_expr(
563            Column::Status,
564            Expr::value(MailOutboxStatus::Failed.to_value()),
565        )
566        .col_expr(Column::AttemptCount, Expr::value(attempt_count))
567        .col_expr(Column::NextAttemptAt, Expr::value(failed_at))
568        .col_expr(
569            Column::ProcessingStartedAt,
570            Expr::value(Option::<DateTime<Utc>>::None),
571        )
572        .col_expr(Column::LastError, Expr::value(Some(last_error)))
573        .col_expr(
574            Column::PayloadJson,
575            Expr::value(StoredMailPayload::CLEARED_JSON),
576        )
577        .col_expr(Column::UpdatedAt, Expr::value(failed_at))
578        .filter(Column::Id.eq(id))
579        .filter(Column::Status.eq(MailOutboxStatus::Processing))
580        .exec(db)
581        .await
582        .map_err(DbError::from)?;
583    Ok(result.rows_affected == 1)
584}
585
586async fn count_active<C>(db: &C) -> crate::Result<u64>
587where
588    C: ConnectionTrait,
589{
590    Entity::find()
591        .filter(Column::Status.is_in([MailOutboxStatus::Pending, MailOutboxStatus::Retry]))
592        .count(db)
593        .await
594        .map_err(DbError::from)
595}
596
597fn claimable_condition(now: DateTime<Utc>, stale_before: DateTime<Utc>) -> Condition {
598    Condition::any()
599        .add(
600            Condition::all()
601                .add(Column::Status.is_in([MailOutboxStatus::Pending, MailOutboxStatus::Retry]))
602                .add(Column::NextAttemptAt.lte(now)),
603        )
604        .add(
605            Condition::all()
606                .add(Column::Status.eq(MailOutboxStatus::Processing))
607                .add(Column::ProcessingStartedAt.lte(stale_before)),
608        )
609}
610
611fn validate_create(request: &MailOutboxCreate) -> crate::Result<()> {
612    validate_non_empty("mail outbox recipient address", &request.to_address)?;
613    validate_max_len(
614        "mail outbox recipient address",
615        &request.to_address,
616        MAIL_OUTBOX_TO_ADDRESS_MAX_LEN,
617    )?;
618    if let Some(to_name) = &request.to_name {
619        validate_max_len(
620            "mail outbox recipient name",
621            to_name,
622            MAIL_OUTBOX_TO_NAME_MAX_LEN,
623        )?;
624    }
625    validate_max_len(
626        "mail outbox template code",
627        request.template_code.as_str(),
628        MAIL_TEMPLATE_CODE_MAX_BYTES,
629    )
630}
631
632fn validate_non_empty(name: &str, value: &str) -> crate::Result<()> {
633    if value.trim().is_empty() {
634        return Err(DbError::non_retryable(format!("{name} must not be empty")));
635    }
636    Ok(())
637}
638
639fn validate_max_len(name: &str, value: &str, max_len: usize) -> crate::Result<()> {
640    if value.len() > max_len {
641        return Err(DbError::non_retryable(format!(
642            "{name} must be at most {max_len} bytes"
643        )));
644    }
645    Ok(())
646}
647
648#[cfg(test)]
649mod tests {
650    use chrono::{Duration as ChronoDuration, TimeZone, Utc};
651    use sea_orm::sea_query::{MysqlQueryBuilder, PostgresQueryBuilder, SqliteQueryBuilder};
652    use sea_orm::{ConnectionTrait, Database, DatabaseBackend, EntityTrait};
653    use std::sync::{
654        Arc, Mutex,
655        atomic::{AtomicUsize, Ordering},
656    };
657
658    use super::{
659        Entity, MailOutboxCreate, MailOutboxDbStore, create_mail_outbox_due_index,
660        create_mail_outbox_processing_index, create_mail_outbox_sent_at_index,
661        create_mail_outbox_table,
662    };
663    use crate::DbError;
664    use aster_forge_mail::{
665        DEFAULT_ERROR_MAX_LEN, MailOutboxDispatchConfig, MailOutboxDispatchContext,
666        MailOutboxRetryPolicy, MailOutboxStatus, MailTemplateCode, StoredMailPayload,
667    };
668
669    async fn sqlite_store() -> MailOutboxDbStore {
670        let db = Database::connect("sqlite::memory:")
671            .await
672            .expect("sqlite memory database should connect");
673        let backend = db.get_database_backend();
674        db.execute(&create_mail_outbox_table(backend))
675            .await
676            .expect("mail outbox table builder should execute");
677        db.execute(&create_mail_outbox_due_index())
678            .await
679            .expect("mail outbox due index builder should execute");
680        db.execute(&create_mail_outbox_processing_index())
681            .await
682            .expect("mail outbox processing index builder should execute");
683        db.execute(&create_mail_outbox_sent_at_index())
684            .await
685            .expect("mail outbox sent index builder should execute");
686        MailOutboxDbStore::new(db)
687    }
688
689    fn create_table_sql(backend: DatabaseBackend) -> String {
690        let table = create_mail_outbox_table(backend);
691        match backend {
692            DatabaseBackend::MySql => table.to_string(MysqlQueryBuilder),
693            DatabaseBackend::Postgres => table.to_string(PostgresQueryBuilder),
694            DatabaseBackend::Sqlite => table.to_string(SqliteQueryBuilder),
695            _ => unreachable!("unsupported backend in mail outbox table test"),
696        }
697    }
698
699    fn create_request(now: chrono::DateTime<Utc>) -> MailOutboxCreate {
700        MailOutboxCreate {
701            template_code: MailTemplateCode::LoginEmailCode,
702            to_address: "operator@example.com".to_string(),
703            to_name: Some("Operator".to_string()),
704            payload_json: StoredMailPayload::from("{\"code\":\"123456\"}".to_string()),
705            next_attempt_at: now,
706            now,
707        }
708    }
709
710    #[test]
711    fn create_mail_outbox_table_uses_stable_shape() {
712        let sqlite_sql = create_table_sql(DatabaseBackend::Sqlite);
713        assert!(sqlite_sql.contains("CREATE TABLE IF NOT EXISTS \"mail_outbox\""));
714        assert!(sqlite_sql.contains("\"template_code\" varchar(64) NOT NULL"));
715        assert!(sqlite_sql.contains("\"status\" varchar(16) NOT NULL"));
716        assert!(sqlite_sql.contains("\"next_attempt_at\" timestamp_with_timezone_text NOT NULL"));
717        let due_index = create_mail_outbox_due_index().to_string(SqliteQueryBuilder);
718        assert!(due_index.contains("idx_mail_outbox_due"));
719        assert!(due_index.contains("\"status\", \"next_attempt_at\", \"created_at\""));
720
721        let mysql_sql = create_table_sql(DatabaseBackend::MySql);
722        assert!(mysql_sql.contains("`next_attempt_at` datetime(6) NOT NULL"));
723
724        let postgres_sql = create_table_sql(DatabaseBackend::Postgres);
725        assert!(postgres_sql.contains("\"next_attempt_at\" timestamp with time zone NOT NULL"));
726    }
727
728    #[tokio::test]
729    async fn mail_outbox_store_creates_and_counts_active_rows() {
730        let store = sqlite_store().await;
731        let now = Utc.with_ymd_and_hms(2026, 6, 26, 1, 0, 0).unwrap();
732
733        let row = store
734            .create(create_request(now))
735            .await
736            .expect("mail outbox row should insert");
737
738        assert_eq!(row.template_code, MailTemplateCode::LoginEmailCode);
739        assert_eq!(row.status, MailOutboxStatus::Pending);
740        assert_eq!(store.count_active().await.expect("count should query"), 1);
741    }
742
743    #[tokio::test]
744    async fn mail_outbox_store_rejects_invalid_create_values() {
745        let store = sqlite_store().await;
746        let now = Utc.with_ymd_and_hms(2026, 6, 26, 1, 0, 0).unwrap();
747
748        let error = store
749            .create(MailOutboxCreate {
750                to_address: " ".to_string(),
751                ..create_request(now)
752            })
753            .await
754            .expect_err("blank recipient should be rejected");
755        assert!(error.to_string().contains("must not be empty"));
756    }
757
758    #[tokio::test]
759    async fn mail_outbox_store_claims_due_rows_once() {
760        let store = sqlite_store().await;
761        let now = Utc.with_ymd_and_hms(2026, 6, 26, 1, 0, 0).unwrap();
762        let row = store
763            .create(create_request(now))
764            .await
765            .expect("mail outbox row should insert");
766
767        let claimable = store
768            .list_claimable(now, now - ChronoDuration::minutes(5), 10)
769            .await
770            .expect("claimable rows should query");
771        assert_eq!(claimable.len(), 1);
772
773        assert!(
774            store
775                .try_claim(row.id, now, now - ChronoDuration::minutes(5))
776                .await
777                .expect("claim should query")
778        );
779        assert!(
780            !store
781                .try_claim(
782                    row.id,
783                    now + ChronoDuration::seconds(1),
784                    now - ChronoDuration::minutes(5)
785                )
786                .await
787                .expect("fresh duplicate claim should query")
788        );
789    }
790
791    #[tokio::test]
792    async fn mail_outbox_store_reclaims_stale_processing_rows() {
793        let store = sqlite_store().await;
794        let now = Utc.with_ymd_and_hms(2026, 6, 26, 1, 0, 0).unwrap();
795        let row = store
796            .create(create_request(now))
797            .await
798            .expect("mail outbox row should insert");
799        assert!(
800            store
801                .try_claim(row.id, now, now - ChronoDuration::minutes(5))
802                .await
803                .expect("claim should query")
804        );
805
806        let reclaimed = store
807            .list_claimable(
808                now + ChronoDuration::minutes(10),
809                now + ChronoDuration::minutes(1),
810                10,
811            )
812            .await
813            .expect("stale processing rows should query");
814        assert_eq!(reclaimed.len(), 1);
815    }
816
817    #[tokio::test]
818    async fn mail_outbox_store_marks_sent_and_clears_payload() {
819        let store = sqlite_store().await;
820        let now = Utc.with_ymd_and_hms(2026, 6, 26, 1, 0, 0).unwrap();
821        let row = store
822            .create(create_request(now))
823            .await
824            .expect("mail outbox row should insert");
825        assert!(
826            store
827                .try_claim(row.id, now, now - ChronoDuration::minutes(5))
828                .await
829                .expect("claim should query")
830        );
831
832        assert!(
833            store
834                .mark_sent(row.id, now + ChronoDuration::seconds(2))
835                .await
836                .expect("mark sent should query")
837        );
838
839        let stored = Entity::find_by_id(row.id)
840            .one(&store.db)
841            .await
842            .expect("sent row should query")
843            .expect("sent row should exist");
844        assert_eq!(stored.status, MailOutboxStatus::Sent);
845        assert_eq!(
846            stored.payload_json.as_ref(),
847            StoredMailPayload::CLEARED_JSON
848        );
849        assert_eq!(store.count_active().await.expect("count should query"), 0);
850    }
851
852    #[tokio::test]
853    async fn mail_outbox_store_marks_retry_and_failed_only_from_processing() {
854        let store = sqlite_store().await;
855        let now = Utc.with_ymd_and_hms(2026, 6, 26, 1, 0, 0).unwrap();
856        let row = store
857            .create(create_request(now))
858            .await
859            .expect("mail outbox row should insert");
860        assert!(
861            !store
862                .mark_retry(row.id, 1, now + ChronoDuration::seconds(5), "smtp down")
863                .await
864                .expect("retry without processing should query")
865        );
866        assert!(
867            store
868                .try_claim(row.id, now, now - ChronoDuration::minutes(5))
869                .await
870                .expect("claim should query")
871        );
872        assert!(
873            store
874                .mark_retry(row.id, 1, now + ChronoDuration::seconds(5), "smtp down")
875                .await
876                .expect("retry should query")
877        );
878
879        let retry = Entity::find_by_id(row.id)
880            .one(&store.db)
881            .await
882            .expect("retry row should query")
883            .expect("retry row should exist");
884        assert_eq!(retry.status, MailOutboxStatus::Retry);
885        assert_eq!(retry.attempt_count, 1);
886        assert_eq!(retry.last_error.as_deref(), Some("smtp down"));
887
888        assert!(
889            store
890                .try_claim(row.id, now + ChronoDuration::seconds(6), now)
891                .await
892                .expect("retry claim should query")
893        );
894        assert!(
895            store
896                .mark_failed(row.id, 2, now + ChronoDuration::seconds(7), "permanent")
897                .await
898                .expect("failed should query")
899        );
900
901        let failed = Entity::find_by_id(row.id)
902            .one(&store.db)
903            .await
904            .expect("failed row should query")
905            .expect("failed row should exist");
906        assert_eq!(failed.status, MailOutboxStatus::Failed);
907        assert_eq!(
908            failed.payload_json.as_ref(),
909            StoredMailPayload::CLEARED_JSON
910        );
911    }
912
913    #[tokio::test]
914    async fn mail_outbox_store_dispatch_due_marks_sent_and_reports_context() {
915        let store = sqlite_store().await;
916        let now = Utc::now();
917        let row = store
918            .create(create_request(now - ChronoDuration::seconds(5)))
919            .await
920            .expect("mail outbox row should insert");
921        let config = MailOutboxDispatchConfig::new(
922            20,
923            60,
924            1,
925            MailOutboxRetryPolicy::new(3, DEFAULT_ERROR_MAX_LEN),
926        );
927        let delivered_payload_len = Arc::new(AtomicUsize::new(0));
928        let delivered_payload_len_for_deliver = delivered_payload_len.clone();
929        let sent_context = Arc::new(Mutex::new(None::<MailOutboxDispatchContext>));
930        let sent_context_for_hook = sent_context.clone();
931
932        let stats = store
933            .dispatch_due(
934                &config,
935                move |row| {
936                    let delivered_payload_len = delivered_payload_len_for_deliver.clone();
937                    async move {
938                        delivered_payload_len
939                            .store(row.payload_json.as_ref().len(), Ordering::SeqCst);
940                        Ok::<_, DbError>("Sent subject".to_string())
941                    }
942                },
943                move |context, _attempt_count, _subject| {
944                    let sent_context = sent_context_for_hook.clone();
945                    async move {
946                        *sent_context
947                            .lock()
948                            .expect("sent context mutex should not be poisoned") = Some(context);
949                    }
950                },
951                |_context, _attempt_count, _error_message| async {},
952            )
953            .await
954            .expect("dispatch should succeed");
955
956        assert_eq!(stats.sent, 1);
957        assert_eq!(stats.claimed, 1);
958        assert_eq!(
959            delivered_payload_len.load(Ordering::SeqCst),
960            "{\"code\":\"123456\"}".len()
961        );
962        let context = sent_context
963            .lock()
964            .expect("sent context mutex should not be poisoned")
965            .clone()
966            .expect("sent hook should receive context");
967        assert_eq!(context.id, row.id);
968        assert_eq!(context.to_name.as_deref(), Some("Operator"));
969
970        let stored = Entity::find_by_id(row.id)
971            .one(&store.db)
972            .await
973            .expect("sent row should query")
974            .expect("sent row should exist");
975        assert_eq!(stored.status, MailOutboxStatus::Sent);
976        assert_eq!(
977            stored.payload_json.as_ref(),
978            StoredMailPayload::CLEARED_JSON
979        );
980    }
981
982    #[tokio::test]
983    async fn mail_outbox_store_dispatch_due_retries_delivery_errors() {
984        let store = sqlite_store().await;
985        let now = Utc::now();
986        let row = store
987            .create(create_request(now - ChronoDuration::seconds(5)))
988            .await
989            .expect("mail outbox row should insert");
990        let config = MailOutboxDispatchConfig::new(
991            20,
992            60,
993            1,
994            MailOutboxRetryPolicy::new(3, DEFAULT_ERROR_MAX_LEN),
995        );
996
997        let stats = store
998            .dispatch_due(
999                &config,
1000                |_row| async { Err::<String, _>(DbError::non_retryable("smtp down")) },
1001                |_context, _attempt_count, _subject| async {},
1002                |_context, _attempt_count, _error_message| async {},
1003            )
1004            .await
1005            .expect("dispatch should handle delivery failure as retry state");
1006
1007        assert_eq!(stats.claimed, 1);
1008        assert_eq!(stats.retried, 1);
1009        let stored = Entity::find_by_id(row.id)
1010            .one(&store.db)
1011            .await
1012            .expect("retry row should query")
1013            .expect("retry row should exist");
1014        assert_eq!(stored.status, MailOutboxStatus::Retry);
1015        assert_eq!(stored.attempt_count, 1);
1016        assert_eq!(
1017            stored.last_error.as_deref(),
1018            Some("non-retryable error: smtp down")
1019        );
1020    }
1021}
aster_forge_db/mail_outbox.rs

aster_forge_db/
mail_outbox.rs
