refactor(email): refresh crate across lib, service, transport and metrics
zhenyi committed
Modified lib/email/lib.rs lib/email/lib.rs +1 −0 View lib/email/lib.rs
@@ -1,6 +1,7 @@
11pub mod config;
22pub mod error;
33pub mod events;
4pub mod metrics;
45pub mod mime;
56pub mod runtime;
67pub mod service;
Added lib/email/metrics.rs lib/email/metrics.rs +131 −0 View lib/email/metrics.rs
@@ -0,0 +1,131 @@
1use std::sync::Once;
2
3use metrics::{Unit, counter, describe_counter, describe_histogram, histogram};
4
5use crate::error::EmailError;
6use crate::events::EmailSent;
7
8pub const OUTCOME_SUCCESS: &str = "success";
9
10pub const KIND_SENT: &str = "sent";
11pub const KIND_FAILED: &str = "failed";
12
13
14static DESCRIBE_ONCE: Once = Once::new();
15
16pub fn ensure_described() {
17 DESCRIBE_ONCE.call_once(describe_all);
18}
19
20fn describe_all() {
21 describe_counter!(
22 "email_send_total",
23 "Total email sends partitioned by outcome"
24 );
25 describe_histogram!(
26 "email_send_duration_seconds",
27 Unit::Seconds,
28 "Wall-clock duration of a single SMTP send"
29 );
30 describe_counter!(
31 "email_batch_sent_recipients_total",
32 "Total recipients successfully delivered across batch sends"
33 );
34 describe_counter!(
35 "email_render_send_total",
36 "Total render-and-send requests partitioned by outcome"
37 );
38 describe_counter!(
39 "email_publish_total",
40 "Total email result events published to the queue partitioned by kind and outcome"
41 );
42 describe_counter!(
43 "email_transport_send_total",
44 "Total SMTP transport sends partitioned by outcome"
45 );
46}
47
48
49fn outcome_for<T>(result: &Result<T, EmailError>) -> &'static str {
50 match result {
51 Ok(_) => OUTCOME_SUCCESS,
52 Err(e) => match e {
53 EmailError::Queue(_) => "queue",
54 EmailError::Serde(_) => "serde",
55 EmailError::Lettre(_) | EmailError::LettreOther(_) | EmailError::Address(_) => "smtp",
56 EmailError::Template(_)
57 | EmailError::TemplateRegister(_)
58 | EmailError::TemplateNotFound(_) => "template",
59 EmailError::Config(_) => "config",
60 EmailError::Base64(_) => "input",
61 EmailError::AttachmentTooLarge(_, _) | EmailError::Missing(_) => "input",
62 EmailError::Io(_) => "io",
63 },
64 }
65}
66
67#[must_use = "timer is dropped silently if not bound"]
68pub struct SendTimer {
69 start: std::time::Instant,
70}
71
72impl SendTimer {
73 pub fn start() -> Self {
74 ensure_described();
75 Self {
76 start: std::time::Instant::now(),
77 }
78 }
79
80 pub fn observe(self) {
81 histogram!("email_send_duration_seconds").record(self.start.elapsed().as_secs_f64());
82 }
83}
84
85impl Drop for SendTimer {
86 fn drop(&mut self) {
87 histogram!("email_send_duration_seconds").record(self.start.elapsed().as_secs_f64());
88 }
89}
90
91pub fn record_send(result: &Result<EmailSent, EmailError>) {
92 ensure_described();
93 counter!(
94 "email_send_total",
95 "outcome" => outcome_for(result)
96 )
97 .increment(1);
98}
99
100pub fn record_batch_sent(n: usize) {
101 ensure_described();
102 counter!("email_batch_sent_recipients_total").increment(n as u64);
103}
104
105pub fn record_render_send(result: &Result<EmailSent, EmailError>) {
106 ensure_described();
107 counter!(
108 "email_render_send_total",
109 "outcome" => outcome_for(result)
110 )
111 .increment(1);
112}
113
114pub fn record_publish(kind: &'static str, result: &Result<(), EmailError>) {
115 ensure_described();
116 counter!(
117 "email_publish_total",
118 "kind" => kind,
119 "outcome" => outcome_for(result)
120 )
121 .increment(1);
122}
123
124pub fn record_transport_send(result: &Result<String, EmailError>) {
125 ensure_described();
126 counter!(
127 "email_transport_send_total",
128 "outcome" => outcome_for(result)
129 )
130 .increment(1);
131}
No newline at end of file
Modified lib/email/service.rs lib/email/service.rs +74 −51 View lib/email/service.rs
@@ -10,6 +10,7 @@
1010 EmailFailed, EmailSent, RenderAndSendEmailRequest, SendBatchEmailRequest, SendEmailRequest,
1111 now_ms,
1212};
13use crate::metrics;
1314use crate::mime::{MimeBuildOptions, build_message};
1415use crate::template::TemplateRegistry;
1516use crate::transport::EmailTransport;
@@ -86,23 +87,29 @@
8687 }
8788
8889 pub async fn send(&self, mut req: SendEmailRequest) -> Result<EmailSent, EmailError> {
89 if req.to.is_empty() {
90 return Err(EmailError::Missing("to recipients".to_string()));
90 let _timer = metrics::SendTimer::start();
91 let result: Result<EmailSent, EmailError> = async {
92 if req.to.is_empty() {
93 return Err(EmailError::Missing("to recipients".to_string()));
94 }
95 let opts = MimeBuildOptions {
96 max_attachment_bytes: self.inner.cfg.max_attachment_bytes,
97 };
98 let message = build_message(&req, self.inner.transport.from_mailbox(), &opts)?;
99 let message_id = self.inner.transport.send(message).await?;
100 let total_recipients = req.to.len() + req.cc.len() + req.bcc.len();
101 let sent = EmailSent {
102 request_id: req.request_id.take(),
103 message_id,
104 recipient_count: total_recipients,
105 metadata: req.metadata.clone(),
106 sent_at_ms: now_ms(),
107 };
108 Ok(sent)
91109 }
92 let opts = MimeBuildOptions {
93 max_attachment_bytes: self.inner.cfg.max_attachment_bytes,
94 };
95 let message = build_message(&req, self.inner.transport.from_mailbox(), &opts)?;
96 let message_id = self.inner.transport.send(message).await?;
97 let total_recipients = req.to.len() + req.cc.len() + req.bcc.len();
98 let sent = EmailSent {
99 request_id: req.request_id.take(),
100 message_id,
101 recipient_count: total_recipients,
102 metadata: req.metadata.clone(),
103 sent_at_ms: now_ms(),
104 };
105 Ok(sent)
110 .await;
111 metrics::record_send(&result);
112 result
106113 }
107114
108115 pub async fn send_batch(
@@ -143,6 +150,7 @@
143150 }
144151 }
145152 }
153 metrics::record_batch_sent(out.len());
146154 Ok(out)
147155 }
148156
@@ -150,48 +158,63 @@
150158 &self,
151159 req: RenderAndSendEmailRequest,
152160 ) -> Result<EmailSent, EmailError> {
153 let locale_str = req.locale.clone();
154 let locale_borrow = locale_str.as_deref();
155 let templates = self.inner.templates.read().await;
156 let html = templates
157 .render_html(&req.template_id, locale_borrow, &req.vars)
158 .ok();
159 let text = templates
160 .render_text(&req.template_id, locale_borrow, &req.vars)
161 .ok();
162 drop(templates);
163 if html.is_none() && text.is_none() {
164 return Err(EmailError::TemplateNotFound(req.template_id));
161 let result: Result<EmailSent, EmailError> = async {
162 let locale_str = req.locale.clone();
163 let locale_borrow = locale_str.as_deref();
164 let templates = self.inner.templates.read().await;
165 let html = templates
166 .render_html(&req.template_id, locale_borrow, &req.vars)
167 .ok();
168 let text = templates
169 .render_text(&req.template_id, locale_borrow, &req.vars)
170 .ok();
171 drop(templates);
172 if html.is_none() && text.is_none() {
173 return Err(EmailError::TemplateNotFound(req.template_id));
174 }
175 let single = SendEmailRequest {
176 to: vec![req.to],
177 cc: req.cc,
178 bcc: req.bcc,
179 reply_to: req.reply_to,
180 subject: req.subject,
181 text,
182 html,
183 attachments: req.attachments,
184 headers: vec![],
185 locale: locale_str,
186 request_id: req.request_id,
187 metadata: req.metadata,
188 };
189 self.send(single).await
165190 }
166 let single = SendEmailRequest {
167 to: vec![req.to],
168 cc: req.cc,
169 bcc: req.bcc,
170 reply_to: req.reply_to,
171 subject: req.subject,
172 text,
173 html,
174 attachments: req.attachments,
175 headers: vec![],
176 locale: locale_str,
177 request_id: req.request_id,
178 metadata: req.metadata,
179 };
180 self.send(single).await
191 .await;
192 metrics::record_render_send(&result);
193 result
181194 }
182195
183196 pub async fn publish_sent(&self, sent: EmailSent) -> Result<(), EmailError> {
184 self.inner.queue.typed::<EmailSent>().publish(&sent).await?;
185 Ok(())
197 let result: Result<(), EmailError> = async {
198 self.inner.queue.typed::<EmailSent>().publish(&sent).await?;
199 Ok(())
200 }
201 .await;
202 metrics::record_publish(metrics::KIND_SENT, &result);
203 result
186204 }
187205
188206 pub async fn publish_failed(&self, failed: EmailFailed) -> Result<(), EmailError> {
189 self.inner
190 .queue
191 .typed::<EmailFailed>()
192 .publish(&failed)
193 .await?;
194 Ok(())
207 let result: Result<(), EmailError> = async {
208 self.inner
209 .queue
210 .typed::<EmailFailed>()
211 .publish(&failed)
212 .await?;
213 Ok(())
214 }
215 .await;
216 metrics::record_publish(metrics::KIND_FAILED, &result);
217 result
195218 }
196219
197220 pub fn start_subscribers(&self) -> Result<EmailSubscribers, EmailError> {
Modified lib/email/transport.rs lib/email/transport.rs +9 −3 View lib/email/transport.rs
@@ -10,6 +10,7 @@
1010
1111use crate::config::AppEmailSettings;
1212use crate::error::EmailError;
13use crate::metrics;
1314
1415pub struct EmailTransport {
1516 inner: AsyncSmtpTransport<Tokio1Executor>,
@@ -55,9 +56,14 @@
5556 }
5657
5758 pub async fn send(&self, message: Message) -> Result<String, EmailError> {
58 let response = self.inner.send(message).await?;
59 let text = response.message().collect::<Vec<_>>().join(" ");
60 Ok(text)
59 let result: Result<String, EmailError> = async {
60 let response = self.inner.send(message).await?;
61 let text = response.message().collect::<Vec<_>>().join(" ");
62 Ok(text)
63 }
64 .await;
65 metrics::record_transport_send(&result);
66 result
6167 }
6268
6369 pub fn from_mailbox(&self) -> Mailbox {