crates/features/src/mailflow/cache.rs

@@ -0,0 +1,53 @@
+/*
+ * SPDX-FileCopyrightText: 2026 Coffey Labs
+ *
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+//! The compiled rules, kept per node so a message doesn't read the store.
+//! A change made on this node applies at once; one made on another node
+//! within [`TTL`], when the copy here is next refreshed.
+
+use super::{engine::Compiled, rules};
+use std::{
+    sync::{Arc, RwLock},
+    time::{Duration, Instant},
+};
+use store::Store;
+
+/// How long a node keeps its copy before reading the rules again.
+pub const TTL: Duration = Duration::from_secs(30);
+
+static CACHE: RwLock<Option<(Instant, Arc<Compiled>)>> = RwLock::new(None);
+
+/// Forgets the copy, so the next message reads the rules again.
+pub fn invalidate() {
+    if let Ok(mut cache) = CACHE.write() {
+        *cache = None;
+    }
+}
+
+/// The enabled rules, compiled. A rule that no longer compiles is left out
+/// and reported, once per refresh.
+pub async fn compiled(data: &Store) -> trc::Result<Arc<Compiled>> {
+    if let Ok(cache) = CACHE.read()
+        && let Some((at, compiled)) = cache.as_ref()
+        && at.elapsed() < TTL
+    {
+        return Ok(compiled.clone());
+    }
+    let (compiled, skipped) = Compiled::new(&rules::all(data).await?);
+    for (id, reason) in skipped {
+        trc::event!(
+            Store(trc::StoreEvent::DataCorruption),
+            Id = u64::from(id),
+            Reason = reason,
+            Details = "Mail rule skipped: it no longer compiles"
+        );
+    }
+    let compiled = Arc::new(compiled);
+    if let Ok(mut cache) = CACHE.write() {
+        *cache = Some((Instant::now(), compiled.clone()));
+    }
+    Ok(compiled)
+}

crates/features/src/mailflow/engine.rs

@@ -0,0 +1,695 @@
+/*
+ * SPDX-FileCopyrightText: 2026 Coffey Labs
+ *
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+//! Evaluating rules against a message (§2.1–§2.4). Rules are compiled once,
+//! when they change: word lists become automata, patterns regexes. A message
+//! is then checked against every enabled rule in order; each detector runs
+//! at most once per message, and only when some rule asks for it.
+//!
+//! Pure: the caller parses the message, extracts attachment text
+//! ([`super::extract`]) and knows the sender's groups and tenant. What comes
+//! back is which rules matched, with each detector's count, and what DLP
+//! decided; the matched text itself never leaves here (§2.7).
+
+use super::{
+    detectors::{self, Findings},
+    extract::Extracted,
+    rules::{Action, Condition, Direction, Kind, Rule},
+    words::{Pattern, WordList},
+};
+use ahash::AHashMap;
+use std::borrow::Cow;
+
+/// Who sent a message, and to whom.
+#[derive(Debug, Clone, Default)]
+pub struct Envelope<'a> {
+    /// Outgoing (an authenticated sender) or incoming.
+    pub outgoing: bool,
+    pub sender: &'a str,
+    pub sender_groups: &'a [u32],
+    pub sender_tenant: Option<u32>,
+    pub recipients: Vec<Recipient<'a>>,
+}
+
+#[derive(Debug, Clone, Default)]
+pub struct Recipient<'a> {
+    pub address: &'a str,
+    /// At a domain this server hosts.
+    pub local: bool,
+    pub groups: &'a [u32],
+}
+
+#[derive(Debug, Clone)]
+pub struct Attachment<'a> {
+    pub name: Option<&'a str>,
+    /// Declared type, or detected where the caller knows better.
+    pub content_type: &'a str,
+    pub size: u64,
+    pub extracted: Extracted,
+}
+
+/// What rules look at.
+#[derive(Debug, Clone, Default)]
+pub struct Content<'a> {
+    pub subject: &'a str,
+    /// Each text and HTML part, as text.
+    pub bodies: Vec<Cow<'a, str>>,
+    pub headers: Vec<(&'a str, &'a str)>,
+    pub attachments: Vec<Attachment<'a>>,
+    pub size: u64,
+    /// Text past the inspection limit wasn't read.
+    pub truncated: bool,
+}
+
+impl Content<'_> {
+    fn texts(&self) -> impl Iterator<Item = &str> {
+        std::iter::once(self.subject)
+            .chain(self.bodies.iter().map(|b| b.as_ref()))
+            .chain(self.attachments.iter().filter_map(|a| match &a.extracted {
+                Extracted::Text(text) => Some(text.as_str()),
+                _ => None,
+            }))
+    }
+
+    fn cant_be_inspected(&self) -> bool {
+        self.truncated
+            || self
+                .attachments
+                .iter()
+                .any(|a| matches!(a.extracted, Extracted::NotInspectable(_)))
+    }
+}
+
+enum Check {
+    Plain(Condition),
+    Words(WordList, u32),
+    Pattern(Pattern, u32),
+    Header {
+        name: String,
+        contains: Option<String>,
+        matches: Option<Pattern>,
+    },
+    AttachmentName(Pattern),
+}
+
+struct CompiledRule {
+    rule: Rule,
+    conditions: Vec<Check>,
+    exceptions: Vec<Check>,
+}
+
+/// The enabled rules, ready to run.
+pub struct Compiled {
+    rules: Vec<CompiledRule>,
+}
+
+/// A rule reference, for notices and the audit record.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct RuleRef {
+    pub id: u32,
+    pub name: String,
+    pub notice: String,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct Match {
+    pub rule_id: u32,
+    pub name: String,
+    pub kind: Kind,
+    pub actions: Vec<Action>,
+    /// Each detector (or `words`, `pattern`) that counted, and its count.
+    pub counts: Vec<(String, usize)>,
+}
+
+#[derive(Debug, Default)]
+pub struct Outcome {
+    pub matched: Vec<Match>,
+    pub blocks: Vec<RuleRef>,
+    pub holds: Vec<(RuleRef, bool)>,
+    pub warns: Vec<RuleRef>,
+}
+
+/// What DLP decided, strictest first (§2.4).
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum Decision {
+    Pass,
+    Block(Vec<RuleRef>),
+    Hold {
+        rules: Vec<RuleRef>,
+        notify_sender: bool,
+    },
+    Warn(Vec<RuleRef>),
+}
+
+impl Outcome {
+    /// Block beats hold beats warn. An override (§2.5) answers the warnings
+    /// only: a block or hold still applies.
+    pub fn decision(&self, overridden: bool) -> Decision {
+        if !self.blocks.is_empty() {
+            Decision::Block(self.blocks.clone())
+        } else if !self.holds.is_empty() {
+            Decision::Hold {
+                rules: self.holds.iter().map(|(r, _)| r.clone()).collect(),
+                notify_sender: self.holds.iter().any(|(_, notify)| *notify),
+            }
+        } else if !self.warns.is_empty() && !overridden {
+            Decision::Warn(self.warns.clone())
+        } else {
+            Decision::Pass
+        }
+    }
+}
+
+fn compile_check(condition: &Condition) -> Result<Check, String> {
+    Ok(match condition {
+        Condition::Words { words, at_least } => Check::Words(WordList::new(words)?, *at_least),
+        Condition::Pattern { pattern, at_least } => {
+            Check::Pattern(Pattern::new(pattern)?, *at_least)
+        }
+        Condition::Header {
+            name,
+            contains,
+            matches,
+        } => Check::Header {
+            name: name.to_ascii_lowercase(),
+            contains: contains.as_ref().map(|c| c.to_lowercase()),
+            matches: matches.as_deref().map(Pattern::new).transpose()?,
+        },
+        Condition::AttachmentName { pattern } => Check::AttachmentName(Pattern::new(pattern)?),
+        other => Check::Plain(other.clone()),
+    })
+}
+
+impl Compiled {
+    /// Compiles the enabled rules; one that no longer compiles (a detector
+    /// renamed since it was saved) is skipped and named in the second list.
+    pub fn new(rules: &[Rule]) -> (Self, Vec<(u32, String)>) {
+        let mut compiled = Vec::new();
+        let mut skipped = Vec::new();
+        for rule in rules.iter().filter(|r| r.enabled) {
+            let result = rule.validate().map_err(|e| e.reason).and_then(|_| {
+                Ok(CompiledRule {
+                    rule: rule.clone(),
+                    conditions: rule
+                        .conditions
+                        .iter()
+                        .map(compile_check)
+                        .collect::<Result<_, _>>()?,
+                    exceptions: rule
+                        .exceptions
+                        .iter()
+                        .map(compile_check)
+                        .collect::<Result<_, _>>()?,
+                })
+            });
+            match result {
+                Ok(c) => compiled.push(c),
+                Err(reason) => skipped.push((rule.id, reason)),
+            }
+        }
+        compiled.sort_by_key(|c| (c.rule.priority, c.rule.id));
+        (Self { rules: compiled }, skipped)
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.rules.is_empty()
+    }
+
+    /// Whether any rule could apply to mail going this way, so a caller can
+    /// skip parsing when none can.
+    pub fn applies_to(&self, outgoing: bool) -> bool {
+        self.rules
+            .iter()
+            .any(|c| direction_matches(c.rule.direction, outgoing))
+    }
+
+    pub fn evaluate(&self, envelope: &Envelope<'_>, content: &Content<'_>) -> Outcome {
+        let mut state = State {
+            content,
+            detected: AHashMap::new(),
+        };
+        let mut outcome = Outcome::default();
+        for compiled in &self.rules {
+            let rule = &compiled.rule;
+            if !direction_matches(rule.direction, envelope.outgoing) {
+                continue;
+            }
+            let mut counts = Vec::new();
+            let all_match = compiled
+                .conditions
+                .iter()
+                .all(|check| state.check(check, envelope, &mut counts));
+            if !all_match {
+                continue;
+            }
+            let mut ignored = Vec::new();
+            if compiled
+                .exceptions
+                .iter()
+                .any(|check| state.check(check, envelope, &mut ignored))
+            {
+                continue;
+            }
+            for action in &rule.actions {
+                let reference = |notice: &str| RuleRef {
+                    id: rule.id,
+                    name: rule.name.clone(),
+                    notice: notice.to_string(),
+                };
+                match action {
+                    Action::Block { notice } => outcome.blocks.push(reference(notice)),
+                    Action::Hold {
+                        notice,
+                        notify_sender,
+                    } => outcome.holds.push((reference(notice), *notify_sender)),
+                    Action::Warn { notice } => outcome.warns.push(reference(notice)),
+                    _ => {}
+                }
+            }
+            outcome.matched.push(Match {
+                rule_id: rule.id,
+                name: rule.name.clone(),
+                kind: rule.kind,
+                actions: rule.actions.clone(),
+                counts,
+            });
+            if rule.stop_processing {
+                break;
+            }
+        }
+        outcome
+    }
+}
+
+fn direction_matches(direction: Direction, outgoing: bool) -> bool {
+    match direction {
+        Direction::Any => true,
+        Direction::Outgoing => outgoing,
+        Direction::Incoming => !outgoing,
+    }
+}
+
+fn domain_of(address: &str) -> &str {
+    address.rsplit_once('@').map_or("", |(_, d)| d)
+}
+
+fn in_list(value: &str, list: &[String]) -> bool {
+    list.iter().any(|v| v.eq_ignore_ascii_case(value))
+}
+
+struct State<'c, 'a> {
+    content: &'c Content<'a>,
+    /// Each detector's count, run once per message.
+    detected: AHashMap<&'static str, usize>,
+}
+
+impl State<'_, '_> {
+    fn detector_count(&mut self, id: &str) -> usize {
+        let Some(detector) = detectors::by_id(id) else {
+            return 0;
+        };
+        if let Some(count) = self.detected.get(detector.id) {
+            return *count;
+        }
+        let mut findings = Findings::default();
+        for text in self.content.texts() {
+            detector.find(text, &mut findings);
+        }
+        self.detected.insert(detector.id, findings.len());
+        findings.len()
+    }
+
+    fn check(
+        &mut self,
+        check: &Check,
+        envelope: &Envelope<'_>,
+        counts: &mut Vec<(String, usize)>,
+    ) -> bool {
+        let content = self.content;
+        match check {
+            Check::Words(list, at_least) => {
+                let n: usize = content.texts().map(|t| list.count(t)).sum();
+                counts.push(("words".into(), n));
+                n >= *at_least as usize
+            }
+            Check::Pattern(pattern, at_least) => {
+                let n: usize = content.texts().map(|t| pattern.count(t)).sum();
+                counts.push(("pattern".into(), n));
+                n >= *at_least as usize
+            }
+            Check::Header {
+                name,
+                contains,
+                matches,
+            } => content
+                .headers
+                .iter()
+                .filter(|(n, _)| n.eq_ignore_ascii_case(name))
+                .any(|(_, value)| match (contains, matches) {
+                    (Some(needle), _) => value.to_lowercase().contains(needle.as_str()),
+                    (_, Some(pattern)) => pattern.count(value) > 0,
+                    _ => true,
+                }),
+            Check::AttachmentName(pattern) => content
+                .attachments
+                .iter()
+                .any(|a| a.name.is_some_and(|n| pattern.count(n) > 0)),
+            Check::Plain(condition) => match condition {
+                Condition::SenderAddress { addresses } => in_list(envelope.sender, addresses),
+                Condition::SenderDomain { domains } => in_list(domain_of(envelope.sender), domains),
+                Condition::SenderGroup { groups } => {
+                    envelope.sender_groups.iter().any(|g| groups.contains(g))
+                }
+                Condition::SenderTenant { tenants } => {
+                    envelope.sender_tenant.is_some_and(|t| tenants.contains(&t))
+                }
+                Condition::RecipientAddress { addresses } => envelope
+                    .recipients
+                    .iter()
+                    .any(|r| in_list(r.address, addresses)),
+                Condition::RecipientDomain { domains } => envelope
+                    .recipients
+                    .iter()
+                    .any(|r| in_list(domain_of(r.address), domains)),
+                Condition::RecipientGroup { groups } => envelope
+                    .recipients
+                    .iter()
+                    .any(|r| r.groups.iter().any(|g| groups.contains(g))),
+                Condition::RecipientOutside => envelope.recipients.iter().any(|r| !r.local),
+                Condition::AttachmentType { types } => content.attachments.iter().any(|a| {
+                    let ct = a.content_type.to_ascii_lowercase();
+                    types
+                        .iter()
+                        .any(|t| ct.starts_with(&t.to_ascii_lowercase()))
+                }),
+                Condition::AttachmentExtension { extensions } => {
+                    content.attachments.iter().any(|a| {
+                        a.name
+                            .and_then(|n| n.rsplit_once('.'))
+                            .is_some_and(|(_, ext)| {
+                                extensions
+                                    .iter()
+                                    .any(|e| e.trim_start_matches('.').eq_ignore_ascii_case(ext))
+                            })
+                    })
+                }
+                Condition::AttachmentSizeOver { bytes } => {
+                    content.attachments.iter().any(|a| a.size > *bytes)
+                }
+                Condition::AttachmentCountOver { count } => {
+                    content.attachments.len() > *count as usize
+                }
+                Condition::CantBeInspected => content.cant_be_inspected(),
+                Condition::MessageSizeOver { bytes } => content.size > *bytes,
+                Condition::Detected { detectors } => {
+                    let mut any = false;
+                    for d in detectors {
+                        let n = self.detector_count(&d.id);
+                        counts.push((d.id.clone(), n));
+                        any |= n >= d.at_least as usize;
+                    }
+                    any
+                }
+                // Compiled into their own checks
+                Condition::Words { .. }
+                | Condition::Pattern { .. }
+                | Condition::Header { .. }
+                | Condition::AttachmentName { .. } => false,
+            },
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::mailflow::{
+        extract::Why,
+        rules::{DetectorMin, Position},
+    };
+
+    fn rule(id: u32, kind: Kind, conditions: Vec<Condition>, action: Action) -> Rule {
+        Rule {
+            id,
+            name: format!("rule {id}"),
+            description: String::new(),
+            kind,
+            enabled: true,
+            priority: id as i32,
+            direction: if kind == Kind::Dlp {
+                Direction::Outgoing
+            } else {
+                Direction::Any
+            },
+            conditions,
+            exceptions: vec![],
+            actions: vec![action],
+            stop_processing: false,
+            created_by: String::new(),
+            created_at: 0,
+            updated_at: 0,
+        }
+    }
+
+    fn envelope(outside: bool) -> Envelope<'static> {
+        Envelope {
+            outgoing: true,
+            sender: "[email protected]",
+            sender_groups: &[7],
+            sender_tenant: None,
+            recipients: vec![Recipient {
+                address: if outside {
+                    "[email protected]"
+                } else {
+                    "[email protected]"
+                },
+                local: !outside,
+                groups: &[],
+            }],
+        }
+    }
+
+    fn cards(n: usize) -> Content<'static> {
+        let body: String = [
+            "4242 4242 4242 4242",
+            "5555-5555-5555-4444",
+            "378282246310005",
+            "6011111111111117",
+            "3566002020360505",
+        ]
+        .iter()
+        .take(n)
+        .map(|c| format!("card {c}\n"))
+        .collect();
+        Content {
+            subject: "Numbers",
+            bodies: vec![body.into()],
+            ..Default::default()
+        }
+    }
+
+    fn five_cards_outside(action: Action) -> Rule {
+        rule(
+            1,
+            Kind::Dlp,
+            vec![
+                Condition::RecipientOutside,
+                Condition::Detected {
+                    detectors: vec![DetectorMin {
+                        id: "payment-card".into(),
+                        at_least: 5,
+                    }],
+                },
+            ],
+            action,
+        )
+    }
+
+    #[test]
+    fn detector_threshold_and_recipients() {
+        let (rules, skipped) = Compiled::new(&[five_cards_outside(Action::Hold {
+            notice: "Held".into(),
+            notify_sender: true,
+        })]);
+        assert!(skipped.is_empty());
+        let outcome = rules.evaluate(&envelope(true), &cards(5));
+        assert_eq!(
+            outcome.matched[0].counts,
+            vec![("payment-card".to_string(), 5)]
+        );
+        assert!(matches!(
+            outcome.decision(false),
+            Decision::Hold {
+                notify_sender: true,
+                ..
+            }
+        ));
+        // Four cards, or everyone inside: nothing
+        assert_eq!(
+            rules.evaluate(&envelope(true), &cards(4)).decision(false),
+            Decision::Pass
+        );
+        assert_eq!(
+            rules.evaluate(&envelope(false), &cards(5)).decision(false),
+            Decision::Pass
+        );
+    }
+
+    #[test]
+    fn strictest_wins_and_override_answers_warnings_only() {
+        let warn = five_cards_outside(Action::Warn {
+            notice: "Sure?".into(),
+        });
+        let mut block = five_cards_outside(Action::Block {
+            notice: "No".into(),
+        });
+        block.id = 2;
+        let (rules, _) = Compiled::new(&[warn.clone(), block]);
+        let outcome = rules.evaluate(&envelope(true), &cards(5));
+        assert!(matches!(outcome.decision(true), Decision::Block(_)));
+        let (rules, _) = Compiled::new(&[warn]);
+        let outcome = rules.evaluate(&envelope(true), &cards(5));
+        assert!(matches!(outcome.decision(false), Decision::Warn(ref w) if w[0].notice == "Sure?"));
+        assert_eq!(outcome.decision(true), Decision::Pass);
+    }
+
+    #[test]
+    fn exceptions_order_and_stop_processing() {
+        let disclaimer = |id| {
+            rule(
+                id,
+                Kind::Transport,
+                vec![Condition::RecipientOutside],
+                Action::AddDisclaimer {
+                    text: "t".into(),
+                    html: None,
+                    position: Position::Bottom,
+                },
+            )
+        };
+        let mut first = disclaimer(1);
+        first.stop_processing = true;
+        let (rules, _) = Compiled::new(&[disclaimer(2), first.clone()]);
+        let outcome = rules.evaluate(&envelope(true), &cards(0));
+        assert_eq!(
+            outcome
+                .matched
+                .iter()
+                .map(|m| m.rule_id)
+                .collect::<Vec<_>>(),
+            vec![1]
+        );
+
+        first.stop_processing = false;
+        first.exceptions = vec![Condition::SenderGroup { groups: vec![7] }];
+        let (rules, _) = Compiled::new(&[disclaimer(2), first]);
+        let outcome = rules.evaluate(&envelope(true), &cards(0));
+        assert_eq!(
+            outcome
+                .matched
+                .iter()
+                .map(|m| m.rule_id)
+                .collect::<Vec<_>>(),
+            vec![2]
+        );
+    }
+
+    #[test]
+    fn content_conditions() {
+        let content = Content {
+            subject: "Project Falcon",
+            bodies: vec!["see attached".into()],
+            headers: vec![("X-Class", "Internal only")],
+            attachments: vec![
+                Attachment {
+                    name: Some("plan.docx"),
+                    content_type: "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
+                    size: 40_000,
+                    extracted: Extracted::Text("IBAN GB29 NWBK 6016 1331 9268 19".into()),
+                },
+                Attachment {
+                    name: Some("scan.pdf"),
+                    content_type: "application/pdf",
+                    size: 900_000,
+                    extracted: Extracted::NotInspectable(Why::Pdf),
+                },
+            ],
+            size: 1_000_000,
+            truncated: false,
+        };
+        let block = || Action::Block { notice: "n".into() };
+        let checks = [
+            (
+                Condition::Words {
+                    words: vec!["project falcon".into()],
+                    at_least: 1,
+                },
+                true,
+            ),
+            (
+                Condition::Header {
+                    name: "x-class".into(),
+                    contains: Some("internal".into()),
+                    matches: None,
+                },
+                true,
+            ),
+            (
+                Condition::AttachmentExtension {
+                    extensions: vec![".PDF".into()],
+                },
+                true,
+            ),
+            (
+                Condition::AttachmentType {
+                    types: vec!["image/".into()],
+                },
+                false,
+            ),
+            (Condition::AttachmentSizeOver { bytes: 500_000 }, true),
+            (Condition::AttachmentCountOver { count: 2 }, false),
+            (Condition::CantBeInspected, true),
+            (Condition::MessageSizeOver { bytes: 2_000_000 }, false),
+            (
+                Condition::Detected {
+                    detectors: vec![DetectorMin {
+                        id: "iban".into(),
+                        at_least: 1,
+                    }],
+                },
+                true,
+            ),
+            (
+                Condition::SenderDomain {
+                    domains: vec!["EXAMPLE.com".into()],
+                },
+                true,
+            ),
+        ];
+        for (condition, expected) in checks {
+            let (rules, skipped) =
+                Compiled::new(&[rule(1, Kind::Dlp, vec![condition.clone()], block())]);
+            assert!(skipped.is_empty(), "{condition:?}");
+            let matched = !rules.evaluate(&envelope(true), &content).matched.is_empty();
+            assert_eq!(matched, expected, "{condition:?}");
+        }
+    }
+
+    #[test]
+    fn direction_and_disabled_rules() {
+        let mut r = five_cards_outside(Action::Block { notice: "n".into() });
+        let (rules, _) = Compiled::new(std::slice::from_ref(&r));
+        assert!(rules.applies_to(true) && !rules.applies_to(false));
+        let mut incoming = envelope(true);
+        incoming.outgoing = false;
+        assert_eq!(
+            rules.evaluate(&incoming, &cards(5)).decision(false),
+            Decision::Pass
+        );
+        r.enabled = false;
+        assert!(Compiled::new(&[r]).0.is_empty());
+    }
+}

crates/features/src/mailflow/mod.rs

@@ -6,18 +6,24 @@
 
 //! Data loss prevention and mail flow rules (dlp-and-mail-flow-rules spec).
 //!
-//! Pure functions over text and attachment bytes, so everything here is
+//! Mostly pure functions over text and attachment bytes, unit-tested
-//! unit-tested without a server:
+//! without a server:
 //!
 //! - [`detectors`]: find identifiers in text (payment cards, IBANs,
 //!   national ID numbers, keys), each by its published format and check
 //!   (§2.3);
 //! - [`words`]: an organization's own word lists and patterns;
-//! - [`extract`]: the text of an attachment, or why it can't be read.
+//! - [`extract`]: the text of an attachment, or why it can't be read;
+//! - [`rules`]: what a rule is, its checks, and where rules are kept;
+//! - [`engine`]: rules compiled and run against a message;
+//! - [`cache`]: each node's compiled copy.
 //!
 //! Nothing here writes what it finds anywhere: callers get counts, and the
 //! matched text never leaves the evaluation (§2.7).
 
+pub mod cache;
 pub mod detectors;
+pub mod engine;
 pub mod extract;
+pub mod rules;
 pub mod words;

crates/features/src/mailflow/rules.rs

@@ -0,0 +1,717 @@
+/*
+ * SPDX-FileCopyrightText: 2026 Coffey Labs
+ *
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+//! Mail flow rules and DLP rules (dlp-and-mail-flow-rules spec, §2.2–§2.4):
+//! what a rule is, what makes one valid, and where it's kept.
+//!
+//! Kept in the fork's subspace (`store::SUBSPACE_INBUXA`), never in the
+//! registry, so an upstream schema import never touches them. Every key
+//! starts with `R`, then one byte for the kind:
+//!
+//! - `r` + rule id (u32): the rule, as JSON.
+//!
+//! Numbers are big-endian. There are few rules, so they're read whole.
+
+use super::{detectors, words};
+use serde::{Deserialize as SerdeDeserialize, Serialize as SerdeSerialize, de::DeserializeOwned};
+use store::{
+    Deserialize, IterateParams, SUBSPACE_INBUXA, Serialize, Store, ValueKey,
+    write::{AnyClass, BatchBuilder, ValueClass, assert::AssertValue},
+};
+use trc::AddContext;
+
+const FEATURE: u8 = b'R';
+const KIND_RULE: u8 = b'r';
+const CREATE_ATTEMPTS: usize = 5;
+
+/// Longest text a rule may carry (a notice, a disclaimer), in bytes.
+const MAX_TEXT: usize = 16 * 1024;
+/// Most entries in one list (words, addresses, domains).
+const MAX_LIST: usize = 5_000;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, SerdeSerialize, SerdeDeserialize)]
+#[serde(rename_all = "camelCase")]
+pub enum Kind {
+    Dlp,
+    Transport,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, SerdeSerialize, SerdeDeserialize)]
+#[serde(rename_all = "camelCase")]
+pub enum Direction {
+    /// Mail an authenticated sender submits, over SMTP or JMAP.
+    Outgoing,
+    /// Everything else the server accepts.
+    Incoming,
+    Any,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, SerdeSerialize, SerdeDeserialize)]
+#[serde(rename_all = "camelCase")]
+pub enum Position {
+    Top,
+    Bottom,
+}
+
+fn one() -> u32 {
+    1
+}
+
+/// A detector and the least it must find.
+#[derive(Debug, Clone, PartialEq, Eq, SerdeSerialize, SerdeDeserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct DetectorMin {
+    pub id: String,
+    #[serde(default = "one")]
+    pub at_least: u32,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, SerdeSerialize, SerdeDeserialize)]
+#[serde(
+    tag = "type",
+    rename_all = "camelCase",
+    rename_all_fields = "camelCase"
+)]
+pub enum Condition {
+    SenderAddress {
+        addresses: Vec<String>,
+    },
+    SenderDomain {
+        domains: Vec<String>,
+    },
+    SenderGroup {
+        groups: Vec<u32>,
+    },
+    SenderTenant {
+        tenants: Vec<u32>,
+    },
+    /// Any recipient is one of these.
+    RecipientAddress {
+        addresses: Vec<String>,
+    },
+    RecipientDomain {
+        domains: Vec<String>,
+    },
+    RecipientGroup {
+        groups: Vec<u32>,
+    },
+    /// Any recipient isn't at a domain this server hosts.
+    RecipientOutside,
+    /// Words or phrases in the subject, body or readable attachments.
+    Words {
+        words: Vec<String>,
+        #[serde(default = "one")]
+        at_least: u32,
+    },
+    /// The organization's regular expression, in the same places.
+    Pattern {
+        pattern: String,
+        #[serde(default = "one")]
+        at_least: u32,
+    },
+    /// A header exists, or its value contains or matches.
+    Header {
+        name: String,
+        #[serde(default)]
+        contains: Option<String>,
+        #[serde(default)]
+        matches: Option<String>,
+    },
+    /// An attachment's declared or detected type starts with one of these.
+    AttachmentType {
+        types: Vec<String>,
+    },
+    AttachmentExtension {
+        extensions: Vec<String>,
+    },
+    AttachmentName {
+        pattern: String,
+    },
+    AttachmentSizeOver {
+        bytes: u64,
+    },
+    AttachmentCountOver {
+        count: u32,
+    },
+    /// An attachment is encrypted, a PDF, a legacy Office file, an archive
+    /// inside an archive, or past the inspection limit.
+    CantBeInspected,
+    MessageSizeOver {
+        bytes: u64,
+    },
+    /// Any of these detectors finds at least its minimum (DLP rules only).
+    Detected {
+        detectors: Vec<DetectorMin>,
+    },
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, SerdeSerialize, SerdeDeserialize)]
+#[serde(
+    tag = "type",
+    rename_all = "camelCase",
+    rename_all_fields = "camelCase"
+)]
+pub enum Action {
+    // Transport actions
+    AddDisclaimer {
+        text: String,
+        #[serde(default)]
+        html: Option<String>,
+        position: Position,
+    },
+    AddHeader {
+        name: String,
+        value: String,
+    },
+    RemoveHeader {
+        name: String,
+    },
+    PrefixSubject {
+        text: String,
+    },
+    AddRecipient {
+        address: String,
+    },
+    Redirect {
+        addresses: Vec<String>,
+    },
+    Refuse {
+        text: String,
+    },
+    Route {
+        queue: String,
+    },
+    // DLP actions
+    Block {
+        notice: String,
+    },
+    Warn {
+        notice: String,
+    },
+    Hold {
+        notice: String,
+        #[serde(default)]
+        notify_sender: bool,
+    },
+}
+
+impl Action {
+    pub fn is_dlp(&self) -> bool {
+        matches!(
+            self,
+            Action::Block { .. } | Action::Warn { .. } | Action::Hold { .. }
+        )
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, SerdeSerialize, SerdeDeserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct Rule {
+    #[serde(default)]
+    pub id: u32,
+    pub name: String,
+    #[serde(default)]
+    pub description: String,
+    pub kind: Kind,
+    #[serde(default = "enabled")]
+    pub enabled: bool,
+    #[serde(default)]
+    pub priority: i32,
+    pub direction: Direction,
+    #[serde(default)]
+    pub conditions: Vec<Condition>,
+    #[serde(default)]
+    pub exceptions: Vec<Condition>,
+    pub actions: Vec<Action>,
+    #[serde(default)]
+    pub stop_processing: bool,
+    #[serde(default)]
+    pub created_by: String,
+    #[serde(default)]
+    pub created_at: u64,
+    #[serde(default)]
+    pub updated_at: u64,
+}
+
+fn enabled() -> bool {
+    true
+}
+
+/// Why a rule can't be saved: the property at fault, and a sentence.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct Invalid {
+    pub property: &'static str,
+    pub reason: String,
+}
+
+fn invalid(property: &'static str, reason: impl Into<String>) -> Invalid {
+    Invalid {
+        property,
+        reason: reason.into(),
+    }
+}
+
+impl Rule {
+    /// Everything that can be checked without the rest of the server: the
+    /// shape (§2.2, §2.4), the detectors, word lists and patterns.
+    pub fn validate(&self) -> Result<(), Invalid> {
+        if self.name.trim().is_empty() {
+            return Err(invalid("name", "A rule needs a name."));
+        }
+        if self.name.len() > 200 || self.description.len() > MAX_TEXT {
+            return Err(invalid("name", "The name or description is too long."));
+        }
+        if self.actions.is_empty() {
+            return Err(invalid("actions", "A rule needs something to do."));
+        }
+        let dlp_actions = self.actions.iter().filter(|a| a.is_dlp()).count();
+        match self.kind {
+            Kind::Dlp => {
+                if self.direction != Direction::Outgoing {
+                    return Err(invalid("direction", "DLP rules check outgoing mail only."));
+                }
+                if dlp_actions != 1 || self.actions.len() != 1 {
+                    return Err(invalid(
+                        "actions",
+                        "A DLP rule has exactly one action: block, warn or hold.",
+                    ));
+                }
+            }
+            Kind::Transport => {
+                if dlp_actions > 0 {
+                    return Err(invalid(
+                        "actions",
+                        "Block, warn and hold belong to DLP rules.",
+                    ));
+                }
+                if self
+                    .conditions
+                    .iter()
+                    .chain(&self.exceptions)
+                    .any(|c| matches!(c, Condition::Detected { .. }))
+                {
+                    return Err(invalid("conditions", "Detectors belong to DLP rules."));
+                }
+            }
+        }
+        for (property, list) in [
+            ("conditions", &self.conditions),
+            ("exceptions", &self.exceptions),
+        ] {
+            for condition in list {
+                validate_condition(condition).map_err(|reason| invalid(property, reason))?;
+            }
+        }
+        for action in &self.actions {
+            validate_action(action).map_err(|reason| invalid("actions", reason))?;
+        }
+        Ok(())
+    }
+}
+
+fn nonempty_list<T>(list: &[T], what: &str) -> Result<(), String> {
+    if list.is_empty() {
+        Err(format!("The {what} list is empty."))
+    } else if list.len() > MAX_LIST {
+        Err(format!(
+            "The {what} list is longer than {MAX_LIST} entries."
+        ))
+    } else {
+        Ok(())
+    }
+}
+
+fn header_name(name: &str) -> Result<(), String> {
+    if !name.is_empty()
+        && name.len() <= 100
+        && name.bytes().all(|b| b.is_ascii_graphic() && b != b':')
+    {
+        Ok(())
+    } else {
+        Err(format!("\"{name}\" isn't a header name."))
+    }
+}
+
+fn text(value: &str, what: &str) -> Result<(), String> {
+    if value.trim().is_empty() {
+        Err(format!("The {what} is empty."))
+    } else if value.len() > MAX_TEXT {
+        Err(format!("The {what} is longer than {MAX_TEXT} bytes."))
+    } else {
+        Ok(())
+    }
+}
+
+fn validate_condition(condition: &Condition) -> Result<(), String> {
+    match condition {
+        Condition::SenderAddress { addresses } | Condition::RecipientAddress { addresses } => {
+            nonempty_list(addresses, "address")
+        }
+        Condition::SenderDomain { domains } | Condition::RecipientDomain { domains } => {
+            nonempty_list(domains, "domain")
+        }
+        Condition::SenderGroup { groups } | Condition::RecipientGroup { groups } => {
+            nonempty_list(groups, "group")
+        }
+        Condition::SenderTenant { tenants } => nonempty_list(tenants, "tenant"),
+        Condition::Words { words, at_least } => {
+            nonempty_list(words, "word")?;
+            if *at_least == 0 {
+                return Err("The least number of words must be 1 or more.".into());
+            }
+            words::WordList::new(words).map(|_| ())
+        }
+        Condition::Pattern { pattern, at_least } => {
+            if *at_least == 0 {
+                return Err("The least number of matches must be 1 or more.".into());
+            }
+            words::Pattern::new(pattern).map(|_| ())
+        }
+        Condition::Header {
+            name,
+            contains,
+            matches,
+        } => {
+            header_name(name)?;
+            if let Some(pattern) = matches {
+                words::Pattern::new(pattern)?;
+            }
+            if contains.is_some() && matches.is_some() {
+                return Err("A header condition is either contains or matches.".into());
+            }
+            Ok(())
+        }
+        Condition::AttachmentType { types } => nonempty_list(types, "type"),
+        Condition::AttachmentExtension { extensions } => nonempty_list(extensions, "extension"),
+        Condition::AttachmentName { pattern } => words::Pattern::new(pattern).map(|_| ()),
+        Condition::Detected { detectors } => {
+            nonempty_list(detectors, "detector")?;
+            for d in detectors {
+                if detectors::by_id(&d.id).is_none() {
+                    return Err(format!("There is no detector \"{}\".", d.id));
+                }
+                if d.at_least == 0 {
+                    return Err("A detector's least count must be 1 or more.".into());
+                }
+            }
+            Ok(())
+        }
+        Condition::RecipientOutside
+        | Condition::AttachmentSizeOver { .. }
+        | Condition::AttachmentCountOver { .. }
+        | Condition::CantBeInspected
+        | Condition::MessageSizeOver { .. } => Ok(()),
+    }
+}
+
+fn validate_action(action: &Action) -> Result<(), String> {
+    match action {
+        Action::AddDisclaimer { text: t, html, .. } => {
+            text(t, "disclaimer")?;
+            html.as_deref()
+                .map_or(Ok(()), |h| text(h, "disclaimer's HTML"))
+        }
+        Action::AddHeader { name, value } => {
+            header_name(name)?;
+            if value.len() > 998 || value.contains(['\r', '\n']) {
+                Err("A header value is one line of at most 998 characters.".into())
+            } else {
+                Ok(())
+            }
+        }
+        Action::RemoveHeader { name } => header_name(name),
+        Action::PrefixSubject { text: t } => text(t, "subject prefix"),
+        Action::AddRecipient { address } => {
+            if address.contains('@') {
+                Ok(())
+            } else {
+                Err(format!("\"{address}\" isn't an address."))
+            }
+        }
+        Action::Redirect { addresses } => {
+            nonempty_list(addresses, "address")?;
+            match addresses.iter().find(|a| !a.contains('@')) {
+                Some(a) => Err(format!("\"{a}\" isn't an address.")),
+                None => Ok(()),
+            }
+        }
+        Action::Refuse { text: t } => text(t, "refusal text"),
+        Action::Route { queue } => text(queue, "queue"),
+        Action::Block { notice } | Action::Warn { notice } | Action::Hold { notice, .. } => {
+            text(notice, "notice")
+        }
+    }
+}
+
+// --- Storage --------------------------------------------------------------
+
+struct Json<T>(T);
+
+impl<T: SerdeSerialize> Serialize for Json<T> {
+    fn serialize(&self) -> trc::Result<Vec<u8>> {
+        serde_json::to_vec(&self.0).map_err(|err| {
+            trc::StoreEvent::UnexpectedError
+                .into_err()
+                .details("Failed to serialize mail rule")
+                .reason(err)
+        })
+    }
+}
+
+impl<T: DeserializeOwned + Sync + Send> Deserialize for Json<T> {
+    fn deserialize(bytes: &[u8]) -> trc::Result<Self> {
+        serde_json::from_slice(bytes).map(Json).map_err(|err| {
+            trc::StoreEvent::DataCorruption
+                .into_err()
+                .details("Invalid mail rule")
+                .reason(err)
+        })
+    }
+}
+
+fn class(id: u32) -> ValueClass {
+    let mut key = Vec::with_capacity(6);
+    key.push(FEATURE);
+    key.push(KIND_RULE);
+    key.extend_from_slice(&id.to_be_bytes());
+    ValueClass::Any(AnyClass {
+        subspace: SUBSPACE_INBUXA,
+        key,
+    })
+}
+
+fn key(id: u32) -> ValueKey<ValueClass> {
+    ValueKey::from(class(id))
+}
+
+pub async fn get(data: &Store, id: u32) -> trc::Result<Option<Rule>> {
+    Ok(data
+        .get_value::<Json<Rule>>(key(id))
+        .await
+        .caused_by(trc::location!())?
+        .map(|Json(rule)| rule))
+}
+
+/// Every rule, in the order they run: by priority, then oldest first.
+pub async fn all(data: &Store) -> trc::Result<Vec<Rule>> {
+    let mut rules = Vec::new();
+    data.iterate(IterateParams::new(key(0), key(u32::MAX)), |_, value| {
+        if let Ok(Json(rule)) = Json::<Rule>::deserialize(value) {
+            rules.push(rule);
+        }
+        Ok(true)
+    })
+    .await
+    .caused_by(trc::location!())?;
+    rules.sort_by_key(|rule| (rule.priority, rule.id));
+    Ok(rules)
+}
+
+/// Writes a new rule under the next free id, which it returns. Two nodes
+/// creating rules at once can't take the same id: the key must be absent.
+pub async fn create(data: &Store, rule: &Rule) -> trc::Result<u32> {
+    let mut attempt = 0;
+    loop {
+        attempt += 1;
+        let id = all(data).await?.iter().map(|r| r.id).max().unwrap_or(0) + 1;
+        let stored = Rule { id, ..rule.clone() };
+        let mut batch = BatchBuilder::new();
+        batch.assert_value(class(id), AssertValue::None);
+        batch.set(class(id), Json(&stored).serialize()?);
+        match data.write(batch.build_all()).await {
+            Ok(_) => {
+                super::cache::invalidate();
+                return Ok(id);
+            }
+            Err(err)
+                if attempt < CREATE_ATTEMPTS
+                    && matches!(
+                        err.as_ref(),
+                        trc::EventType::Store(trc::StoreEvent::AssertValueFailed)
+                    ) => {}
+            Err(err) => return Err(err.caused_by(trc::location!())),
+        }
+    }
+}
+
+/// Replaces a stored rule (same id).
+pub async fn update(data: &Store, rule: &Rule) -> trc::Result<()> {
+    let mut batch = BatchBuilder::new();
+    batch.set(class(rule.id), Json(rule).serialize()?);
+    data.write(batch.build_all())
+        .await
+        .caused_by(trc::location!())?;
+    super::cache::invalidate();
+    Ok(())
+}
+
+pub async fn delete(data: &Store, id: u32) -> trc::Result<()> {
+    let mut batch = BatchBuilder::new();
+    batch.clear(class(id));
+    data.write(batch.build_all())
+        .await
+        .caused_by(trc::location!())?;
+    super::cache::invalidate();
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn rule(kind: Kind, actions: Vec<Action>) -> Rule {
+        Rule {
+            id: 0,
+            name: "Cards outside".into(),
+            description: String::new(),
+            kind,
+            enabled: true,
+            priority: 0,
+            direction: Direction::Outgoing,
+            conditions: vec![Condition::RecipientOutside],
+            exceptions: vec![],
+            actions,
+            stop_processing: false,
+            created_by: String::new(),
+            created_at: 0,
+            updated_at: 0,
+        }
+    }
+
+    #[test]
+    fn wire_format() {
+        let json = r#"{"name":"Cards","kind":"dlp","direction":"outgoing",
+            "conditions":[{"type":"recipientOutside"},{"type":"detected","detectors":[{"id":"payment-card","atLeast":5}]}],
+            "actions":[{"type":"hold","notice":"Held for review","notifySender":true}]}"#;
+        let parsed: Rule = serde_json::from_str(json).unwrap();
+        assert!(parsed.enabled);
+        assert_eq!(
+            parsed.conditions[1],
+            Condition::Detected {
+                detectors: vec![DetectorMin {
+                    id: "payment-card".into(),
+                    at_least: 5
+                }]
+            }
+        );
+        assert_eq!(
+            parsed.actions[0],
+            Action::Hold {
+                notice: "Held for review".into(),
+                notify_sender: true
+            }
+        );
+        assert!(parsed.validate().is_ok());
+        let back = serde_json::to_value(&parsed).unwrap();
+        assert_eq!(back["actions"][0]["notifySender"], true);
+    }
+
+    #[test]
+    fn dlp_rules_have_one_dlp_action_on_outgoing_mail() {
+        let block = Action::Block {
+            notice: "No.".into(),
+        };
+        assert!(rule(Kind::Dlp, vec![block.clone()]).validate().is_ok());
+        let two = rule(
+            Kind::Dlp,
+            vec![
+                block.clone(),
+                Action::Warn {
+                    notice: "Hm.".into(),
+                },
+            ],
+        );
+        assert_eq!(two.validate().unwrap_err().property, "actions");
+        let mixed = rule(
+            Kind::Dlp,
+            vec![block.clone(), Action::PrefixSubject { text: "[x]".into() }],
+        );
+        assert_eq!(mixed.validate().unwrap_err().property, "actions");
+        let mut inbound = rule(Kind::Dlp, vec![block.clone()]);
+        inbound.direction = Direction::Incoming;
+        assert_eq!(inbound.validate().unwrap_err().property, "direction");
+        assert_eq!(
+            rule(Kind::Transport, vec![block])
+                .validate()
+                .unwrap_err()
+                .property,
+            "actions"
+        );
+    }
+
+    #[test]
+    fn conditions_and_actions_are_checked() {
+        let disclaimer = Action::AddDisclaimer {
+            text: "Sent from Example Co.".into(),
+            html: None,
+            position: Position::Bottom,
+        };
+        let mut r = rule(Kind::Transport, vec![disclaimer]);
+        assert!(r.validate().is_ok());
+        r.conditions.push(Condition::Detected {
+            detectors: vec![DetectorMin {
+                id: "iban".into(),
+                at_least: 1,
+            }],
+        });
+        assert_eq!(r.validate().unwrap_err().property, "conditions");
+
+        let mut r = rule(
+            Kind::Dlp,
+            vec![Action::Block {
+                notice: "No.".into(),
+            }],
+        );
+        r.conditions = vec![Condition::Detected {
+            detectors: vec![DetectorMin {
+                id: "nope".into(),
+                at_least: 1,
+            }],
+        }];
+        assert!(r.validate().unwrap_err().reason.contains("nope"));
+        r.conditions = vec![Condition::Pattern {
+            pattern: "(".into(),
+            at_least: 1,
+        }];
+        assert!(r.validate().is_err());
+        r.conditions = vec![Condition::Words {
+            words: vec![],
+            at_least: 1,
+        }];
+        assert!(r.validate().is_err());
+        r.exceptions = vec![Condition::Header {
+            name: "X-Bad: yes".into(),
+            contains: None,
+            matches: None,
+        }];
+        r.conditions = vec![];
+        assert_eq!(r.validate().unwrap_err().property, "exceptions");
+
+        let header = rule(
+            Kind::Transport,
+            vec![Action::AddHeader {
+                name: "X-Tag".into(),
+                value: "a\r\nBcc: x@y".into(),
+            }],
+        );
+        assert!(header.validate().is_err());
+        let redirect = rule(
+            Kind::Transport,
+            vec![Action::Redirect {
+                addresses: vec!["nobody".into()],
+            }],
+        );
+        assert!(redirect.validate().is_err());
+        let mut unnamed = rule(
+            Kind::Transport,
+            vec![Action::RemoveHeader {
+                name: "X-Tag".into(),
+            }],
+        );
+        unnamed.name = " ".into();
+        assert_eq!(unnamed.validate().unwrap_err().property, "name");
+    }
+}