diff --git a/crates/temps-email/src/errors.rs b/crates/temps-email/src/errors.rs index 343209ec1..a43e22de6 100644 --- a/crates/temps-email/src/errors.rs +++ b/crates/temps-email/src/errors.rs @@ -2,6 +2,8 @@ use thiserror::Error; +use crate::providers::EmailProviderType; + #[derive(Error, Debug)] pub enum EmailError { #[error("Database error: {0}")] @@ -51,6 +53,19 @@ pub enum EmailError { #[error("Tracking rewrite failed for email {email_id}: {reason}")] TrackingRewrite { email_id: String, reason: String }, + + /// A provider's `send()` call failed. Unlike the flat per-provider + /// variants above (used for identity/domain management calls), + /// this carries a `retryable` classification derived from the + /// underlying transport error (HTTP status, SMTP reply code, or AWS SDK + /// error kind) so the send path knows whether retrying — same provider + /// or the next one in the failover chain — can plausibly succeed. + #[error("Failed to send email via {provider}: {reason}")] + SendFailed { + provider: EmailProviderType, + retryable: bool, + reason: String, + }, } impl From for EmailError { @@ -58,3 +73,13 @@ impl From for EmailError { EmailError::Serialization(err.to_string()) } } + +impl EmailError { + /// Whether this failure is transient and worth retrying (against the + /// same provider or the next one in a domain's failover chain). Every + /// variant besides `SendFailed { retryable: true, .. }` represents a + /// config, validation, or lookup problem that a retry cannot fix. + pub fn is_retryable(&self) -> bool { + matches!(self, EmailError::SendFailed { retryable: true, .. }) + } +} diff --git a/crates/temps-email/src/handlers/domains.rs b/crates/temps-email/src/handlers/domains.rs index caa1cebf4..c21b2e0ff 100644 --- a/crates/temps-email/src/handlers/domains.rs +++ b/crates/temps-email/src/handlers/domains.rs @@ -65,6 +65,10 @@ impl From for Problem { .with_title("Internal Server Error") .with_detail(error.to_string()) } + + EmailError::SendFailed { .. } => problemdetails::new(StatusCode::BAD_GATEWAY) + .with_title("Provider Send Failed") + .with_detail(error.to_string()), } } } diff --git a/crates/temps-email/src/handlers/tracking_tests.rs b/crates/temps-email/src/handlers/tracking_tests.rs index d035bdc57..26ec6c52c 100644 --- a/crates/temps-email/src/handlers/tracking_tests.rs +++ b/crates/temps-email/src/handlers/tracking_tests.rs @@ -24,8 +24,8 @@ mod tests { use crate::handlers::tracking::{public_routes, routes}; use crate::handlers::types::AppState; use crate::services::{ - DomainService, EmailService, ProviderService, TrackingService, ValidationConfig, - ValidationService, + DomainService, EmailService, ProviderService, SuppressionService, TrackingService, + ValidationConfig, ValidationService, }; // ============================================ @@ -120,11 +120,13 @@ mod tests { config_service, "http://localhost:3000".to_string(), )); + let suppression_service = Arc::new(SuppressionService::new(db.db.clone())); let email_service = Arc::new(EmailService::new( db.db.clone(), provider_service.clone(), domain_service.clone(), tracking_service.clone(), + suppression_service, )); let validation_service = Arc::new(ValidationService::new(ValidationConfig::default())); diff --git a/crates/temps-email/src/plugin.rs b/crates/temps-email/src/plugin.rs index 360ac3c85..b1bbba789 100644 --- a/crates/temps-email/src/plugin.rs +++ b/crates/temps-email/src/plugin.rs @@ -13,8 +13,8 @@ use utoipa::OpenApi as OpenApiTrait; use crate::handlers::{self, AppState, EmailApiDoc}; use crate::services::{ - DomainService, EmailService, ProviderService, TrackingService, ValidationConfig, - ValidationService, + DomainService, EmailService, ProviderService, SuppressionService, TrackingService, + ValidationConfig, ValidationService, }; use temps_dns::services::DnsProviderService; @@ -61,12 +61,17 @@ impl TempsPlugin for EmailPlugin { let tracking_service = Arc::new(TrackingService::new(db.clone(), config_service)); context.register_service(tracking_service.clone()); - // Create EmailService with tracking support + // Create SuppressionService — bounce/complaint do-not-send list + let suppression_service = Arc::new(SuppressionService::new(db.clone())); + context.register_service(suppression_service.clone()); + + // Create EmailService with tracking + suppression support let email_service = Arc::new(EmailService::new( db.clone(), provider_service.clone(), domain_service.clone(), tracking_service.clone(), + suppression_service.clone(), )); context.register_service(email_service.clone()); diff --git a/crates/temps-email/src/providers/scaleway.rs b/crates/temps-email/src/providers/scaleway.rs index c1f726ce3..4583fbecd 100644 --- a/crates/temps-email/src/providers/scaleway.rs +++ b/crates/temps-email/src/providers/scaleway.rs @@ -398,7 +398,14 @@ impl EmailProvider for ScalewayProvider { .json(&request) .send() .await - .map_err(|e| EmailError::Scaleway(format!("Failed to send email: {}", e)))?; + .map_err(|e| EmailError::SendFailed { + provider: EmailProviderType::Scaleway, + // A request that never reached Scaleway (DNS/connect/timeout) is + // worth retrying — it says nothing about whether the message itself + // is deliverable. + retryable: true, + reason: format!("Failed to send email: {}", e), + })?; if !response.status().is_success() { let status = response.status(); @@ -407,23 +414,33 @@ impl EmailProvider for ScalewayProvider { .await .unwrap_or_else(|_| "Unknown error".to_string()); error!("Failed to send email via Scaleway ({}): {}", status, body); - return Err(EmailError::Scaleway(format!( - "Failed to send email ({}): {}", - status, body - ))); + return Err(EmailError::SendFailed { + provider: EmailProviderType::Scaleway, + // 5xx and 429 are Scaleway-side transient conditions; any other + // 4xx (bad request, invalid recipient, auth failure) won't change + // on retry. + retryable: status.is_server_error() || status.as_u16() == 429, + reason: format!("Failed to send email ({}): {}", status, body), + }); } - let email_response: ScalewayEmailResponse = response - .json() - .await - .map_err(|e| EmailError::Scaleway(format!("Failed to parse email response: {}", e)))?; + let email_response: ScalewayEmailResponse = + response.json().await.map_err(|e| EmailError::SendFailed { + provider: EmailProviderType::Scaleway, + retryable: false, + reason: format!("Failed to parse email response: {}", e), + })?; let message_id = email_response .emails .first() .and_then(|e| e.message_id.clone()) .or_else(|| email_response.emails.first().map(|e| e.id.clone())) - .ok_or_else(|| EmailError::Scaleway("No message ID returned".to_string()))?; + .ok_or_else(|| EmailError::SendFailed { + provider: EmailProviderType::Scaleway, + retryable: false, + reason: "No message ID returned".to_string(), + })?; debug!("Email sent successfully, message_id: {}", message_id); diff --git a/crates/temps-email/src/providers/ses.rs b/crates/temps-email/src/providers/ses.rs index 221a9ccc8..e54b70e3c 100644 --- a/crates/temps-email/src/providers/ses.rs +++ b/crates/temps-email/src/providers/ses.rs @@ -57,6 +57,52 @@ fn extract_ses_error_details( } } +/// Classify whether an SES `send_email` failure is worth retrying. +/// Network/timeout failures never reached AWS and are always worth another +/// attempt. Service errors are retryable only for AWS's own typed +/// throttling/capacity exceptions — a message-level rejection (bad +/// recipient, unverified sender, suspended/paused account) will fail +/// identically on retry. Matches on the concrete `SendEmailError` variants +/// rather than string-matching the error message, which is fragile against +/// SDK wording changes. +fn is_send_email_error_retryable( + e: &aws_sdk_sesv2::error::SdkError, +) -> bool { + use aws_sdk_sesv2::error::SdkError; + use aws_sdk_sesv2::operation::send_email::SendEmailError; + use aws_sdk_sesv2::error::ProvideErrorMetadata; + + match e { + SdkError::TimeoutError(_) => true, + SdkError::DispatchFailure(dispatch_err) => { + dispatch_err.is_io() || dispatch_err.is_timeout() + } + SdkError::ResponseError(_) => true, + SdkError::ConstructionFailure(_) => false, + SdkError::ServiceError(service_err) => match service_err.err() { + SendEmailError::TooManyRequestsException(_) + | SendEmailError::LimitExceededException(_) => true, + SendEmailError::AccountSuspendedException(_) + | SendEmailError::BadRequestException(_) + | SendEmailError::MailFromDomainNotVerifiedException(_) + | SendEmailError::MessageRejected(_) + | SendEmailError::NotFoundException(_) + | SendEmailError::SendingPausedException(_) => false, + // Any future/unhandled variant: fall back to the error code for + // AWS-side throttling/capacity conditions rather than assuming + // permanent. + other => matches!( + other.code(), + Some("ThrottlingException") + | Some("ServiceUnavailable") + | Some("InternalFailure") + | Some("InternalServerError") + ), + }, + _ => false, + } +} + /// AWS SES credentials configuration #[derive(Debug, Clone, Serialize, Deserialize)] pub struct SesCredentials { @@ -468,13 +514,26 @@ impl EmailProvider for SesProvider { let result = send_request.send().await.map_err(|e| { // Extract detailed error information from AWS SDK error let error_message = extract_ses_error_details(&e); - error!("Failed to send email via SES: {}", error_message); - EmailError::AwsSes(error_message) + let retryable = is_send_email_error_retryable(&e); + error!( + "Failed to send email via SES ({}): {}", + if retryable { "retryable" } else { "permanent" }, + error_message + ); + EmailError::SendFailed { + provider: EmailProviderType::Ses, + retryable, + reason: error_message, + } })?; let message_id = result .message_id() - .ok_or_else(|| EmailError::AwsSes("No message ID returned".to_string()))? + .ok_or_else(|| EmailError::SendFailed { + provider: EmailProviderType::Ses, + retryable: false, + reason: "No message ID returned".to_string(), + })? .to_string(); debug!("Email sent successfully, message_id: {}", message_id); diff --git a/crates/temps-email/src/providers/smtp.rs b/crates/temps-email/src/providers/smtp.rs index 61c59c30c..62cc4d68d 100644 --- a/crates/temps-email/src/providers/smtp.rs +++ b/crates/temps-email/src/providers/smtp.rs @@ -279,7 +279,17 @@ impl EmailProvider for SmtpProvider { self.transport.send(message).await.map_err(|e| { error!("Failed to send email via SMTP: {}", e); - EmailError::Smtp(format!("Failed to send email: {}", e)) + // 4xx SMTP replies (is_transient) and connection-level failures + // (is_client/is_timeout/is_transport_shutdown) are worth retrying; + // a 5xx permanent reply from the relay will reject the same + // message again. + let retryable = + e.is_transient() || e.is_timeout() || e.is_client() || e.is_transport_shutdown(); + EmailError::SendFailed { + provider: EmailProviderType::Smtp, + retryable, + reason: format!("Failed to send email: {}", e), + } })?; debug!("Email sent via SMTP, message_id: {}", message_id); diff --git a/crates/temps-email/src/services/email_service.rs b/crates/temps-email/src/services/email_service.rs index 4831d0217..ec0a34442 100644 --- a/crates/temps-email/src/services/email_service.rs +++ b/crates/temps-email/src/services/email_service.rs @@ -12,7 +12,7 @@ use uuid::Uuid; use crate::errors::EmailError; use crate::providers::SendEmailRequest as ProviderSendRequest; -use crate::services::{DomainService, ProviderService, TrackingService}; +use crate::services::{DomainService, ProviderService, SuppressionService, TrackingService}; /// Trait for rewriting HTML to inject tracking (pixel + click links). /// Implemented by `temps-email-tracking::HtmlTrackingRewriter`. @@ -27,6 +27,7 @@ pub struct EmailService { domain_service: Arc, tracking_rewriter: Option>, tracking_service: Arc, + suppression_service: Arc, } /// Request to send an email @@ -75,6 +76,7 @@ impl EmailService { provider_service: Arc, domain_service: Arc, tracking_service: Arc, + suppression_service: Arc, ) -> Self { Self { db, @@ -82,6 +84,7 @@ impl EmailService { domain_service, tracking_rewriter: None, tracking_service, + suppression_service, } } @@ -181,6 +184,62 @@ impl EmailService { } } + // Refuse to send to previously hard-bounced/complained addresses — + // repeatedly emailing one is exactly what gets a sending domain's + // reputation downgraded by receiving mail providers. Checked after + // the row is inserted (still visible for debugging) but before any + // domain/provider work, since it's independent of both. + // + // Checks to/cc/bcc together (a suppressed address left in cc/bcc + // would otherwise still receive mail), and drops only the + // suppressed addresses rather than capturing the whole send — a + // suppressed address mixed into `to` alongside legitimate + // recipients used to silently deny delivery to everyone on the + // email, not just the bad address. + let mut all_recipients: Vec = request.to.clone(); + all_recipients.extend(request.cc.iter().flatten().cloned()); + all_recipients.extend(request.bcc.iter().flatten().cloned()); + + let suppressed = self + .suppression_service + .suppressed_among(&all_recipients) + .await?; + + if !suppressed.is_empty() { + info!( + "Dropping suppressed recipient(s) from email {}: {:?}", + email_id, suppressed + ); + } + let (to, cc, bcc) = + filter_suppressed_recipients(request.to, request.cc, request.bcc, &suppressed); + + // Nothing left to send to (either every `to` address was + // suppressed, or `to` was already empty) — capture instead of + // sending an email with no primary recipient. + if to.is_empty() { + info!( + "Refusing to send email {} — all recipient(s) suppressed: {:?}", + email_id, suppressed + ); + + let mut active_model: emails::ActiveModel = email_model.into(); + active_model.status = Set("captured".to_string()); + active_model.error_message = Set(Some(format!( + "Recipient(s) suppressed (previous hard bounce or complaint): {}", + suppressed.join(", ") + ))); + active_model.sent_at = Set(Some(Utc::now())); + + active_model.update(self.db.as_ref()).await?; + + return Ok(SendEmailResponse { + id: email_id, + status: "captured".to_string(), + provider_message_id: None, + }); + } + // If no domain configured, capture email without sending (Mailhog-like behavior) let domain = match domain { Some(d) => d, @@ -198,7 +257,7 @@ impl EmailService { info!( "Email captured (no domain configured), id: {}, from: {}, to: {:?}", - email_id, request.from, request.to + email_id, request.from, to ); return Ok(SendEmailResponse { @@ -233,7 +292,10 @@ impl EmailService { }); } - // Try to get provider - if not configured, capture email + // Look up the domain's configured provider. Previously `is_active` + // only hid a provider from selection UI while the send path kept + // using it regardless — checked explicitly below so disabling a + // provider actually takes it out of the send path. let provider = match self.provider_service.get(domain.provider_id).await { Ok(p) => Some(p), Err(e) => { @@ -246,27 +308,45 @@ impl EmailService { } }; - // If no provider, mark as captured and return success - if provider.is_none() { - let mut active_model: emails::ActiveModel = email_model.into(); - active_model.status = Set("captured".to_string()); - active_model.sent_at = Set(Some(Utc::now())); + let provider = match provider { + Some(p) if p.is_active => p, + Some(p) => { + info!( + "Provider '{}' for domain '{}' is inactive, capturing email without sending", + p.name, domain.domain + ); + let mut active_model: emails::ActiveModel = email_model.into(); + active_model.status = Set("captured".to_string()); + active_model.error_message = + Set(Some(format!("Provider '{}' is inactive", p.name))); + active_model.sent_at = Set(Some(Utc::now())); + active_model.update(self.db.as_ref()).await?; - active_model.update(self.db.as_ref()).await?; + return Ok(SendEmailResponse { + id: email_id, + status: "captured".to_string(), + provider_message_id: None, + }); + } + None => { + let mut active_model: emails::ActiveModel = email_model.into(); + active_model.status = Set("captured".to_string()); + active_model.sent_at = Set(Some(Utc::now())); - info!( - "Email captured (no provider), id: {}, from: {}, to: {:?}", - email_id, request.from, request.to - ); + active_model.update(self.db.as_ref()).await?; - return Ok(SendEmailResponse { - id: email_id, - status: "captured".to_string(), - provider_message_id: None, - }); - } + info!( + "Email captured (no provider), id: {}, from: {}, to: {:?}", + email_id, request.from, to + ); - let provider = provider.unwrap(); + return Ok(SendEmailResponse { + id: email_id, + status: "captured".to_string(), + provider_message_id: None, + }); + } + }; let provider_instance = match self .provider_service @@ -294,14 +374,12 @@ impl EmailService { } }; - // Use tracked HTML (with open/click tracking injected) if available - let provider_request = ProviderSendRequest { from: request.from, from_name: request.from_name, - to: request.to, - cc: request.cc, - bcc: request.bcc, + to, + cc, + bcc, reply_to: request.reply_to, subject: request.subject, html: tracked_html, @@ -309,19 +387,73 @@ impl EmailService { headers: request.headers, }; - match provider_instance.send(&provider_request).await { - Ok(response) => { - // Update email with success status - let mut active_model: emails::ActiveModel = email_model.clone().into(); + // Retry once more only if the failure was classified as transient + // (`EmailError::is_retryable`) — a permanent rejection (bad + // recipient, auth failure, unverified sender) fails identically on + // retry, so a second attempt would just waste time. The whole + // attempt sequence is bounded by an overall deadline so a + // slow/hanging provider can't stall this request indefinitely even + // across multiple attempts. + const MAX_ATTEMPTS: u32 = 2; + const RETRY_DELAY: std::time::Duration = std::time::Duration::from_millis(500); + const TOTAL_SEND_DEADLINE: std::time::Duration = std::time::Duration::from_secs(45); + + let attempt_result = tokio::time::timeout(TOTAL_SEND_DEADLINE, async { + let mut attempt_count: i32 = 0; + let mut last_error: Option = None; + + for attempt in 1..=MAX_ATTEMPTS { + attempt_count += 1; + + match provider_instance.send(&provider_request).await { + Ok(response) => return (attempt_count, Some(response), None), + Err(e) => { + let retryable = e.is_retryable(); + warn!( + "Send attempt {}/{} via {} failed for email {}: {}", + attempt, MAX_ATTEMPTS, provider.name, email_id, e + ); + last_error = Some(e.to_string()); + + if !retryable || attempt == MAX_ATTEMPTS { + break; + } + tokio::time::sleep(RETRY_DELAY).await; + } + } + } + + (attempt_count, None, last_error) + }) + .await; + + let (attempt_count, sent, last_error) = match attempt_result { + Ok(result) => result, + Err(_) => ( + MAX_ATTEMPTS as i32, + None, + Some(format!( + "Send timed out after {:?} across all attempts", + TOTAL_SEND_DEADLINE + )), + ), + }; + + let mut active_model: emails::ActiveModel = email_model.into(); + active_model.attempt_count = Set(attempt_count); + active_model.provider_id = Set(Some(provider.id)); + + match sent { + Some(response) => { active_model.status = Set("sent".to_string()); active_model.provider_message_id = Set(Some(response.message_id.clone())); active_model.sent_at = Set(Some(Utc::now())); - let _email_model = active_model.update(self.db.as_ref()).await?; + active_model.update(self.db.as_ref()).await?; info!( - "Email sent successfully, id: {}, provider_message_id: {}", - email_id, response.message_id + "Email sent successfully, id: {}, provider_message_id: {}, attempts: {}", + email_id, response.message_id, attempt_count ); Ok(SendEmailResponse { @@ -330,16 +462,15 @@ impl EmailService { provider_message_id: Some(response.message_id), }) } - Err(e) => { - // Provider send failed - capture email instead of failing + None => { + let reason = last_error.unwrap_or_else(|| "unknown error".to_string()); info!( - "Failed to send email via provider, capturing instead: {}", - e + "Failed to send email {} via provider {}, capturing instead: {}", + email_id, provider.name, reason ); - let mut active_model: emails::ActiveModel = email_model.into(); active_model.status = Set("captured".to_string()); - active_model.error_message = Set(Some(format!("Send failed: {}", e))); + active_model.error_message = Set(Some(format!("Send failed: {}", reason))); active_model.sent_at = Set(Some(Utc::now())); active_model.update(self.db.as_ref()).await?; @@ -447,6 +578,33 @@ pub struct EmailStats { pub captured: u64, } +/// Drop suppressed addresses from `to`/`cc`/`bcc`. `suppressed` is the +/// (already-normalized) output of `SuppressionService::suppressed_among`. +/// +/// Filters each list independently rather than rejecting the whole send — +/// a suppressed address mixed into `to` alongside legitimate recipients +/// must not deny delivery to everyone on the email, just to itself. +fn filter_suppressed_recipients( + to: Vec, + cc: Option>, + bcc: Option>, + suppressed: &[String], +) -> (Vec, Option>, Option>) { + if suppressed.is_empty() { + return (to, cc, bcc); + } + + let suppressed_set: std::collections::HashSet<&str> = + suppressed.iter().map(String::as_str).collect(); + let keep = |addr: &String| !suppressed_set.contains(SuppressionService::normalize(addr).as_str()); + + ( + to.into_iter().filter(&keep).collect(), + cc.map(|list| list.into_iter().filter(&keep).collect()), + bcc.map(|list| list.into_iter().filter(&keep).collect()), + ) +} + #[cfg(test)] mod tests { use super::*; @@ -501,11 +659,13 @@ mod tests { config_service, "http://localhost:3000".to_string(), )); + let suppression_service = Arc::new(SuppressionService::new(db.db.clone())); let email_service = EmailService::new( db.db.clone(), Arc::new(provider_service.clone()), Arc::new(domain_service.clone()), tracking_service, + suppression_service, ); (db, email_service, provider_service, domain_service) } @@ -662,6 +822,81 @@ mod tests { assert!(domain.is_none()); } + #[test] + fn filter_suppressed_recipients_passes_through_when_nothing_suppressed() { + let (to, cc, bcc) = filter_suppressed_recipients( + vec!["a@example.com".to_string()], + Some(vec!["b@example.com".to_string()]), + None, + &[], + ); + assert_eq!(to, vec!["a@example.com"]); + assert_eq!(cc, Some(vec!["b@example.com".to_string()])); + assert_eq!(bcc, None); + } + + #[test] + fn filter_suppressed_recipients_drops_only_the_suppressed_cc_address() { + // A suppressed address in `cc` used to be invisible to the check + // entirely — it must be dropped from the send, and it must not take + // the legitimate `to` recipient down with it. + let (to, cc, bcc) = filter_suppressed_recipients( + vec!["good@example.com".to_string()], + Some(vec![ + "bad@example.com".to_string(), + "also-good@example.com".to_string(), + ]), + None, + &["bad@example.com".to_string()], + ); + assert_eq!(to, vec!["good@example.com"]); + assert_eq!(cc, Some(vec!["also-good@example.com".to_string()])); + assert_eq!(bcc, None); + } + + #[test] + fn filter_suppressed_recipients_keeps_other_to_addresses() { + // One suppressed address mixed into `to` used to capture the whole + // send — the other `to` recipients must still get the email. + let (to, cc, bcc) = filter_suppressed_recipients( + vec![ + "bad@example.com".to_string(), + "good@example.com".to_string(), + ], + None, + None, + &["bad@example.com".to_string()], + ); + assert_eq!(to, vec!["good@example.com"]); + assert_eq!(cc, None); + assert_eq!(bcc, None); + } + + #[test] + fn filter_suppressed_recipients_matches_case_and_whitespace_insensitively() { + // `suppressed_among` returns normalized (trimmed/lowercased) forms + // from the DB — the filter must normalize candidates the same way, + // not compare raw strings. + let (to, _, _) = filter_suppressed_recipients( + vec![" Bad@Example.COM ".to_string(), "good@example.com".to_string()], + None, + None, + &["bad@example.com".to_string()], + ); + assert_eq!(to, vec!["good@example.com"]); + } + + #[test] + fn filter_suppressed_recipients_empties_to_when_all_suppressed() { + let (to, _, _) = filter_suppressed_recipients( + vec!["bad@example.com".to_string()], + None, + None, + &["bad@example.com".to_string()], + ); + assert!(to.is_empty()); + } + #[test] fn test_list_emails_options_builder() { // Test that list options can be constructed with various filters @@ -836,6 +1071,49 @@ mod tests { assert_eq!(response.status, "captured"); } + #[tokio::test] + async fn test_send_email_inactive_provider_is_captured_not_sent() { + // Previously `is_active` only hid a provider from selection UI — + // the send path kept using it regardless. A verified domain + // pointing at a disabled provider must be captured, not attempt an + // actual send through credentials the operator turned off. + let (db, email_service, provider_service, _domain_service) = setup_test_env().await; + + let provider = create_test_provider(&provider_service).await; + provider_service.set_active(provider.id, false).await.unwrap(); + + let domain = create_test_domain(&db.db, provider.id, "test-inactive.example.com").await; + let mut active_model: temps_entities::email_domains::ActiveModel = domain.into(); + active_model.status = sea_orm::ActiveValue::Set("verified".to_string()); + active_model.update(db.db.as_ref()).await.unwrap(); + + let request = SendEmailRequest { + from: "sender@test-inactive.example.com".to_string(), + from_name: None, + to: vec!["recipient@test.com".to_string()], + cc: None, + bcc: None, + reply_to: None, + subject: "Test".to_string(), + html: Some("

Test

".to_string()), + text: None, + headers: None, + tags: None, + track_opens: false, + track_clicks: false, + }; + + let result = email_service.send(request).await.unwrap(); + + assert_eq!(result.status, "captured"); + let stored = email_service.get(result.id).await.unwrap(); + assert!(stored + .error_message + .as_deref() + .unwrap() + .contains("is inactive")); + } + #[tokio::test] async fn test_list_emails_with_filters() { let (_db, email_service, _provider_service, _domain_service) = setup_test_env().await; diff --git a/crates/temps-email/src/services/mod.rs b/crates/temps-email/src/services/mod.rs index 94be4b7ba..5686c453f 100644 --- a/crates/temps-email/src/services/mod.rs +++ b/crates/temps-email/src/services/mod.rs @@ -3,6 +3,7 @@ mod domain_service; mod email_service; mod provider_service; +mod suppression_service; mod tracking_service; #[cfg(test)] mod tracking_service_integration_tests; @@ -17,6 +18,7 @@ pub use provider_service::{ CreateProviderRequest, ProviderCredentials, ProviderService, TestEmailResult, UpdateProviderOutcome, UpdateProviderRequest, }; +pub use suppression_service::{SuppressionReason, SuppressionService}; pub use tracking_service::{ExtractedLink, TrackingEvent, TrackingService, TransformResult}; pub use validation::{ MiscResult, MxResult, ProxyConfig, ReachabilityStatus, SmtpResult, SyntaxResult, diff --git a/crates/temps-email/src/services/suppression_service.rs b/crates/temps-email/src/services/suppression_service.rs new file mode 100644 index 000000000..ecdb021b9 --- /dev/null +++ b/crates/temps-email/src/services/suppression_service.rs @@ -0,0 +1,287 @@ +//! Suppression list — recipients who must not receive further email due to +//! a hard bounce, a spam complaint, or a manual admin action. Checked by +//! `EmailService::send` before every send: without this, a permanently-bad +//! or complained address kept getting mailed on every subsequent send, +//! which is the exact pattern that gets a sending domain downgraded by +//! receiving mail providers. + +use sea_orm::{ + ActiveModelTrait, ActiveValue::Set, ColumnTrait, DatabaseConnection, EntityTrait, + PaginatorTrait, QueryFilter, QueryOrder, +}; +use std::sync::Arc; +use temps_entities::suppressed_recipients; + +use crate::errors::EmailError; + +/// Why an address was suppressed. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum SuppressionReason { + /// A hard/permanent bounce (mailbox doesn't exist, domain rejects mail, …). + Bounced, + /// The recipient marked a message as spam. + Complained, + /// An admin suppressed (or un-suppressed) the address by hand. + Manual, +} + +impl SuppressionReason { + fn as_str(&self) -> &'static str { + match self { + SuppressionReason::Bounced => "bounced", + SuppressionReason::Complained => "complained", + SuppressionReason::Manual => "manual", + } + } +} + +/// Service for managing the email suppression list. +pub struct SuppressionService { + db: Arc, +} + +impl SuppressionService { + pub fn new(db: Arc) -> Self { + Self { db } + } + + /// Trim + lowercase for storage/lookup comparison. `pub(crate)` so + /// callers filtering a recipient list against `suppressed_among`'s + /// results (which come back normalized, not in their original casing) + /// can match them correctly. + pub(crate) fn normalize(email: &str) -> String { + email.trim().to_lowercase() + } + + /// Add an address to the suppression list, or update its reason if it's + /// already there (e.g. a bounce followed later by a complaint). + pub async fn suppress( + &self, + email: &str, + reason: SuppressionReason, + domain_id: Option, + detail: Option, + ) -> Result<(), EmailError> { + let normalized = Self::normalize(email); + + let existing = suppressed_recipients::Entity::find() + .filter(suppressed_recipients::Column::Email.eq(normalized.clone())) + .one(self.db.as_ref()) + .await?; + + match existing { + Some(model) => { + let mut active: suppressed_recipients::ActiveModel = model.into(); + active.reason = Set(reason.as_str().to_string()); + active.domain_id = Set(domain_id); + active.detail = Set(detail); + active.update(self.db.as_ref()).await?; + } + None => { + let active = suppressed_recipients::ActiveModel { + email: Set(normalized), + reason: Set(reason.as_str().to_string()), + domain_id: Set(domain_id), + detail: Set(detail), + ..Default::default() + }; + active.insert(self.db.as_ref()).await?; + } + } + + Ok(()) + } + + /// Remove an address from the suppression list (manual admin override — + /// e.g. the mailbox was fixed, or the bounce/complaint was a mistake). + pub async fn unsuppress(&self, email: &str) -> Result<(), EmailError> { + let normalized = Self::normalize(email); + suppressed_recipients::Entity::delete_many() + .filter(suppressed_recipients::Column::Email.eq(normalized)) + .exec(self.db.as_ref()) + .await?; + Ok(()) + } + + /// Is this single address currently suppressed? + pub async fn is_suppressed(&self, email: &str) -> Result { + let normalized = Self::normalize(email); + let count = suppressed_recipients::Entity::find() + .filter(suppressed_recipients::Column::Email.eq(normalized)) + .count(self.db.as_ref()) + .await?; + Ok(count > 0) + } + + /// Which of these addresses are currently suppressed — one query instead + /// of N, for `EmailService::send` checking every `to` recipient at once. + pub async fn suppressed_among(&self, emails: &[String]) -> Result, EmailError> { + if emails.is_empty() { + return Ok(Vec::new()); + } + let normalized: Vec = emails.iter().map(|e| Self::normalize(e)).collect(); + let rows = suppressed_recipients::Entity::find() + .filter(suppressed_recipients::Column::Email.is_in(normalized)) + .all(self.db.as_ref()) + .await?; + Ok(rows.into_iter().map(|r| r.email).collect()) + } + + /// Paginated list of the whole suppression list, most recent first. + pub async fn list( + &self, + page: u64, + page_size: u64, + ) -> Result<(Vec, u64), EmailError> { + let page = page.max(1); + let page_size = std::cmp::min(page_size, 100).max(1); + + let paginator = suppressed_recipients::Entity::find() + .order_by_desc(suppressed_recipients::Column::CreatedAt) + .paginate(self.db.as_ref(), page_size); + + let total = paginator.num_items().await?; + let items = paginator.fetch_page(page - 1).await?; + Ok((items, total)) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use temps_database::test_utils::TestDatabase; + + async fn setup() -> (TestDatabase, SuppressionService) { + let db = TestDatabase::with_migrations().await.unwrap(); + let service = SuppressionService::new(db.db.clone()); + (db, service) + } + + #[test] + fn suppression_reason_as_str() { + assert_eq!(SuppressionReason::Bounced.as_str(), "bounced"); + assert_eq!(SuppressionReason::Complained.as_str(), "complained"); + assert_eq!(SuppressionReason::Manual.as_str(), "manual"); + } + + #[test] + fn normalize_trims_and_lowercases() { + assert_eq!( + SuppressionService::normalize(" Person@Example.COM "), + "person@example.com" + ); + } + + #[tokio::test] + async fn not_suppressed_by_default() { + let (_db, service) = setup().await; + assert!(!service.is_suppressed("nobody@example.com").await.unwrap()); + } + + #[tokio::test] + async fn suppress_then_is_suppressed() { + let (_db, service) = setup().await; + service + .suppress( + "Bounced@Example.com", + SuppressionReason::Bounced, + None, + Some("mailbox does not exist".to_string()), + ) + .await + .unwrap(); + + // Case/whitespace-insensitive lookup. + assert!(service + .is_suppressed(" bounced@example.com ") + .await + .unwrap()); + } + + #[tokio::test] + async fn suppress_is_idempotent_and_updates_reason() { + let (_db, service) = setup().await; + let email = "person@example.com"; + + service + .suppress(email, SuppressionReason::Bounced, None, None) + .await + .unwrap(); + service + .suppress(email, SuppressionReason::Complained, None, None) + .await + .unwrap(); + + let (rows, total) = service.list(1, 10).await.unwrap(); + assert_eq!(total, 1, "re-suppressing must not create a duplicate row"); + assert_eq!(rows[0].reason, "complained"); + } + + #[tokio::test] + async fn unsuppress_removes_the_address() { + let (_db, service) = setup().await; + let email = "person@example.com"; + service + .suppress(email, SuppressionReason::Manual, None, None) + .await + .unwrap(); + assert!(service.is_suppressed(email).await.unwrap()); + + service.unsuppress(email).await.unwrap(); + assert!(!service.is_suppressed(email).await.unwrap()); + } + + #[tokio::test] + async fn unsuppress_nonexistent_is_a_noop() { + let (_db, service) = setup().await; + assert!(service.unsuppress("nobody@example.com").await.is_ok()); + } + + #[tokio::test] + async fn suppressed_among_returns_only_matches() { + let (_db, service) = setup().await; + service + .suppress("bad@example.com", SuppressionReason::Bounced, None, None) + .await + .unwrap(); + + let result = service + .suppressed_among(&[ + "bad@example.com".to_string(), + "good@example.com".to_string(), + ]) + .await + .unwrap(); + + assert_eq!(result, vec!["bad@example.com".to_string()]); + } + + #[tokio::test] + async fn suppressed_among_empty_input_short_circuits() { + let (_db, service) = setup().await; + assert_eq!(service.suppressed_among(&[]).await.unwrap(), Vec::::new()); + } + + #[tokio::test] + async fn list_is_paginated_most_recent_first() { + let (_db, service) = setup().await; + for i in 0..3 { + service + .suppress( + &format!("person{i}@example.com"), + SuppressionReason::Manual, + None, + None, + ) + .await + .unwrap(); + } + + let (page1, total) = service.list(1, 2).await.unwrap(); + assert_eq!(total, 3); + assert_eq!(page1.len(), 2); + + let (page2, _) = service.list(2, 2).await.unwrap(); + assert_eq!(page2.len(), 1); + } +} diff --git a/crates/temps-entities/src/emails.rs b/crates/temps-entities/src/emails.rs index 93288128a..190b14a54 100644 --- a/crates/temps-entities/src/emails.rs +++ b/crates/temps-entities/src/emails.rs @@ -39,6 +39,13 @@ pub struct Model { pub click_count: i32, pub first_opened_at: Option, pub first_clicked_at: Option, + /// Which provider attempted (or ultimately completed) the send — set on + /// both success and exhausted-retry capture. `None` if the domain has no + /// active provider configured (never attempted). + pub provider_id: Option, + /// Total send attempts against `provider_id` (a clean first-try success + /// records 1), incremented on each retryable failure. + pub attempt_count: i32, } #[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] diff --git a/crates/temps-entities/src/lib.rs b/crates/temps-entities/src/lib.rs index f49a1789c..c803c909f 100644 --- a/crates/temps-entities/src/lib.rs +++ b/crates/temps-entities/src/lib.rs @@ -94,6 +94,7 @@ pub mod sessions; pub mod source_type; pub mod static_asset_cache; pub mod static_bundles; +pub mod suppressed_recipients; pub mod tls_acme_certificates; pub mod types; pub mod upstream_config; diff --git a/crates/temps-entities/src/suppressed_recipients.rs b/crates/temps-entities/src/suppressed_recipients.rs new file mode 100644 index 000000000..22a73e7b5 --- /dev/null +++ b/crates/temps-entities/src/suppressed_recipients.rs @@ -0,0 +1,41 @@ +//! Suppressed recipients entity — addresses that must not receive further +//! email due to a hard bounce, a spam complaint, or a manual admin action. +//! Checked by `EmailService::send` before every send. + +use sea_orm::entity::prelude::*; +use serde::{Deserialize, Serialize}; +use temps_core::DBDateTime; + +#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq, Serialize, Deserialize)] +#[sea_orm(table_name = "suppressed_recipients")] +pub struct Model { + #[sea_orm(primary_key)] + pub id: i32, + /// Always lowercased/trimmed before storage — see SuppressionService::normalize. + pub email: String, + /// "bounced" | "complained" | "manual" + pub reason: String, + /// Which domain's send triggered this, if known. Diagnostic only — + /// suppression is enforced globally, not scoped to one sending domain. + pub domain_id: Option, + pub detail: Option, + pub created_at: DBDateTime, +} + +#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] +pub enum Relation { + #[sea_orm( + belongs_to = "super::email_domains::Entity", + from = "Column::DomainId", + to = "super::email_domains::Column::Id" + )] + EmailDomain, +} + +impl Related for Entity { + fn to() -> RelationDef { + Relation::EmailDomain.def() + } +} + +impl ActiveModelBehavior for ActiveModel {} diff --git a/crates/temps-migrations/src/migration/m20260711_000002_create_suppressed_recipients.rs b/crates/temps-migrations/src/migration/m20260711_000002_create_suppressed_recipients.rs new file mode 100644 index 000000000..99100c06c --- /dev/null +++ b/crates/temps-migrations/src/migration/m20260711_000002_create_suppressed_recipients.rs @@ -0,0 +1,49 @@ +//! Suppression list: recipients who must not receive further email due to a +//! hard bounce, a spam complaint, or a manual admin action. Without this, +//! nothing stopped a permanently-bad or complained address from being +//! emailed again on the next send, which is exactly the pattern that gets a +//! sending domain's reputation downgraded by receiving mail providers. +//! Enforced globally (not per sending domain) in `EmailService::send`. + +use sea_orm_migration::prelude::*; + +pub struct Migration; + +impl MigrationName for Migration { + fn name(&self) -> &str { + "m20260711_000002_create_suppressed_recipients" + } +} + +#[async_trait::async_trait] +impl MigrationTrait for Migration { + async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> { + manager + .get_connection() + .execute_unprepared( + r#" + CREATE TABLE IF NOT EXISTS suppressed_recipients ( + id SERIAL PRIMARY KEY, + email TEXT NOT NULL, + reason TEXT NOT NULL, + domain_id INTEGER REFERENCES email_domains(id) ON DELETE SET NULL, + detail TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now() + ); + + CREATE UNIQUE INDEX IF NOT EXISTS idx_suppressed_recipients_email + ON suppressed_recipients (email); + "#, + ) + .await?; + Ok(()) + } + + async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> { + manager + .get_connection() + .execute_unprepared("DROP TABLE IF EXISTS suppressed_recipients;") + .await?; + Ok(()) + } +} diff --git a/crates/temps-migrations/src/migration/m20260712_000001_add_email_retry_tracking.rs b/crates/temps-migrations/src/migration/m20260712_000001_add_email_retry_tracking.rs new file mode 100644 index 000000000..d4de0d819 --- /dev/null +++ b/crates/temps-migrations/src/migration/m20260712_000001_add_email_retry_tracking.rs @@ -0,0 +1,48 @@ +//! Retry/error-classification support for the email send path. Previously +//! any send failure went straight to "captured" — a transient SES throttle +//! or a Scaleway blip permanently dropped the email instead of retrying. +//! +//! `emails.provider_id` / `emails.attempt_count` record which provider +//! attempted the send and how many attempts it took, so the UI can show why +//! a send eventually succeeded or was captured. + +use sea_orm_migration::prelude::*; + +pub struct Migration; + +impl MigrationName for Migration { + fn name(&self) -> &str { + "m20260712_000001_add_email_retry_tracking" + } +} + +#[async_trait::async_trait] +impl MigrationTrait for Migration { + async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> { + manager + .get_connection() + .execute_unprepared( + r#" + ALTER TABLE emails + ADD COLUMN IF NOT EXISTS provider_id INTEGER REFERENCES email_providers(id) ON DELETE SET NULL, + ADD COLUMN IF NOT EXISTS attempt_count INTEGER NOT NULL DEFAULT 0; + "#, + ) + .await?; + Ok(()) + } + + async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> { + manager + .get_connection() + .execute_unprepared( + r#" + ALTER TABLE emails + DROP COLUMN IF EXISTS attempt_count, + DROP COLUMN IF EXISTS provider_id; + "#, + ) + .await?; + Ok(()) + } +} diff --git a/crates/temps-migrations/src/migration/mod.rs b/crates/temps-migrations/src/migration/mod.rs index adfef0b7e..d446295e2 100644 --- a/crates/temps-migrations/src/migration/mod.rs +++ b/crates/temps-migrations/src/migration/mod.rs @@ -146,7 +146,9 @@ mod m20260707_000002_add_external_services_container_name; mod m20260708_000001_add_node_id_to_monitoring_alert_rules; mod m20260711_000001_add_proxy_logs_stats_cagg; mod m20260711_000002_add_ip_geolocations_hosting_provider; +mod m20260711_000002_create_suppressed_recipients; mod m20260711_000003_add_visitor_non_crawler_partial_index; +mod m20260712_000001_add_email_retry_tracking; pub struct Migrator; @@ -298,7 +300,9 @@ impl MigratorTrait for Migrator { Box::new(m20260708_000001_add_node_id_to_monitoring_alert_rules::Migration), Box::new(m20260711_000001_add_proxy_logs_stats_cagg::Migration), Box::new(m20260711_000002_add_ip_geolocations_hosting_provider::Migration), + Box::new(m20260711_000002_create_suppressed_recipients::Migration), Box::new(m20260711_000003_add_visitor_non_crawler_partial_index::Migration), + Box::new(m20260712_000001_add_email_retry_tracking::Migration), ] } }