Skip to main content

max / makenotwork

18.8 KB · 537 lines History Blame Raw
1 //! Email service for sending transactional emails via Postmark.
2 //!
3 //! - `templates`, email composition methods (one per email type)
4 //! - `tokens`, HMAC-signed URL generation/verification for email actions
5
6 mod templates;
7 mod tokens;
8 pub use tokens::*;
9
10 use std::sync::Arc;
11
12 use crate::error::{AppError, Result};
13
14 /// Format an optional display name as a greeting suffix: " Alice" or "".
15 fn greeting(name: Option<&str>) -> String {
16 name.map(|n| format!(" {n}")).unwrap_or_default()
17 }
18
19 /// A recipient list proven to be within [`BROADCAST_MAX_RECIPIENTS`](crate::constants::BROADCAST_MAX_RECIPIENTS).
20 ///
21 /// Broadcasts must not materialize an unbounded recipient set in memory and
22 /// fan out unthrottled email. The only way to obtain this type is
23 /// [`BoundedRecipients::new`], which enforces the cap at construction, so a new
24 /// broadcast site physically cannot skip the check. This replaces the
25 /// copy-pasted `if count > BROADCAST_MAX_RECIPIENTS` guards that had drifted
26 /// between the public and internal broadcast handlers (the cap constant now
27 /// lives only in this constructor; a grep guard below enforces that). On
28 /// overflow `new` returns the actual recipient count so the caller can build a
29 /// user-facing message and roll back any rate-limit slot it already consumed.
30 #[derive(Debug)]
31 pub struct BoundedRecipients<T>(Vec<T>);
32
33 impl<T> BoundedRecipients<T> {
34 /// Construct from a raw recipient list, enforcing the broadcast cap.
35 /// Returns `Err(count)` with the actual recipient count on overflow.
36 pub fn new(recipients: Vec<T>) -> std::result::Result<Self, usize> {
37 let n = recipients.len();
38 if n > crate::constants::BROADCAST_MAX_RECIPIENTS {
39 Err(n)
40 } else {
41 Ok(Self(recipients))
42 }
43 }
44
45 /// Number of recipients (always `<= BROADCAST_MAX_RECIPIENTS`).
46 pub fn len(&self) -> usize {
47 self.0.len()
48 }
49
50 /// Whether the recipient list is empty.
51 pub fn is_empty(&self) -> bool {
52 self.0.is_empty()
53 }
54
55 /// Consume into the inner recipient vector for fan-out iteration.
56 pub fn into_inner(self) -> Vec<T> {
57 self.0
58 }
59 }
60
61 /// Email service configuration
62 #[derive(Clone)]
63 pub struct EmailConfig {
64 /// Postmark API token (optional, logs if not set)
65 pub postmark_token: Option<String>,
66 /// Default from address
67 pub from_address: String,
68 /// Default from name
69 pub from_name: String,
70 }
71
72 impl std::fmt::Debug for EmailConfig {
73 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
74 f.debug_struct("EmailConfig")
75 .field(
76 "postmark_token",
77 &self.postmark_token.as_ref().map(|_| "[REDACTED]"),
78 )
79 .field("from_address", &self.from_address)
80 .field("from_name", &self.from_name)
81 .finish()
82 }
83 }
84
85 impl EmailConfig {
86 /// Load email configuration from environment
87 pub fn from_env() -> Self {
88 EmailConfig {
89 postmark_token: std::env::var("POSTMARK_TOKEN").ok(),
90 from_address: std::env::var("EMAIL_FROM_ADDRESS")
91 .unwrap_or_else(|_| "noreply@makenot.work".to_string()),
92 from_name: std::env::var("EMAIL_FROM_NAME")
93 .unwrap_or_else(|_| "Makenotwork".to_string()),
94 }
95 }
96 }
97
98 /// Core email sending abstraction. Implement this to provide a custom
99 /// transport (Postmark, logging, recording for tests, etc.).
100 #[async_trait::async_trait]
101 pub trait EmailTransport: Send + Sync {
102 /// Send a plain email.
103 async fn send_email(&self, to: &str, subject: &str, body: &str) -> Result<()>;
104
105 /// Send an email with an optional unsubscribe link.
106 async fn send_email_with_unsub(
107 &self,
108 to: &str,
109 subject: &str,
110 body: &str,
111 unsub_url: Option<&str>,
112 ) -> Result<()>;
113
114 /// Send an email with extra headers and an optional unsubscribe link.
115 async fn send_email_with_headers_and_unsub(
116 &self,
117 to: &str,
118 subject: &str,
119 body: &str,
120 extra_headers: &[(&str, String)],
121 unsub_url: Option<&str>,
122 ) -> Result<()>;
123
124 /// Send via the broadcast stream with an optional unsubscribe link.
125 async fn send_email_broadcast_with_unsub(
126 &self,
127 to: &str,
128 subject: &str,
129 body: &str,
130 unsub_url: Option<&str>,
131 ) -> Result<()>;
132 }
133
134 /// Send creator-departure notifications to historical buyers, bounded.
135 ///
136 /// Called from the two account-deletion confirmation paths (POST `/api/users/me`
137 /// and the email-link `GET` form-confirm). Account deletion is rare and the
138 /// notification is courtesy, but a creator with a very large completed-buyer
139 /// pool would otherwise turn one deletion into a Postmark spend bomb, which is
140 /// the same disease class the broadcast cap closes. Same parallelism + cadence
141 /// shape as `routes/api/users/broadcast.rs`.
142 ///
143 /// Recipients are capped at `BUYER_DEPARTURE_MAX_NOTIFICATIONS`; if the cap is
144 /// hit, the oldest-buyers slice (the SQL has no ORDER BY, so it's
145 /// implementation-dependent, but bounded) is notified and a warning is logged
146 /// so support can follow up manually for the remainder.
147 #[tracing::instrument(skip(pool, email_client, creator_name))]
148 pub async fn send_creator_departure_notifications(
149 pool: &sqlx::PgPool,
150 email_client: &EmailClient,
151 user_id: crate::db::UserId,
152 creator_name: String,
153 ) {
154 let buyers = match crate::db::transactions::get_all_buyers_for_seller(
155 pool,
156 user_id,
157 crate::constants::BUYER_DEPARTURE_MAX_NOTIFICATIONS,
158 )
159 .await
160 {
161 Ok(b) => b,
162 Err(e) => {
163 tracing::error!(error = ?e, %user_id, "failed to query buyers for departure notification");
164 return;
165 }
166 };
167 let count = buyers.len();
168 let cap = crate::constants::BUYER_DEPARTURE_MAX_NOTIFICATIONS as usize;
169 if count >= cap {
170 tracing::warn!(
171 %user_id, count, cap,
172 "creator-departure notification capped; remainder requires manual outreach"
173 );
174 } else {
175 tracing::info!(%user_id, buyer_count = count, "sending creator departure notifications");
176 }
177 let mut set = tokio::task::JoinSet::new();
178 let delay = std::time::Duration::from_millis(crate::constants::BROADCAST_CHUNK_DELAY_MS);
179 for buyer in buyers {
180 if set.len() >= crate::constants::BROADCAST_PARALLELISM {
181 let _ = set.join_next().await;
182 }
183 let email_client = email_client.clone();
184 let creator_name = creator_name.clone();
185 set.spawn(async move {
186 if let Err(e) = email_client
187 .send_creator_departure_notification(
188 &buyer.email,
189 buyer.display_name.as_deref(),
190 &creator_name,
191 )
192 .await
193 {
194 tracing::error!(error = ?e, buyer_email = %buyer.email, "failed to send creator departure notification");
195 }
196 });
197 tokio::time::sleep(delay).await;
198 }
199 while set.join_next().await.is_some() {}
200 }
201
202 /// Email client for sending emails
203 #[derive(Clone)]
204 pub struct EmailClient {
205 transport: Arc<dyn EmailTransport>,
206 }
207
208 impl EmailClient {
209 /// Create a new email client with Postmark transport.
210 pub fn new(config: EmailConfig, pool: Option<sqlx::PgPool>) -> Self {
211 EmailClient {
212 transport: Arc::new(PostmarkTransport::new(config, pool)),
213 }
214 }
215
216 /// Create an email client with a custom transport (for testing).
217 pub fn with_transport(transport: Arc<dyn EmailTransport>) -> Self {
218 EmailClient { transport }
219 }
220 }
221
222 /// Postmark-backed email transport (the production implementation).
223 #[derive(Clone)]
224 pub(crate) struct PostmarkTransport {
225 config: EmailConfig,
226 http_client: reqwest::Client,
227 pool: Option<sqlx::PgPool>,
228 }
229
230 impl PostmarkTransport {
231 fn new(config: EmailConfig, pool: Option<sqlx::PgPool>) -> Self {
232 let http_client = reqwest::Client::builder()
233 .timeout(std::time::Duration::from_secs(10))
234 .build()
235 .expect("Failed to build email HTTP client");
236
237 PostmarkTransport {
238 config,
239 http_client,
240 pool,
241 }
242 }
243
244 /// Shared implementation for send-with-unsubscribe, supporting optional message stream.
245 async fn send_with_unsub_inner(
246 &self,
247 to: &str,
248 subject: &str,
249 body: &str,
250 unsub_url: Option<&str>,
251 stream: Option<&str>,
252 ) -> Result<()> {
253 match unsub_url {
254 Some(url) => {
255 let body_with_footer = format!("{body}\n\nUnsubscribe from these emails:\n{url}");
256 let headers = [
257 ("List-Unsubscribe", format!("<{url}>")),
258 (
259 "List-Unsubscribe-Post",
260 "List-Unsubscribe=One-Click".to_string(),
261 ),
262 ];
263 self.send_email_inner(to, subject, &body_with_footer, &headers, stream)
264 .await
265 }
266 None => self.send_email_inner(to, subject, body, &[], stream).await,
267 }
268 }
269
270 /// Internal send implementation supporting optional custom headers and message stream.
271 async fn send_email_inner(
272 &self,
273 to: &str,
274 subject: &str,
275 body: &str,
276 extra_headers: &[(&str, String)],
277 stream: Option<&str>,
278 ) -> Result<()> {
279 // Check suppression list before sending
280 if let Some(ref pool) = self.pool {
281 match crate::db::email_suppressions::is_suppressed(pool, to).await {
282 Ok(true) => {
283 tracing::info!(recipient = %to, subject = %subject, "email skipped (suppressed)");
284 return Ok(());
285 }
286 Ok(false) => {}
287 Err(e) => {
288 // Log but don't block sending on suppression check failure
289 tracing::warn!(recipient = %to, error = %e, "suppression check failed, sending anyway");
290 }
291 }
292 }
293
294 if let Some(ref token) = self.config.postmark_token {
295 self.send_via_postmark(token, to, subject, body, extra_headers, stream)
296 .await
297 } else {
298 tracing::info!(
299 recipient = %to, subject = %subject,
300 "email sent (dev mode, body redacted)"
301 );
302 Ok(())
303 }
304 }
305
306 /// Send email via Postmark API
307 async fn send_via_postmark(
308 &self,
309 token: &str,
310 to: &str,
311 subject: &str,
312 body: &str,
313 extra_headers: &[(&str, String)],
314 stream: Option<&str>,
315 ) -> Result<()> {
316 let from = format!("{} <{}>", self.config.from_name, self.config.from_address);
317
318 let mut payload = serde_json::json!({
319 "From": from,
320 "To": to,
321 "Subject": subject,
322 "TextBody": body,
323 });
324
325 if let Some(stream_id) = stream {
326 payload["MessageStream"] = serde_json::Value::String(stream_id.to_string());
327 }
328
329 if !extra_headers.is_empty() {
330 let headers: Vec<serde_json::Value> = extra_headers
331 .iter()
332 .map(|(name, value)| serde_json::json!({ "Name": name, "Value": value }))
333 .collect();
334 payload["Headers"] = serde_json::Value::Array(headers);
335 }
336
337 // Retry transient failures (network/timeout, 5xx, 429) with bounded
338 // exponential backoff so a brief Postmark blip doesn't permanently drop
339 // critical mail, password resets, purchase receipts, Fan+ credit codes
340 // (Run 20 Resilience). Permanent 4xx (bad request, inactive recipient,
341 // hard bounce) are NOT retried: retrying can't help and only delays the
342 // caller. Bounded to EMAIL_SEND_MAX_ATTEMPTS so an awaited caller adds at
343 // most ~1s on a failing send.
344 const EMAIL_SEND_MAX_ATTEMPTS: u32 = 3;
345 let mut attempt: u32 = 0;
346 loop {
347 attempt += 1;
348 let send_result = self
349 .http_client
350 .post("https://api.postmarkapp.com/email")
351 .header("X-Postmark-Server-Token", token)
352 .header("Content-Type", "application/json")
353 .json(&payload)
354 .send()
355 .await;
356
357 match send_result {
358 Ok(response) if response.status().is_success() => {
359 tracing::info!(recipient = %to, subject = %subject, attempt, "email sent");
360 return Ok(());
361 }
362 Ok(response) => {
363 let status = response.status();
364 let transient = status.is_server_error() || status.as_u16() == 429;
365 let error_text = response.text().await.unwrap_or_default();
366 if transient && attempt < EMAIL_SEND_MAX_ATTEMPTS {
367 let backoff = std::time::Duration::from_millis(200 * 2u64.pow(attempt - 1));
368 tracing::warn!(status = %status, attempt, error = %error_text, "transient email send failure, retrying after backoff");
369 tokio::time::sleep(backoff).await;
370 continue;
371 }
372 tracing::error!(status = %status, error = %error_text, attempt, "failed to send email");
373 return Err(AppError::Internal(anyhow::anyhow!(
374 "Failed to send email: {status}"
375 )));
376 }
377 Err(e) => {
378 // Network/timeout: always transient.
379 if attempt < EMAIL_SEND_MAX_ATTEMPTS {
380 let backoff = std::time::Duration::from_millis(200 * 2u64.pow(attempt - 1));
381 tracing::warn!(attempt, error = %e, "email send request error, retrying after backoff");
382 tokio::time::sleep(backoff).await;
383 continue;
384 }
385 return Err(AppError::Internal(anyhow::anyhow!(
386 "postmark http request: {e}"
387 )));
388 }
389 }
390 }
391 }
392 }
393
394 #[async_trait::async_trait]
395 impl EmailTransport for PostmarkTransport {
396 async fn send_email(&self, to: &str, subject: &str, body: &str) -> Result<()> {
397 self.send_email_inner(to, subject, body, &[], None).await
398 }
399
400 async fn send_email_with_unsub(
401 &self,
402 to: &str,
403 subject: &str,
404 body: &str,
405 unsub_url: Option<&str>,
406 ) -> Result<()> {
407 self.send_with_unsub_inner(to, subject, body, unsub_url, None)
408 .await
409 }
410
411 async fn send_email_with_headers_and_unsub(
412 &self,
413 to: &str,
414 subject: &str,
415 body: &str,
416 extra_headers: &[(&str, String)],
417 unsub_url: Option<&str>,
418 ) -> Result<()> {
419 match unsub_url {
420 Some(url) => {
421 let body_with_footer = format!("{body}\n\nUnsubscribe from these emails:\n{url}");
422 let mut all_headers: Vec<(&str, String)> = extra_headers.to_vec();
423 all_headers.push(("List-Unsubscribe", format!("<{url}>")));
424 all_headers.push((
425 "List-Unsubscribe-Post",
426 "List-Unsubscribe=One-Click".to_string(),
427 ));
428 self.send_email_inner(to, subject, &body_with_footer, &all_headers, None)
429 .await
430 }
431 None => {
432 self.send_email_inner(to, subject, body, extra_headers, None)
433 .await
434 }
435 }
436 }
437
438 async fn send_email_broadcast_with_unsub(
439 &self,
440 to: &str,
441 subject: &str,
442 body: &str,
443 unsub_url: Option<&str>,
444 ) -> Result<()> {
445 self.send_with_unsub_inner(to, subject, body, unsub_url, Some("broadcast"))
446 .await
447 }
448 }
449
450 #[cfg(test)]
451 mod bounded_recipients_tests {
452 use super::*;
453
454 #[test]
455 fn accepts_under_cap() {
456 let r = BoundedRecipients::new(vec![1, 2, 3]).expect("under cap");
457 assert_eq!(r.len(), 3);
458 assert!(!r.is_empty());
459 assert_eq!(r.into_inner(), vec![1, 2, 3]);
460 }
461
462 #[test]
463 fn accepts_exactly_at_cap() {
464 let v = vec![0u8; crate::constants::BROADCAST_MAX_RECIPIENTS];
465 assert!(BoundedRecipients::new(v).is_ok());
466 }
467
468 #[test]
469 fn rejects_over_cap_with_count() {
470 let n = crate::constants::BROADCAST_MAX_RECIPIENTS + 1;
471 let v = vec![0u8; n];
472 assert_eq!(BoundedRecipients::new(v).unwrap_err(), n);
473 }
474
475 #[test]
476 fn empty_is_allowed() {
477 let r = BoundedRecipients::<u8>::new(vec![]).expect("empty ok");
478 assert!(r.is_empty());
479 }
480 }
481
482 /// Seal for the broadcast recipient cap.
483 ///
484 /// The cap was a copy-pasted `if count > BROADCAST_MAX_RECIPIENTS` check that had
485 /// drifted between the public and internal broadcast handlers. The fix routes
486 /// both through [`BoundedRecipients`], so the constant must appear ONLY in its
487 /// definition (`constants.rs`) and this module's constructor. This test fails
488 /// the build if any other file references the constant, forcing a new broadcast
489 /// site through the sealed constructor instead of re-inlining the check.
490 #[cfg(test)]
491 mod broadcast_cap_seal_guard {
492 use std::path::Path;
493
494 #[test]
495 fn cap_constant_used_only_in_sealed_constructor() {
496 let src_dir = Path::new(env!("CARGO_MANIFEST_DIR")).join("src");
497 let allowed = [
498 Path::new(env!("CARGO_MANIFEST_DIR")).join("src/constants.rs"),
499 Path::new(env!("CARGO_MANIFEST_DIR")).join("src/email/mod.rs"),
500 ];
501 let mut offenders = Vec::new();
502 walk(&src_dir, &mut |path, contents| {
503 if allowed.iter().any(|a| a == path) {
504 return;
505 }
506 for (i, line) in contents.lines().enumerate() {
507 if line.contains("BROADCAST_MAX_RECIPIENTS") {
508 offenders.push(format!("{}:{}: {}", path.display(), i + 1, line.trim()));
509 }
510 }
511 });
512 assert!(
513 offenders.is_empty(),
514 "broadcast-cap seal violated, enforce the recipient cap via \
515 email::BoundedRecipients::new, never by referencing BROADCAST_MAX_RECIPIENTS \
516 directly in a handler. Offending lines:\n{}",
517 offenders.join("\n")
518 );
519 }
520
521 fn walk(dir: &Path, f: &mut impl FnMut(&Path, &str)) {
522 let Ok(entries) = std::fs::read_dir(dir) else {
523 return;
524 };
525 for entry in entries.flatten() {
526 let path = entry.path();
527 if path.is_dir() {
528 walk(&path, f);
529 } else if path.extension().is_some_and(|e| e == "rs")
530 && let Ok(contents) = std::fs::read_to_string(&path)
531 {
532 f(&path, &contents);
533 }
534 }
535 }
536 }
537