mailrs_outbound_queue/queue/
admin.rs

1//! Everything the admin surface asks of the queue: counts, listings,
2//! cancel, retry, reschedule, and the suppression list.
3
4use super::*;
5#[cfg(feature = "pg")]
6use crate::BackendPool;
7
8/// get queue statistics
9#[cfg(feature = "pg")]
10pub async fn queue_stats(pool: &BackendPool) -> Result<Vec<(String, i64)>, sqlx::Error> {
11    let rows: Vec<(String, i64)> =
12        sqlx::query_as("SELECT status, COUNT(*) FROM outbound_queue GROUP BY status")
13            .fetch_all(pool)
14            .await?;
15    Ok(rows)
16}
17
18/// Count of rows in `status = 'pending'` (any `next_retry`). Used by
19/// the delivery worker to publish a `mailrs_outbound_queue_depth` gauge
20/// per poll tick. O(rows) but cheap with the `status` index.
21#[cfg(feature = "pg")]
22pub async fn count_pending(pool: &BackendPool) -> Result<i64, sqlx::Error> {
23    let (n,): (i64,) =
24        sqlx::query_as("SELECT count(*) FROM outbound_queue WHERE status = 'pending'")
25            .fetch_one(pool)
26            .await?;
27    Ok(n)
28}
29
30/// Count of rows in `status = 'inflight'` (currently being delivered
31/// by a worker). Used alongside `count_pending` so dashboards can show
32/// both "queued" and "in-flight" depth.
33#[cfg(feature = "pg")]
34pub async fn count_inflight(pool: &BackendPool) -> Result<i64, sqlx::Error> {
35    let (n,): (i64,) =
36        sqlx::query_as("SELECT count(*) FROM outbound_queue WHERE status = 'inflight'")
37            .fetch_one(pool)
38            .await?;
39    Ok(n)
40}
41
42/// get a specific queued message by id
43#[cfg(feature = "pg")]
44pub async fn get_message(
45    pool: &BackendPool,
46    id: i64,
47) -> Result<Option<QueuedMessage>, sqlx::Error> {
48    #[allow(clippy::type_complexity)]
49    let row: Option<(i64, String, String, String, Vec<u8>, String, i32, i32, i64, Option<String>, Option<String>, i64, i64, bool)> = sqlx::query_as(
50        "SELECT id, sender, recipient, domain, message_data, status, attempts, max_attempts, next_retry, last_error, message_id, created_at, updated_at, is_forwarded
51         FROM outbound_queue WHERE id = $1",
52    )
53    .bind(id)
54    .fetch_optional(pool)
55    .await?;
56
57    Ok(row.map(|r| QueuedMessage {
58        id: r.0,
59        sender: r.1,
60        recipient: r.2,
61        domain: r.3,
62        message_data: r.4,
63        status: QueueStatus::parse(&r.5).unwrap_or(QueueStatus::Pending),
64        attempts: r.6 as u32,
65        max_attempts: r.7 as u32,
66        next_retry: r.8,
67        last_error: r.9,
68        message_id: r.10,
69        created_at: r.11,
70        updated_at: r.12,
71        is_forwarded: r.13,
72    }))
73}
74
75/// list recent queue entries for admin UI
76#[cfg(feature = "pg")]
77pub async fn list_recent(
78    pool: &BackendPool,
79    limit: i32,
80) -> Result<Vec<QueuedMessage>, sqlx::Error> {
81    #[allow(clippy::type_complexity)]
82    let rows: Vec<(i64, String, String, String, Vec<u8>, String, i32, i32, i64, Option<String>, Option<String>, i64, i64, bool)> = sqlx::query_as(
83        "SELECT id, sender, recipient, domain, message_data, status, attempts, max_attempts, next_retry, last_error, message_id, created_at, updated_at, is_forwarded
84         FROM outbound_queue
85         ORDER BY created_at DESC
86         LIMIT $1",
87    )
88    .bind(limit)
89    .fetch_all(pool)
90    .await?;
91
92    Ok(rows
93        .into_iter()
94        .map(|r| QueuedMessage {
95            id: r.0,
96            sender: r.1,
97            recipient: r.2,
98            domain: r.3,
99            message_data: r.4,
100            status: QueueStatus::parse(&r.5).unwrap_or(QueueStatus::Pending),
101            attempts: r.6 as u32,
102            max_attempts: r.7 as u32,
103            next_retry: r.8,
104            last_error: r.9,
105            message_id: r.10,
106            created_at: r.11,
107            updated_at: r.12,
108            is_forwarded: r.13,
109        })
110        .collect())
111}
112
113/// cancel a pending outbound message (undo send)
114#[cfg(feature = "pg")]
115pub async fn cancel_pending(pool: &BackendPool, id: i64) -> Result<bool, sqlx::Error> {
116    let result = sqlx::query("DELETE FROM outbound_queue WHERE id = $1 AND status = 'pending'")
117        .bind(id)
118        .execute(pool)
119        .await?;
120    Ok(result.rows_affected() > 0)
121}
122
123/// cancel a pending outbound message by message_id (undo send)
124#[cfg(feature = "pg")]
125pub async fn cancel_pending_by_message_id(
126    pool: &BackendPool,
127    message_id: &str,
128    sender: &str,
129) -> Result<bool, sqlx::Error> {
130    let result = sqlx::query(
131        "DELETE FROM outbound_queue WHERE message_id = $1 AND status = 'pending' AND sender = $2",
132    )
133    .bind(message_id)
134    .bind(sender)
135    .execute(pool)
136    .await?;
137    Ok(result.rows_affected() > 0)
138}
139
140/// reset a bounced/failed message back to pending for retry
141#[cfg(feature = "pg")]
142pub async fn retry_message(pool: &BackendPool, id: i64, now: i64) -> Result<bool, sqlx::Error> {
143    let result = sqlx::query(
144        "UPDATE outbound_queue SET status = 'pending', next_retry = $1, updated_at = $1 WHERE id = $2 AND status IN ('bounced', 'failed')",
145    )
146    .bind(now)
147    .bind(id)
148    .execute(pool)
149    .await?;
150    Ok(result.rows_affected() > 0)
151}
152
153/// move a still-pending message's delivery time to `scheduled_at`
154#[cfg(feature = "pg")]
155pub async fn reschedule_pending(
156    pool: &BackendPool,
157    id: i64,
158    scheduled_at: i64,
159    now: i64,
160) -> Result<bool, sqlx::Error> {
161    let result = sqlx::query(
162        "UPDATE outbound_queue SET next_retry = $1, updated_at = $2 WHERE id = $3 AND status = 'pending'",
163    )
164    .bind(scheduled_at)
165    .bind(now)
166    .bind(id)
167    .execute(pool)
168    .await?;
169    Ok(result.rows_affected() > 0)
170}
171
172pub use suppression::is_hard_bounce;
173#[cfg(feature = "pg")]
174pub use suppression::{add_suppression, is_suppressed, list_suppressions, remove_suppression};
175
176#[cfg(test)]
177mod tests {
178    use super::*;
179
180    #[test]
181    fn queue_status_roundtrip() {
182        let variants = [
183            QueueStatus::Pending,
184            QueueStatus::InFlight,
185            QueueStatus::Delivered,
186            QueueStatus::Failed,
187            QueueStatus::Bounced,
188        ];
189        for v in &variants {
190            let s = v.as_str();
191            let parsed = QueueStatus::parse(s).unwrap();
192            assert_eq!(&parsed, v, "roundtrip failed for {s}");
193        }
194    }
195
196    #[test]
197    fn queue_status_parse_unknown() {
198        assert_eq!(QueueStatus::parse("unknown"), None);
199        assert_eq!(QueueStatus::parse(""), None);
200        assert_eq!(QueueStatus::parse("PENDING"), None);
201    }
202
203    #[test]
204    fn queue_status_as_str_values() {
205        assert_eq!(QueueStatus::Pending.as_str(), "pending");
206        assert_eq!(QueueStatus::InFlight.as_str(), "inflight");
207        assert_eq!(QueueStatus::Delivered.as_str(), "delivered");
208        assert_eq!(QueueStatus::Failed.as_str(), "failed");
209        assert_eq!(QueueStatus::Bounced.as_str(), "bounced");
210    }
211
212    #[test]
213    fn queue_status_parse_case_sensitive() {
214        // parse is case-sensitive — uppercase variants are not valid
215        assert_eq!(QueueStatus::parse("Pending"), None);
216        assert_eq!(QueueStatus::parse("InFlight"), None);
217        assert_eq!(QueueStatus::parse("DELIVERED"), None);
218        assert_eq!(QueueStatus::parse("Failed"), None);
219        assert_eq!(QueueStatus::parse("Bounced"), None);
220    }
221
222    #[test]
223    fn queue_status_parse_whitespace_rejected() {
224        assert_eq!(QueueStatus::parse(" pending"), None);
225        assert_eq!(QueueStatus::parse("pending "), None);
226        assert_eq!(QueueStatus::parse("  "), None);
227    }
228
229    #[test]
230    fn queue_status_eq() {
231        assert_eq!(QueueStatus::Pending, QueueStatus::Pending);
232        assert_ne!(QueueStatus::Pending, QueueStatus::Delivered);
233        assert_ne!(QueueStatus::Failed, QueueStatus::Bounced);
234    }
235
236    #[test]
237    fn queue_status_clone() {
238        let s = QueueStatus::InFlight;
239        let c = s.clone();
240        assert_eq!(s, c);
241    }
242
243    #[test]
244    fn queued_message_clone_preserves_fields() {
245        let msg = QueuedMessage {
246            id: 42,
247            sender: "s@example.com".into(),
248            recipient: "r@remote.com".into(),
249            domain: "remote.com".into(),
250            message_data: vec![1, 2, 3],
251            status: QueueStatus::Pending,
252            attempts: 3,
253            max_attempts: 8,
254            next_retry: 1_700_000_000,
255            last_error: Some("temporary failure".into()),
256            message_id: Some("msg-id-123".into()),
257            created_at: 1_699_000_000,
258            updated_at: 1_699_500_000,
259            is_forwarded: true,
260        };
261        let cloned = msg.clone();
262        assert_eq!(cloned.id, 42);
263        assert_eq!(cloned.sender, "s@example.com");
264        assert_eq!(cloned.recipient, "r@remote.com");
265        assert_eq!(cloned.domain, "remote.com");
266        assert_eq!(cloned.message_data, vec![1, 2, 3]);
267        assert_eq!(cloned.attempts, 3);
268        assert_eq!(cloned.max_attempts, 8);
269        assert_eq!(cloned.next_retry, 1_700_000_000);
270        assert_eq!(cloned.last_error, Some("temporary failure".into()));
271        assert_eq!(cloned.message_id, Some("msg-id-123".into()));
272        assert!(cloned.is_forwarded);
273    }
274
275    #[test]
276    fn queued_message_no_last_error() {
277        let msg = QueuedMessage {
278            id: 1,
279            sender: "s@example.com".into(),
280            recipient: "r@remote.com".into(),
281            domain: "remote.com".into(),
282            message_data: vec![],
283            status: QueueStatus::Pending,
284            attempts: 0,
285            max_attempts: 8,
286            next_retry: 0,
287            last_error: None,
288            message_id: None,
289            created_at: 0,
290            updated_at: 0,
291            is_forwarded: false,
292        };
293        assert!(msg.last_error.is_none());
294        assert!(msg.message_id.is_none());
295    }
296}
mailrs_outbound_queue/queue/admin.rs

mailrs_outbound_queue/queue/
admin.rs
