connect esmtp server to queue

We can now queue messages to per-domain queues.
The per-domain queues feed into a destination site keyed
by distinct set of MX host names.

The destination site doesn't know what to do with messages yet.

crates/kumod/Cargo.toml

@@ -12,6 +12,7 @@ lazy_static = "1.4"
 mlua = {version="0.8", features=["vendored", "lua54", "async", "send", "serialize"]}
 serde = {version="1.0", features=["derive"]}
 serde_json = "1.0"
+mail-auth = "0.3"
 message = {path="../message"}
 spool = {path="../spool"}
 timeq = {path="../timeq"}

crates/kumod/src/dest_site.rs

@@ -0,0 +1,186 @@
+use mail_auth::{Resolver, MX};
+use message::Message;
+use std::collections::HashMap;
+use std::sync::Arc;
+use tokio::sync::mpsc::error::TrySendError;
+use tokio::sync::mpsc::{Receiver, Sender};
+use tokio::sync::{Mutex, MutexGuard};
+
+lazy_static::lazy_static! {
+    static ref MANAGER: Mutex<SiteManager> = Mutex::new(SiteManager::new());
+    static ref RESOLVER: Mutex<Resolver> = Mutex::new(Resolver::new_system_conf().unwrap());
+}
+
+pub struct SiteManager {
+    sites: HashMap<String, SiteHandle>,
+}
+
+impl SiteManager {
+    pub fn new() -> Self {
+        Self {
+            sites: HashMap::new(),
+        }
+    }
+
+    pub async fn get() -> MutexGuard<'static, Self> {
+        MANAGER.lock().await
+    }
+
+    pub async fn resolve_domain(name: &str) -> anyhow::Result<SiteHandle> {
+        let resolver = RESOLVER.lock().await;
+        let mx = resolver
+            .mx_lookup(name)
+            .await
+            .map_err(|err| anyhow::anyhow!("MX lookup for {name} failed: {err:#}"))?;
+        let name = factor_mx_list(&mx);
+
+        let mut manager = Self::get().await;
+        let handle = manager.sites.entry(name.clone()).or_insert_with(|| {
+            let (tx, rx) = tokio::sync::mpsc::channel(1024 /* FIXME: configurable */);
+            SiteHandle(Arc::new(Mutex::new(DestinationSite {
+                name: name.clone(),
+                mx,
+                tx,
+                rx,
+            })))
+        });
+        Ok(handle.clone())
+    }
+}
+
+#[derive(Clone)]
+pub struct SiteHandle(Arc<Mutex<DestinationSite>>);
+
+impl SiteHandle {
+    pub async fn lock(&self) -> MutexGuard<DestinationSite> {
+        self.0.lock().await
+    }
+}
+
+pub struct DestinationSite {
+    name: String,
+    mx: Arc<Vec<MX>>,
+    tx: Sender<Message>,
+    rx: Receiver<Message>,
+}
+
+impl DestinationSite {
+    pub fn name(&self) -> &str {
+        &self.name
+    }
+
+    pub fn insert(&self, msg: Message) -> Result<(), TrySendError<Message>> {
+        self.tx.try_send(msg)
+    }
+}
+
+/// Given a set of MX records, produce a pseudo-regex style alternation
+/// list of the underlying hostnames
+fn factor_mx_list(mx: &Arc<Vec<MX>>) -> String {
+    let mut names = vec![];
+    for entry in mx.iter() {
+        for host in &entry.exchanges {
+            names.push(host.as_str());
+        }
+    }
+
+    factor_names(&names)
+}
+
+/// Given a list of host names, produce a pseudo-regex style alternation list
+/// of the different elements of the hostnames.
+/// The goal is to produce a more compact representation of the name list
+/// with the common components factored out.
+fn factor_names(names: &[&str]) -> String {
+    let mut max_element_count = 0;
+
+    let mut elements: Vec<Vec<&str>> = vec![];
+
+    let mut split_names = vec![];
+    for name in names {
+        let mut fields: Vec<_> = name.split('.').map(|s| s.to_lowercase()).collect();
+        fields.reverse();
+        max_element_count = max_element_count.max(fields.len());
+        split_names.push(fields);
+    }
+
+    fn add_element<'a>(elements: &mut Vec<Vec<&'a str>>, field: &'a str, i: usize) {
+        match elements.get_mut(i) {
+            Some(ele) => {
+                if !ele.contains(&field) {
+                    ele.push(field);
+                }
+            }
+            None => {
+                elements.push(vec![field]);
+            }
+        }
+    }
+
+    for fields in &split_names {
+        for (i, field) in fields.iter().enumerate() {
+            add_element(&mut elements, field, i);
+        }
+        for i in fields.len()..max_element_count {
+            add_element(&mut elements, "?", i);
+        }
+    }
+
+    let mut result = vec![];
+    for mut ele in elements {
+        let has_q = ele.contains(&"?");
+        ele.retain(|&e| e != "?");
+        let mut item_text = if ele.len() == 1 {
+            ele[0].to_string()
+        } else {
+            format!("({})", ele.join("|"))
+        };
+        if has_q {
+            item_text.push('?');
+        }
+        result.push(item_text);
+    }
+    result.reverse();
+
+    result.join(".")
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+
+    #[test]
+    fn name_factoring() {
+        assert_eq!(
+            factor_names(&[
+                "mta5.am0.yahoodns.net",
+                "mta6.am0.yahoodns.net",
+                "mta7.am0.yahoodns.net"
+            ]),
+            "(mta5|mta6|mta7).am0.yahoodns.net".to_string()
+        );
+
+        // Verify that the case is normalized to lowercase
+        assert_eq!(
+            factor_names(&[
+                "mta5.AM0.yahoodns.net",
+                "mta6.am0.yAHOodns.net",
+                "mta7.am0.yahoodns.net"
+            ]),
+            "(mta5|mta6|mta7).am0.yahoodns.net".to_string()
+        );
+
+        // When the names have mismatched lengths, do we produce
+        // something reasonable?
+        assert_eq!(
+            factor_names(&[
+                "gmail-smtp-in.l.google.com",
+                "alt1.gmail-smtp-in.l.google.com",
+                "alt2.gmail-smtp-in.l.google.com",
+                "alt3.gmail-smtp-in.l.google.com",
+                "alt4.gmail-smtp-in.l.google.com",
+            ]),
+            "(alt1|alt2|alt3|alt4)?.gmail-smtp-in.l.google.com".to_string()
+        );
+    }
+}

crates/kumod/src/main.rs

@@ -3,6 +3,7 @@ use std::path::PathBuf;
 use tracing_subscriber::prelude::*;
 use tracing_subscriber::{fmt, EnvFilter};
 
+mod dest_site;
 mod lua_config;
 mod mod_kumo;
 mod queue;

crates/kumod/src/queue.rs

@@ -1,8 +1,11 @@
+use crate::dest_site::SiteManager;
 use message::Message;
 use std::collections::HashMap;
-use std::sync::{Arc, Mutex, MutexGuard};
+use std::sync::Arc;
 use std::time::Duration;
 use timeq::{TimeQ, TimerError};
+use tokio::sync::mpsc::error::TrySendError;
+use tokio::sync::{Mutex, MutexGuard};
 use tokio::task::JoinHandle;
 
 lazy_static::lazy_static! {
@@ -13,8 +16,8 @@ lazy_static::lazy_static! {
 pub struct QueueHandle(Arc<Mutex<Queue>>);
 
 impl QueueHandle {
-    pub fn lock(&self) -> MutexGuard<Queue> {
-        self.0.lock().unwrap()
+    pub async fn lock(&self) -> MutexGuard<Queue> {
+        self.0.lock().await
     }
 }
 
@@ -33,7 +36,7 @@ impl Drop for Queue {
 }
 
 impl Queue {
-    pub fn new(name: String) -> QueueHandle {
+    pub async fn new(name: String) -> QueueHandle {
         let handle = QueueHandle(Arc::new(Mutex::new(Queue {
             name: name.clone(),
             queue: TimeQ::new(),
@@ -43,23 +46,49 @@ impl Queue {
         let queue_clone = handle.clone();
         let maintainer = tokio::spawn(async move {
             if let Err(err) = maintain_named_queue(&queue_clone).await {
-                tracing::error!("maintain_named_queue {}: {err:#}", queue_clone.lock().name);
+                tracing::error!(
+                    "maintain_named_queue {}: {err:#}",
+                    queue_clone.lock().await.name
+                );
             }
         });
-        handle.lock().maintainer.replace(maintainer);
+        handle.lock().await.maintainer.replace(maintainer);
         handle
     }
 
-    pub fn insert(&mut self, msg: Message) {
+    pub async fn insert(&mut self, msg: Message) -> anyhow::Result<()> {
         match self.queue.insert(Arc::new(msg)) {
-            Ok(_) => {}
+            Ok(_) => Ok(()),
             Err(TimerError::Expired(msg)) => {
-                // TODO: for immediately ready messages,
-                // immediately instantiate the destination site
-                // and add to its queue
-                tracing::error!("queue to destination site");
+                let msg = (*msg).clone();
+                match SiteManager::resolve_domain(&self.name).await {
+                    Ok(site) => {
+                        let site = site.lock().await;
+                        println!("site is {}", site.name());
+                        match site.insert(msg) {
+                            Ok(_) => {}
+                            Err(TrySendError::Closed(msg)) | Err(TrySendError::Full(msg)) => {
+                                msg.delay_by(Duration::from_secs(60));
+                                self.queue
+                                    .insert(Arc::new(msg))
+                                    .map_err(|_err| anyhow::anyhow!("failed to insert"))?;
+                            }
+                        }
+                    }
+                    Err(err) => {
+                        tracing::error!("Failed to resolve {}: {err:#}", self.name);
+                        msg.delay_by(Duration::from_secs(60));
+                        self.queue
+                            .insert(Arc::new(msg))
+                            .map_err(|_err| anyhow::anyhow!("failed to insert"))?;
+                    }
+                }
+
+                Ok(())
+            }
+            Err(TimerError::NotFound) => {
+                anyhow::bail!("queue.insert returned impossible NotFound error")
             }
-            Err(TimerError::NotFound) => unreachable!(),
         }
     }
 }
@@ -79,17 +108,23 @@ impl QueueManager {
     /// Note that the queue names are case-insensitive, and
     /// internally the lowercased version of `name` is used
     /// to track the queue.
-    pub fn insert(&mut self, name: &str, msg: Message) {
+    pub async fn insert(&mut self, name: &str, msg: Message) -> anyhow::Result<()> {
         let name = name.to_lowercase();
-        let queue = self
-            .named
-            .entry(name.clone())
-            .or_insert_with(|| Queue::new(name));
-        queue.lock().insert(msg);
+        let entry_keeper;
+        let entry = match self.named.get(&name) {
+            Some(e) => e,
+            None => {
+                entry_keeper = Queue::new(name.clone()).await;
+                self.named.insert(name, entry_keeper.clone());
+                &entry_keeper
+            }
+        };
+        let mut entry = entry.lock().await;
+        entry.insert(msg).await
     }
 
-    pub fn get() -> MutexGuard<'static, Self> {
-        MANAGER.lock().unwrap()
+    pub async fn get() -> MutexGuard<'static, Self> {
+        MANAGER.lock().await
     }
 }
 
@@ -97,7 +132,7 @@ async fn maintain_named_queue(queue: &QueueHandle) -> anyhow::Result<()> {
     loop {
         tokio::time::sleep(Duration::from_secs(60)).await;
         {
-            let q = queue.lock();
+            let q = queue.lock().await;
             println!(
                 "maintaining queue {} which has {} entries",
                 q.name,

crates/kumod/src/smtp_server.rs

@@ -1,4 +1,5 @@
 use crate::lua_config::{load_config, LuaConfig};
+use crate::queue::QueueManager;
 use anyhow::anyhow;
 use message::{EnvelopeAddress, Message};
 use std::fmt::Debug;
@@ -203,9 +204,10 @@ impl<T: AsyncRead + AsyncWrite + Debug + Send + 'static> SmtpServer<T> {
                         messages.push(message);
                     }
 
+                    let mut queue_manager = QueueManager::get().await;
                     for msg in messages {
                         let domain = msg.recipient()?.domain().to_string();
-                        crate::queue::QueueManager::get().insert(&domain, msg);
+                        queue_manager.insert(&domain, msg).await?;
                     }
 
                     let ids = ids.join(" ");

crates/message/src/message.rs

@@ -88,6 +88,11 @@ impl Message {
         inner.due
     }
 
+    pub fn delay_by(&self, duration: Duration) {
+        let due = Utc::now() + chrono::Duration::milliseconds(duration.as_millis() as _);
+        self.set_due(Some(due));
+    }
+
     pub fn set_due(&self, due: Option<DateTime<Utc>>) {
         let mut inner = self.inner.lock().unwrap();
         inner.due = due;