mailrs_outbound_queue/queue/
mod.rs

1#[cfg(feature = "pg")]
2use crate::BackendPool;
3#[cfg(feature = "pg")]
4use kevy_embedded::Store;
5
6mod admin;
7mod lifecycle;
8pub mod suppression;
9
10pub use admin::*;
11pub use lifecycle::*;
12
13/// Lifecycle status of a queued message.
14#[derive(Debug, Clone, PartialEq, Eq)]
15pub enum QueueStatus {
16    /// Awaiting first delivery attempt.
17    Pending,
18    /// Currently being delivered by a worker.
19    InFlight,
20    /// Successfully accepted by the remote MX.
21    Delivered,
22    /// Last attempt failed; will retry per backoff schedule.
23    Failed,
24    /// Permanent failure; will NOT retry. A DSN has been queued back.
25    Bounced,
26}
27
28impl QueueStatus {
29    /// Lower-snake-case rendering for SQL persistence.
30    pub fn as_str(&self) -> &'static str {
31        match self {
32            Self::Pending => "pending",
33            Self::InFlight => "inflight",
34            Self::Delivered => "delivered",
35            Self::Failed => "failed",
36            Self::Bounced => "bounced",
37        }
38    }
39
40    /// Inverse of [`Self::as_str`]; `None` on unknown values.
41    pub fn parse(s: &str) -> Option<Self> {
42        match s {
43            "pending" => Some(Self::Pending),
44            "inflight" => Some(Self::InFlight),
45            "delivered" => Some(Self::Delivered),
46            "failed" => Some(Self::Failed),
47            "bounced" => Some(Self::Bounced),
48            _ => None,
49        }
50    }
51}
52
53/// One queued outbound message — the full row stored in the outbound queue.
54#[derive(Debug, Clone)]
55pub struct QueuedMessage {
56    /// Store-native primary key.
57    pub id: i64,
58    /// Envelope sender (reverse path).
59    pub sender: String,
60    /// Envelope recipient (single forward path; multi-recipient messages
61    /// fan out into one row per recipient).
62    pub recipient: String,
63    /// Recipient's domain — extracted for MX-grouped batching.
64    pub domain: String,
65    /// Full RFC 5322 message body (including headers).
66    pub message_data: Vec<u8>,
67    /// Current lifecycle status.
68    pub status: QueueStatus,
69    /// Number of delivery attempts made so far.
70    pub attempts: u32,
71    /// Cap after which `Failed` flips to `Bounced`.
72    pub max_attempts: u32,
73    /// Epoch seconds — the earliest time the next attempt is eligible.
74    pub next_retry: i64,
75    /// Last error response from the remote MX, if any.
76    pub last_error: Option<String>,
77    /// `Message-ID:` header value, for log correlation.
78    pub message_id: Option<String>,
79    /// Epoch seconds when the row was first enqueued.
80    pub created_at: i64,
81    /// Epoch seconds of the most recent update.
82    pub updated_at: i64,
83    /// `true` when the message came from a forwarding rule rather than a
84    /// local sender.
85    pub is_forwarded: bool,
86}
87
88/// enqueue a message for outbound delivery
89#[cfg(feature = "pg")]
90pub async fn enqueue(
91    pool: &BackendPool,
92    sender: &str,
93    recipient: &str,
94    domain: &str,
95    message_data: &[u8],
96    message_id: Option<&str>,
97    now: i64,
98) -> Result<i64, sqlx::Error> {
99    enqueue_ex(
100        pool,
101        sender,
102        recipient,
103        domain,
104        message_data,
105        message_id,
106        now,
107        false,
108    )
109    .await
110}
111
112/// enqueue a message for outbound delivery with forwarding flag
113#[allow(clippy::too_many_arguments)]
114#[cfg(feature = "pg")]
115pub async fn enqueue_ex(
116    pool: &BackendPool,
117    sender: &str,
118    recipient: &str,
119    domain: &str,
120    message_data: &[u8],
121    message_id: Option<&str>,
122    now: i64,
123    is_forwarded: bool,
124) -> Result<i64, sqlx::Error> {
125    let row: (i64,) = sqlx::query_as(
126        "INSERT INTO outbound_queue (sender, recipient, domain, message_data, status, next_retry, message_id, created_at, updated_at, is_forwarded)
127         VALUES ($1, $2, $3, $4, 'pending', $5, $6, $5, $5, $7)
128         RETURNING id",
129    )
130    .bind(sender)
131    .bind(recipient)
132    .bind(domain)
133    .bind(message_data)
134    .bind(now)
135    .bind(message_id)
136    .bind(is_forwarded)
137    .fetch_one(pool)
138    .await?;
139    Ok(row.0)
140}
141
142/// notify the delivery worker that new messages are queued via the
143/// in-process kevy `queue:notify` channel.
144#[cfg(feature = "pg")]
145pub fn notify(kevy: &Store) {
146    let _ = kevy.publish(b"queue:notify", b"1");
147}
148
149/// enqueue a message for scheduled delivery at a future time
150#[allow(clippy::too_many_arguments)]
151#[cfg(feature = "pg")]
152pub async fn enqueue_scheduled(
153    pool: &BackendPool,
154    sender: &str,
155    recipient: &str,
156    domain: &str,
157    message_data: &[u8],
158    message_id: Option<&str>,
159    created_at: i64,
160    scheduled_at: i64,
161) -> Result<i64, sqlx::Error> {
162    let row: (i64,) = sqlx::query_as(
163        "INSERT INTO outbound_queue (sender, recipient, domain, message_data, status, next_retry, message_id, created_at, updated_at, is_forwarded)
164         VALUES ($1, $2, $3, $4, 'pending', $5, $6, $7, $7, false)
165         RETURNING id",
166    )
167    .bind(sender)
168    .bind(recipient)
169    .bind(domain)
170    .bind(message_data)
171    .bind(scheduled_at)
172    .bind(message_id)
173    .bind(created_at)
174    .fetch_one(pool)
175    .await?;
176    Ok(row.0)
177}
mailrs_outbound_queue/queue/mod.rs

mailrs_outbound_queue/queue/
mod.rs
