use serde_json::Value;
use tokio::sync::mpsc;
use crate::channels::{MessagingSubscription, MessagingSubscriptionError, MessagingTask};
use crate::messages::ListenerType;
use crate::types::NormalizedMessage;
use super::helpers::sender_in_list;
struct EmailFilter {
allowed_senders: Vec<String>,
subscribed_channels: Option<Vec<String>>,
}
impl EmailFilter {
fn new(allowed_senders: Vec<String>) -> Self {
Self {
allowed_senders,
subscribed_channels: None,
}
}
fn apply(&self, message: NormalizedMessage) -> Option<NormalizedMessage> {
let allowed = sender_in_list(&message, &self.allowed_senders)
&& self.subscribed_channels.as_ref().is_none_or(|channels| {
message
.metadata
.get("channel_id")
.and_then(Value::as_str)
.is_some_and(|channel| {
channels.iter().any(|selected| {
selected == channel
|| (selected.eq_ignore_ascii_case("INBOX") && channel.eq_ignore_ascii_case("INBOX"))
})
})
});
if allowed { Some(message) } else { None }
}
}
pub(super) fn subscribe(
subscription: MessagingSubscription,
mut rx: mpsc::Receiver<NormalizedMessage>,
tx: mpsc::Sender<NormalizedMessage>,
) -> Result<MessagingTask, MessagingSubscriptionError> {
subscription.validate(ListenerType::Email)?;
let mut policy = EmailFilter::new(subscription.allowed_senders);
policy.subscribed_channels = subscription.subscribed_channels;
Ok(tokio::spawn(async move {
while let Some(message) = tokio::select! { _ = tx.closed() => None, message = rx.recv() => message } {
if let Some(accepted) = policy.apply(message)
&& tx.send(accepted).await.is_err()
{
break;
}
}
Ok(())
}))
}
#[cfg(test)]
mod tests {
use super::super::helpers::normalize_update;
use serde_json::Value;
use super::*;
fn make_raw_email(from: &str, subject: &str, body: &str) -> Vec<u8> {
format!("From: {from}\r\nSubject: {subject}\r\nContent-Type: text/plain\r\n\r\n{body}").into_bytes()
}
#[test]
fn trigger_sender_policy_filters_before_emitting_normalized_updates() {
let raw = make_raw_email("Sender <Sender@Example.com>", "subject", "body");
assert!(
EmailFilter::new(Vec::new())
.apply(normalize_update(&raw).unwrap())
.is_some()
);
assert!(
EmailFilter::new(vec!["sender@example.com".to_owned()])
.apply(normalize_update(&raw).unwrap())
.is_some()
);
assert!(
EmailFilter::new(vec!["other@example.com".to_owned()])
.apply(normalize_update(&raw).unwrap())
.is_none()
);
let unidentified = b"Subject: subject\r\nContent-Type: text/plain\r\n\r\nbody";
assert!(
EmailFilter::new(vec!["sender@example.com".to_owned()])
.apply(normalize_update(unidentified).unwrap())
.is_none()
);
}
#[test]
fn normalize_message_extracts_body_and_subject() {
let raw = make_raw_email("Alice <alice@example.com>", "Hello world", "This is the body.");
let msg = normalize_update(&raw).expect("should parse");
assert_eq!(msg.body, "This is the body.");
assert_eq!(msg.metadata.get("subject").and_then(Value::as_str), Some("Hello world"));
}
#[test]
fn normalize_message_extracts_from_name_and_address() {
let raw = make_raw_email("Alice <alice@example.com>", "Hi", "Body text.");
let msg = normalize_update(&raw).expect("should parse");
assert_eq!(msg.metadata.get("from_name").and_then(Value::as_str), Some("Alice"));
assert_eq!(
msg.metadata.get("from_address").and_then(Value::as_str),
Some("alice@example.com")
);
}
#[test]
fn normalize_message_returns_none_when_body_and_subject_empty() {
let raw = b"From: sender@example.com\r\nContent-Type: text/plain\r\n\r\n";
let result = normalize_update(raw);
assert!(result.is_none(), "expected None for empty body+subject");
}
#[test]
fn normalize_message_accepts_subject_only() {
let raw = b"From: sender@example.com\r\nSubject: Alert!\r\nContent-Type: text/plain\r\n\r\n";
let msg = normalize_update(raw).expect("should parse when subject present");
assert_eq!(msg.metadata.get("subject").and_then(Value::as_str), Some("Alert!"));
assert_eq!(msg.body, "");
}
#[test]
fn normalize_message_accepts_body_only() {
let raw = b"From: sender@example.com\r\nContent-Type: text/plain\r\n\r\nJust a body.";
let msg = normalize_update(raw).expect("should parse when body present");
assert_eq!(msg.body, "Just a body.");
assert_eq!(msg.metadata.get("subject").and_then(Value::as_str), Some(""));
}
#[test]
fn normalize_message_includes_channel_type_in_metadata() {
let raw = make_raw_email("x@example.com", "Test", "Body.");
let msg = normalize_update(&raw).expect("should parse");
assert!(
msg.metadata.get("channel_type").is_some(),
"channel_type must be present in metadata"
);
}
#[test]
fn normalize_message_missing_from_address_is_null_in_metadata() {
let raw = b"Subject: No From\r\nContent-Type: text/plain\r\n\r\nBody here.";
let msg = normalize_update(raw).expect("should parse");
assert_eq!(msg.metadata.get("from_address"), Some(&Value::Null));
}
fn passes_sender_filter(from_address: &str, allowed_senders: &Option<Vec<String>>) -> bool {
if let Some(allowed) = allowed_senders {
if !allowed.is_empty() {
return allowed.iter().any(|a| a.eq_ignore_ascii_case(from_address));
}
}
true
}
#[test]
fn allowed_senders_none_passes_all() {
assert!(passes_sender_filter("anyone@example.com", &None));
}
#[test]
fn allowed_senders_empty_vec_passes_all() {
assert!(passes_sender_filter("anyone@example.com", &Some(vec![])));
}
#[test]
fn allowed_senders_exact_match_passes() {
let allowed = Some(vec!["alice@example.com".to_string()]);
assert!(passes_sender_filter("alice@example.com", &allowed));
}
#[test]
fn allowed_senders_case_insensitive_match_passes() {
let allowed = Some(vec!["Alice@Example.COM".to_string()]);
assert!(passes_sender_filter("alice@example.com", &allowed));
}
#[test]
fn allowed_senders_non_matching_sender_blocked() {
let allowed = Some(vec!["alice@example.com".to_string()]);
assert!(!passes_sender_filter("eve@evil.com", &allowed));
}
#[test]
fn allowed_senders_multiple_entries_any_match_passes() {
let allowed = Some(vec!["alice@example.com".to_string(), "bob@example.com".to_string()]);
assert!(passes_sender_filter("bob@example.com", &allowed));
assert!(!passes_sender_filter("eve@evil.com", &allowed));
}
fn uid_should_process(uid: Option<u32>, last_uid: u32) -> bool {
matches!(uid, Some(u) if u > last_uid)
}
#[test]
fn uid_guard_skips_equal_uid() {
assert!(!uid_should_process(Some(42), 42));
}
#[test]
fn uid_guard_skips_lower_uid() {
assert!(!uid_should_process(Some(41), 42));
}
#[test]
fn uid_guard_passes_strictly_greater_uid() {
assert!(uid_should_process(Some(43), 42));
}
#[test]
fn uid_guard_skips_none_uid() {
assert!(!uid_should_process(None, 42));
}
#[test]
fn uid_guard_passes_when_last_uid_is_zero() {
assert!(uid_should_process(Some(1), 0));
}
}