refactor(email): add telemetry logging across transport, service and worker

Modified app/mails/src/main.rs +28 −8 View app/mails/src/main.rs

@@ -1,27 +1,47 @@

11use std::time::Duration;

22

3use anyhow::Context;

34use email::runtime::{run_until_shutdown, WorkerOptions};

5use tracing::info;

46

57#[tokio::main]

68async fn main() -> anyhow::Result<()> {

7 tracing_subscriber::fmt()

8 .with_env_filter(

9 tracing_subscriber::EnvFilter::try_from_default_env()

10 .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")),

11 )

12 .init();

9 init_tracing();

1310

14 let cfg = config::AppConfig::load()?;

15 cfg.validate().map_err(|e| anyhow::anyhow!(e.to_string()))?;

11 let cfg = config::AppConfig::load().context("loading configuration")?;

12 cfg.validate()

13 .map_err(|e| anyhow::anyhow!(e.to_string()))?;

1614

1715 let redis_url = cfg.redis.url().to_string();

16 let redis_host = match redis_url.rsplit_once('@') {

17 Some((_, host)) => host.to_string(),

18 None => redis_url.splitn(2, "://").nth(1).unwrap_or("?").to_string(),

19 };

1820

21 info!(

22 smtp_enabled = cfg.email.enabled(),

23 queue_enabled = cfg.email_queue.enabled(),

24 redis = %redis_host,

25 shutdown_timeout_secs = 5,

26 "starting mails worker"

27 );

28

1929 let opts = WorkerOptions {

2030 template_dir: None,

2131 shutdown_timeout: Some(Duration::from_secs(5)),

2232 };

33

2334 run_until_shutdown(cfg.email.clone(), cfg.email_queue.clone(), redis_url, opts)

2435 .await

2536 .map_err(|e| anyhow::anyhow!(e.to_string()))?;

37

38 info!("mails worker exited cleanly");

2639 Ok(())

40}

41

42fn init_tracing() {

43 use tracing_subscriber::EnvFilter;

44 let filter = EnvFilter::try_from_default_env()

45 .unwrap_or_else(|_| EnvFilter::new("info"));

46 tracing_subscriber::fmt().with_env_filter(filter).init();

2747}

No newline at end of file

Modified lib/email/src/runtime.rs +17 −1 View lib/email/src/runtime.rs

@@ -1,7 +1,7 @@

11use std::path::PathBuf;

22use std::time::Duration;

33

4use tracing::{info, warn};

4use tracing::{debug, info, warn};

55

66use crate::error::EmailError;

77use crate::service::{EmailService, EmailServiceBuilder};

@@ -25,12 +25,26 @@

2525 opts: WorkerOptions,

2626) -> Result<RunningWorker, EmailError> {

2727 let _ = opts.template_dir;

28 let enabled = email_cfg.enabled();

29 info!(

30 smtp_enabled = enabled,

31 queue_enabled = queue_cfg.enabled(),

32 "building email service"

33 );

34

2835 let builder = EmailServiceBuilder::new(email_cfg)

2936 .with_queue(redis_url, queue_cfg)

3037 .register_default_templates()

3138 .await;

3239 let service = builder.build().await?;

3340 let subscribers = service.start_subscribers().await?;

41

42 info!(

43 smtp_enabled = service.enabled(),

44 subscribers_started = service.queue().is_some_and(|q| q.enabled()),

45 "email worker ready"

46 );

47

3448 Ok(RunningWorker {

3549 service,

3650 subscribers,

@@ -49,8 +63,10 @@

4963 return Ok(());

5064 }

5165 info!("email worker started; waiting for shutdown signal");

66 debug!("entering wait_for_shutdown");

5267 wait_for_shutdown().await;

5368 let timeout = Duration::from_secs(5);

69 debug!(timeout_secs = timeout.as_secs(), "calling subscribers.shutdown");

5470 worker.subscribers.shutdown(timeout).await?;

5571 info!("email worker shut down");

5672 Ok(())

Modified lib/email/src/service.rs +44 −4 View lib/email/src/service.rs

@@ -2,7 +2,7 @@

22

33use lettre::address::Address;

44use lettre::message::Mailbox;

5use tracing::{debug, info, warn};

5use tracing::{debug, info, trace, warn};

66use uuid::Uuid;

77

88use crate::error::EmailError;

@@ -74,6 +74,13 @@

7474 if recipient_count == 0 {

7575 return Err(EmailError::Missing("email has no recipients".into()));

7676 }

77 debug!(

78 request_id = %req.request_id,

79 recipients = recipient_count,

80 subject_len = req.subject.len(),

81 attachments = req.attachments.len(),

82 "service::send dispatch"

83 );

7784

7885 let message = build_message(&req, from, self.inner.max_attachment_bytes)?;

7986 let smtp_message_id = transport.send(message).await?;

@@ -105,6 +112,7 @@

105112 text: &str,

106113 html: &str,

107114 ) -> Result<(), EmailError> {

115 trace!(to, subject_len = subject.len(), "service::send_mail");

108116 let req = SendEmailRequest {

109117 request_id: Uuid::now_v7(),

110118 to: vec![EmailAddress::new(to)],

@@ -138,12 +146,14 @@

138146 let default_meta = req.metadata.clone();

139147 let total = req.recipients.len();

140148 let batch_id = req.request_id;

149 debug!(batch = %batch_id, total, "service::send_batch dispatch");

141150

142151 for (idx, recipient) in req.recipients.into_iter().enumerate() {

143152 if recipient.to.address.is_empty() {

144153 warn!(batch = %batch_id, index = idx, "skipping batch recipient without address");

145154 continue;

146155 }

156 debug!(batch = %batch_id, index = idx, address = %recipient.to.address, "batch recipient processing");

147157 let mut meta = default_meta.clone();

148158 meta.insert("batch_request_id".into(), batch_id.to_string());

149159 meta.insert("batch_index".into(), idx.to_string());

@@ -202,7 +212,20 @@

202212 &self,

203213 req: RenderAndSendEmailRequest,

204214 ) -> Result<EmailSent, EmailError> {

215 let recipient_count = req.to.len() + req.cc.len() + req.bcc.len();

216 debug!(

217 request_id = %req.request_id,

218 template = %req.template,

219 recipients = recipient_count,

220 "service::render_and_send dispatch"

221 );

205222 let html = self.inner.templates.render(&req.template, &req.vars).await?;

223 debug!(

224 request_id = %req.request_id,

225 template = %req.template,

226 bytes = html.len(),

227 "template rendered"

228 );

206229 let send_req = SendEmailRequest {

207230 request_id: req.request_id,

208231 to: req.to,

@@ -305,6 +328,7 @@

305328

306329 pub async fn build(self) -> Result<EmailService, EmailError> {

307330 if !self.email_config.enabled() {

331 info!("email disabled at config; transport=None, queue=None");

308332 return Ok(EmailService {

309333 inner: Arc::new(EmailServiceInner {

310334 config: Arc::new(self.email_config),

@@ -331,9 +355,16 @@

331355 None => Mailbox::new(None, parsed),

332356 };

333357

358 let smtp_host = self.email_config.smtp_host().to_string();

359 let smtp_port = self.email_config.smtp_port();

360 let tls_mode = match smtp_port {

361 465 => "implicit-tls",

362 587 => "starttls",

363 _ => "opportunistic",

364 };

334365 let transport = EmailTransportBuilder::new(

335 self.email_config.smtp_host().to_string(),

336 self.email_config.smtp_port(),

366 smtp_host.clone(),

367 smtp_port,

337368 self.email_config

338369 .smtp_username()

339370 .map(|s| s.to_string()),

@@ -345,6 +376,7 @@

345376 self.email_config.connection_timeout_ms(),

346377 ))

347378 .build()?;

379 info!(host = %smtp_host, port = smtp_port, tls = tls_mode, "smtp transport ready");

348380

349381 let queue = match (&self.redis_url, &self.queue_config) {

350382 (Some(url), Some(qcfg)) if qcfg.enabled() => {

@@ -358,9 +390,17 @@

358390 },

359391 )

360392 .await?;

393 info!(

394 prefix = %qcfg.channel_prefix(),

395 max_retries = qcfg.max_retries(),

396 "redis queue connected"

397 );

361398 Some(queue)

362399 }

363 _ => None,

400 _ => {

401 info!("redis queue disabled; pubsub subscribers will not start");

402 None

403 }

364404 };

365405

366406 let max_attachment_bytes = self.email_config.max_attachment_bytes();

Modified lib/email/src/subscriber.rs +43 −1 View lib/email/src/subscriber.rs

@@ -1,5 +1,7 @@

11use std::time::Duration;

22

3use tracing::{debug, info};

4

35use crate::error::EmailError;

46use crate::events::{

57 now_ms, EmailFailed, RenderAndSendEmailRequest, SendBatchEmailRequest, SendEmailRequest,

@@ -55,12 +57,20 @@

5557) -> Result<EmailSubscribers, EmailError> {

5658 let queue = match service.queue() {

5759 Some(q) if q.enabled() => q,

58 _ => return Ok(EmailSubscribers::empty()),

60 _ => {

61 info!("email queue disabled; skipping pubsub subscribers");

62 return Ok(EmailSubscribers::empty());

63 }

5964 };

6065 let group = service

6166 .config()

6267 .smtp_host()

6368 .to_string();

69 info!(

70 group = %group,

71 prefix = %queue.config().channel_prefix,

72 "subscribing to email channels"

73 );

6474

6575 let single = queue

6676 .subscribe(CHANNEL_SEND, &group, {

@@ -72,8 +82,19 @@

7282 req.to.len() + req.cc.len() + req.bcc.len();

7383 let request_id = req.request_id;

7484 let metadata = req.metadata.clone();

85 debug!(

86 request_id = %request_id,

87 recipients = recipient_count,

88 channel = CHANNEL_SEND,

89 "handler invoked"

90 );

7591 match svc.send(req).await {

7692 Ok(sent) => {

93 debug!(

94 request_id = %sent.request_id,

95 smtp_message_id = %sent.smtp_message_id,

96 "smtp send ok"

97 );

7798 if let Err(err) = svc.publish_sent(&sent).await {

7899 tracing::warn!(error = %err, "publish EmailSent failed");

79100 }

@@ -101,6 +122,7 @@

101122 }

102123 })

103124 .await?;

125 info!(channel = CHANNEL_SEND, "subscribed channel");

104126

105127 let batch = queue

106128 .subscribe(CHANNEL_BATCH, &group, {

@@ -110,6 +132,12 @@

110132 async move {

111133 let total = req.recipients.len();

112134 let request_id = req.request_id;

135 debug!(

136 request_id = %request_id,

137 total,

138 channel = CHANNEL_BATCH,

139 "batch handler invoked"

140 );

113141 match svc.send_batch(req).await {

114142 Ok(_) => {

115143 tracing::debug!(request_id = %request_id, total, "batch completed");

@@ -137,6 +165,7 @@

137165 }

138166 })

139167 .await?;

168 info!(channel = CHANNEL_BATCH, "subscribed channel");

140169

141170 let render = queue

142171 .subscribe(CHANNEL_RENDER, &group, {

@@ -148,8 +177,20 @@

148177 let recipient_count =

149178 req.to.len() + req.cc.len() + req.bcc.len();

150179 let metadata = req.metadata.clone();

180 debug!(

181 request_id = %request_id,

182 template = %req.template,

183 recipients = recipient_count,

184 channel = CHANNEL_RENDER,

185 "render handler invoked"

186 );

151187 match svc.render_and_send(req).await {

152188 Ok(sent) => {

189 debug!(

190 request_id = %sent.request_id,

191 smtp_message_id = %sent.smtp_message_id,

192 "render+send ok"

193 );

153194 if let Err(err) = svc.publish_sent(&sent).await {

154195 tracing::warn!(error = %err, "publish EmailSent failed");

155196 }

@@ -177,6 +218,7 @@

177218 }

178219 })

179220 .await?;

221 info!(channel = CHANNEL_RENDER, "subscribed channel");

180222

181223 Ok(EmailSubscribers {

182224 single: Some(single),

Modified lib/email/src/template.rs +12 −2 View lib/email/src/template.rs

@@ -3,6 +3,7 @@

33

44use maud::Markup;

55use tokio::sync::RwLock;

6use tracing::debug;

67

78use crate::error::EmailError;

89

@@ -23,16 +24,25 @@

2324 F: Fn(&serde_json::Value) -> Markup + Send + Sync + 'static,

2425 {

2526 let mut guard = self.inner.write().await;

27 let replacing = guard.contains_key(name);

2628 guard.insert(name.to_string(), Arc::new(f));

29 debug!(template = %name, replacing, "template registered");

2730 }

2831

2932 pub async fn render(&self, name: &str, vars: &serde_json::Value) -> Result<String, EmailError> {

3033 let guard = self.inner.read().await;

3134 let Some(f) = guard.get(name) else {

35 debug!(template = %name, "template not found");

3236 return Err(EmailError::TemplateNotFound(name.to_string()));

3337 };

34 let markup = f(vars);

35 Ok(markup.into_string())

38 let markup = f(vars).into_string();

39 debug!(

40 template = %name,

41 bytes = markup.len(),

42 vars_bytes = vars.to_string().len(),

43 "template rendered"

44 );

45 Ok(markup)

3646 }

3747

3848 pub async fn contains(&self, name: &str) -> bool {

Modified lib/email/src/transport.rs +27 −2 View lib/email/src/transport.rs

@@ -1,4 +1,4 @@

1use std::time::Duration;

1use std::time::{Duration, Instant};

22

33use base64::Engine;

44use lettre::address::Address;

@@ -10,6 +10,7 @@

1010use lettre::transport::smtp::AsyncSmtpTransport;

1111use lettre::AsyncTransport;

1212use lettre::Tokio1Executor;

13use tracing::debug;

1314

1415use crate::error::EmailError;

1516use crate::events::{EmailAddress, SendEmailRequest};

@@ -20,8 +21,16 @@

2021

2122impl EmailTransport {

2223 pub async fn send(&self, msg: Message) -> Result<String, EmailError> {

24 let started = Instant::now();

2325 let response = self.inner.send(msg).await?;

24 Ok(response.message().next().unwrap_or_default().to_string())

26 let id = response.message().next().unwrap_or_default().to_string();

27 let id_short: String = id.chars().take(16).collect();

28 debug!(

29 elapsed_ms = started.elapsed().as_millis() as u64,

30 message_id = %id_short,

31 "smtp response received"

32 );

33 Ok(id)

2534 }

2635}

2736

@@ -82,6 +91,13 @@

8291 .credentials(Credentials::new(u.clone(), p.clone()))

8392 .authentication(vec![Mechanism::Plain]);

8493 }

94 debug!(

95 host = %self.host,

96 port = self.port,

97 has_credentials = self.username.is_some() && self.password.is_some(),

98 timeout_ms = self.connection_timeout.map(|d| d.as_millis() as u64).unwrap_or(0),

99 "smtp transport builder configured"

100 );

85101 Ok(EmailTransport {

86102 inner: builder.build(),

87103 })

@@ -104,6 +120,15 @@

104120) -> Result<Message, EmailError> {

105121 let has_text = req.text.as_deref().is_some_and(|s| !s.is_empty());

106122 let has_html = req.html.as_deref().is_some_and(|s| !s.is_empty());

123 let recipient_count = req.to.len() + req.cc.len() + req.bcc.len();

124 debug!(

125 subject_len = req.subject.len(),

126 recipients = recipient_count,

127 attachments = req.attachments.len(),

128 has_text,

129 has_html,

130 "building email message"

131 );

107132 if !has_text && !has_html {

108133 return Err(EmailError::Missing(

109134 "email body requires text or html".into(),