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

aster_forge_db/
mail_outbox.rs
