forge
ExploreLog in

Tahinli/forge

public
0 watching
★ 0
3.2 KiB
1//! Outgoing mail. Callers queue a message with [`enqueue`]; the worker in [`worker`] sends it
2//! through the [`Mailer`] held in `AppState`, retrying on the schedule in `forge_core::mail`.
3//! Nothing sends mail inside a request.
4
5pub mod account;
6pub mod log;
7pub mod smtp;
8pub mod templates;
9pub mod worker;
10
11use std::future::Future;
12use std::pin::Pin;
13
14use forge_core::mail::{MailError, OutgoingMail};
15
16use crate::state::AppState;
17
18pub use log::LogMailer;
19pub use smtp::SmtpMailer;
20
21/// The future a [`Mailer::send`] call returns.
22pub type SendFuture<'a> = Pin<Box<dyn Future<Output = Result<(), MailError>> + Send + 'a>>;
23
24/// A transport that delivers one message. Object-safe, so `AppState` holds it as `Arc<dyn Mailer>`.
25pub trait Mailer: Send + Sync {
26 /// Delivers `mail`. Must not return until the transport has finished or timed out.
27 fn send<'a>(&'a self, mail: &'a OutgoingMail) -> SendFuture<'a>;
28}
29
30/// Queues one message, keyed by `idem_key`. Returns `false` when that key was already queued, so
31/// the caller can retry without sending twice. The worker is woken to pick it up.
32pub async fn enqueue(
33 state: &AppState,
34 idem_key: &str,
35 to: &str,
36 subject: &str,
37 body: &str,
38) -> forge_core::Result<bool> {
39 let mail = OutgoingMail {
40 to: to.to_owned(),
41 subject: subject.to_owned(),
42 body: body.to_owned(),
43 message_id: None,
44 in_reply_to: None,
45 };
46 enqueue_message(state, idem_key, &mail).await
47}
48
49/// Queues a message that may carry threading headers. Same contract as [`enqueue`].
50pub async fn enqueue_message(
51 state: &AppState,
52 idem_key: &str,
53 mail: &OutgoingMail,
54) -> forge_core::Result<bool> {
55 let queued = state.store.enqueue_mail(idem_key, mail).await?;
56 if queued {
57 state.mail_waker.notify_one();
58 }
59 Ok(queued)
60}
61
62/// A mailer for tests. It records every message it accepts. [`FakeMailer::fail_next`] makes the
63/// next sends fail, to exercise the retry path.
64#[cfg(test)]
65#[derive(Default)]
66pub struct FakeMailer {
67 sent: std::sync::Mutex<Vec<OutgoingMail>>,
68 failures: std::sync::Mutex<std::collections::VecDeque<MailError>>,
69}
70
71#[cfg(test)]
72impl FakeMailer {
73 /// Messages accepted so far, oldest first.
74 pub fn sent(&self) -> Vec<OutgoingMail> {
75 self.sent
76 .lock()
77 .unwrap_or_else(std::sync::PoisonError::into_inner)
78 .clone()
79 }
80
81 /// Queues one failure for the next send.
82 pub fn fail_next(&self, error: MailError) {
83 self.failures
84 .lock()
85 .unwrap_or_else(std::sync::PoisonError::into_inner)
86 .push_back(error);
87 }
88}
89
90#[cfg(test)]
91impl Mailer for FakeMailer {
92 fn send<'a>(&'a self, mail: &'a OutgoingMail) -> SendFuture<'a> {
93 Box::pin(async move {
94 let failure = self
95 .failures
96 .lock()
97 .unwrap_or_else(std::sync::PoisonError::into_inner)
98 .pop_front();
99 if let Some(error) = failure {
100 return Err(error);
101 }
102 self.sent
103 .lock()
104 .unwrap_or_else(std::sync::PoisonError::into_inner)
105 .push(mail.clone());
106 Ok(())
107 })
108 }
109}