ecr_server/
watcher.rs

1use crate::events::{EventBus, ServerEvent};
2use crate::state::AppState;
3use notify::{RecommendedWatcher, RecursiveMode, Watcher};
4use std::path::Path;
5use std::time::Duration;
6use tokio::sync::mpsc;
7
8const DEBOUNCE: Duration = Duration::from_secs(2);
9
10pub fn spawn(state: AppState) -> anyhow::Result<RecommendedWatcher> {
11    let root = state.store.paths().maildir_root.clone();
12    let (tx, rx) = mpsc::unbounded_channel();
13
14    let mut watcher = notify::recommended_watcher(move |event: notify::Result<notify::Event>| {
15        if let Ok(event) = event {
16            if is_delivery(&event) {
17                let _ = tx.send(());
18            }
19        }
20    })?;
21
22    watcher.watch(Path::new(&root), RecursiveMode::Recursive)?;
23    tokio::spawn(debounce_loop(state, rx));
24
25    tracing::info!(root = %root.display(), "watching maildir for delivered mail");
26    Ok(watcher)
27}
28
29fn is_delivery(event: &notify::Event) -> bool {
30    use notify::EventKind;
31
32    if !matches!(
33        event.kind,
34        EventKind::Create(_) | EventKind::Modify(_) | EventKind::Remove(_)
35    ) {
36        return false;
37    }
38
39    event.paths.iter().any(|p| {
40        let text = p.to_string_lossy();
41        !text.contains("/.notmuch/") && !text.ends_with(".mbsyncstate") && !text.ends_with(".lock")
42    })
43}
44
45async fn debounce_loop(state: AppState, mut rx: mpsc::UnboundedReceiver<()>) {
46    while rx.recv().await.is_some() {
47        while let Ok(Some(())) = tokio::time::timeout(DEBOUNCE, rx.recv()).await {}
48
49        if state.read_only {
50            tracing::debug!("maildir changed but the server is read-only; not indexing");
51            continue;
52        }
53
54        match state.store.notmuch().index_new().await {
55            Ok(revision) => {
56                // A tag write of our own renames the file behind it — notmuch
57                // synchronises maildir flags — and that rename arrives here
58                // looking exactly like a delivery. The database standing where
59                // that write left it is the proof nothing was delivered:
60                // `notmuch new` had nothing to add, so it moved no further.
61                // Announcing it as new mail would refresh every client's list
62                // for the one tag change that happens by itself, taking the
63                // row being read out from under the reader.
64                if state.own_write(&revision).await {
65                    tracing::debug!(%revision, "maildir changed to match our own tag write");
66                    continue;
67                }
68
69                tracing::info!(%revision, "indexed newly delivered mail");
70                state.events.publish(ServerEvent::MailChanged { revision });
71            }
72            Err(err) => {
73                tracing::warn!(%err, "could not index delivered mail");
74                publish_error(&state.events, &err.to_string());
75            }
76        }
77    }
78}
79
80fn publish_error(bus: &EventBus, detail: &str) {
81    bus.publish(ServerEvent::Error {
82        detail: detail.to_string(),
83    });
84}
85
86#[cfg(test)]
87mod tests {
88    use super::*;
89    use notify::event::{CreateKind, EventKind};
90    use std::path::PathBuf;
91
92    fn event(kind: EventKind, path: &str) -> notify::Event {
93        notify::Event {
94            kind,
95            paths: vec![PathBuf::from(path)],
96            attrs: Default::default(),
97        }
98    }
99
100    #[test]
101    fn a_new_maildir_file_counts_as_a_delivery() {
102        assert!(is_delivery(&event(
103            EventKind::Create(CreateKind::File),
104            "/Mail/main/Inbox/cur/123:2,"
105        )));
106    }
107
108    #[test]
109    fn notmuch_writing_its_own_database_does_not_count() {
110        assert!(!is_delivery(&event(
111            EventKind::Create(CreateKind::File),
112            "/Mail/.notmuch/xapian/postlist.glass"
113        )));
114    }
115
116    #[test]
117    fn mbsync_state_files_do_not_count() {
118        assert!(!is_delivery(&event(
119            EventKind::Create(CreateKind::File),
120            "/Mail/main/Inbox/.mbsyncstate"
121        )));
122    }
123
124    #[test]
125    fn access_events_do_not_count() {
126        assert!(!is_delivery(&event(
127            EventKind::Access(notify::event::AccessKind::Read),
128            "/Mail/main/Inbox/cur/123:2,"
129        )));
130    }
131}
ecr_server/watcher.rs

ecr_server/
watcher.rs
