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                // Ahead of the event, not after it: the clients this wakes ask
70                // for the new page immediately, and an index that has not
71                // caught up yet answers the old one — which reads as mail
72                // arriving and then not being there.
73                if let Err(err) = state.store.refresh_index().await {
74                    tracing::warn!(%err, "could not bring the mail index up to the new mail");
75                }
76
77                tracing::info!(%revision, "indexed newly delivered mail");
78                state.events.publish(ServerEvent::MailChanged { revision });
79            }
80            Err(err) => {
81                tracing::warn!(%err, "could not index delivered mail");
82                publish_error(&state.events, &err.to_string());
83            }
84        }
85    }
86}
87
88fn publish_error(bus: &EventBus, detail: &str) {
89    bus.publish(ServerEvent::Error {
90        detail: detail.to_string(),
91    });
92}
93
94#[cfg(test)]
95mod tests {
96    use super::*;
97    use notify::event::{CreateKind, EventKind};
98    use std::path::PathBuf;
99
100    fn event(kind: EventKind, path: &str) -> notify::Event {
101        notify::Event {
102            kind,
103            paths: vec![PathBuf::from(path)],
104            attrs: Default::default(),
105        }
106    }
107
108    #[test]
109    fn a_new_maildir_file_counts_as_a_delivery() {
110        assert!(is_delivery(&event(
111            EventKind::Create(CreateKind::File),
112            "/Mail/main/Inbox/cur/123:2,"
113        )));
114    }
115
116    #[test]
117    fn notmuch_writing_its_own_database_does_not_count() {
118        assert!(!is_delivery(&event(
119            EventKind::Create(CreateKind::File),
120            "/Mail/.notmuch/xapian/postlist.glass"
121        )));
122    }
123
124    #[test]
125    fn mbsync_state_files_do_not_count() {
126        assert!(!is_delivery(&event(
127            EventKind::Create(CreateKind::File),
128            "/Mail/main/Inbox/.mbsyncstate"
129        )));
130    }
131
132    #[test]
133    fn access_events_do_not_count() {
134        assert!(!is_delivery(&event(
135            EventKind::Access(notify::event::AccessKind::Read),
136            "/Mail/main/Inbox/cur/123:2,"
137        )));
138    }
139}
ecr_server/watcher.rs

ecr_server/
watcher.rs
