ecr_store/
notmuch_store.rs

1use crate::error::{Error, Result};
2use crate::index::{Freshness, IndexStatus, MessageIndex};
3use crate::notmuch::Notmuch;
4use crate::paths::MailPaths;
5use crate::store::{BodyOptions, MailStore, ProgressSink};
6use crate::{discovery, index, mbsync, msmtp, oauth};
7use ecr_core::account::{Account, AccountId};
8use ecr_core::doctor::Doctor;
9use ecr_core::message::{
10    Body, Message, MessageId, Part, PartId, Query, SyncReport, TagOp, Thread, ThreadId,
11    ThreadSummary,
12};
13use ecr_core::revision::Revision;
14use std::sync::Arc;
15use std::time::Instant;
16
17pub struct NotmuchStore {
18    paths: Arc<MailPaths>,
19    notmuch: Notmuch,
20    index: Option<MessageIndex>,
21    /// Whether the index may answer this request. See `index::freshness`.
22    freshness: Freshness,
23}
24
25impl NotmuchStore {
26    pub fn open() -> Result<Self> {
27        Ok(Self::new(Arc::new(MailPaths::discover()?)))
28    }
29
30    pub fn new(paths: Arc<MailPaths>) -> Self {
31        let index = paths.use_index.then(|| MessageIndex::open(&paths)).and_then(
32            |opened| match opened {
33                Ok(index) => Some(index),
34                Err(err) => {
35                    tracing::warn!(%err, "could not open the mail index; every read will ask notmuch");
36                    None
37                }
38            },
39        );
40
41        Self {
42            notmuch: Notmuch::new(Arc::clone(&paths)),
43            paths,
44            index,
45            freshness: Freshness::default(),
46        }
47    }
48
49    pub fn paths(&self) -> &MailPaths {
50        &self.paths
51    }
52
53    pub fn notmuch(&self) -> &Notmuch {
54        &self.notmuch
55    }
56
57    pub fn index_status(&self) -> Option<IndexStatus> {
58        self.index.as_ref().map(|index| index.status())
59    }
60
61    /// Builds or catches up the index, rebuilding it if that is what it takes.
62    ///
63    /// This is the caller that is allowed to be slow — a first build of a 46k
64    /// inbox is around 80 seconds — so it is never run from a request. The
65    /// server spawns it at startup and the watcher runs it when the database
66    /// moves; reads fall through to notmuch for as long as it takes.
67    pub async fn refresh_index(&self) -> Result<Option<index::Refreshed>> {
68        let Some(index) = self.index.as_ref() else {
69            return Ok(None);
70        };
71
72        let generation = self.freshness.generation();
73
74        self.freshness.begin_build();
75        let refreshed = index::refresh(index, &self.notmuch).await;
76        self.freshness.end_build();
77
78        self.freshness.vouch(generation);
79        Ok(Some(refreshed?))
80    }
81
82    /// The index, if it can be trusted to answer this request.
83    ///
84    /// Anything that goes wrong here answers `None`, which is a slower request
85    /// and never a wrong one.
86    async fn reading_index(&self) -> Option<&MessageIndex> {
87        let index = self.index.as_ref()?;
88
89        if self.freshness.building() {
90            return None;
91        }
92
93        if self.freshness.fresh() {
94            return Some(index);
95        }
96
97        let generation = self.freshness.generation();
98        let revision = self.notmuch.revision().await.ok()?;
99        let held = index.revision().ok().flatten()?;
100
101        if held != revision {
102            index::refresh_incremental(index, &self.notmuch)
103                .await
104                .ok()
105                .flatten()?;
106        }
107
108        self.freshness.vouch(generation);
109        Some(index)
110    }
111
112    fn channels_for(&self, accounts: &[AccountId]) -> Vec<String> {
113        let discovered = discovery::accounts(&self.paths);
114        discovered
115            .into_iter()
116            .filter(|a| accounts.is_empty() || accounts.contains(&a.id))
117            .filter_map(|a| a.mbsync_channel)
118            .collect()
119    }
120}
121
122impl MailStore for NotmuchStore {
123    async fn revision(&self) -> Result<Revision> {
124        self.notmuch.revision().await
125    }
126
127    async fn accounts(&self) -> Result<Vec<Account>> {
128        Ok(discovery::accounts(&self.paths))
129    }
130
131    async fn search_threads(&self, query: &Query) -> Result<Vec<ThreadSummary>> {
132        if let Some(index) = self.reading_index().await {
133            match index.search_threads(query) {
134                Ok(Some(threads)) => return Ok(threads),
135                Ok(None) => {}
136                Err(err) => tracing::warn!(%err, "the mail index could not answer a search"),
137            }
138        }
139
140        self.notmuch.search_threads(query).await
141    }
142
143    async fn count(&self, query: &Query) -> Result<usize> {
144        if let Some(index) = self.reading_index().await {
145            match index.count(query.effective_text()) {
146                Ok(Some(count)) => return Ok(count as usize),
147                Ok(None) => {}
148                Err(err) => tracing::warn!(%err, "the mail index could not answer a count"),
149            }
150        }
151
152        self.notmuch.count(query).await
153    }
154
155    /// The sidebar's rows are mostly tags, and one saved free-text query among
156    /// them must not cost every other row its answer — so the ones the index
157    /// can take are taken, and only the rest reach notmuch, still in one batch.
158    async fn count_batch(&self, queries: &[String]) -> Result<Vec<u64>> {
159        let mut answers: Vec<Option<u64>> = vec![None; queries.len()];
160
161        if let Some(index) = self.reading_index().await {
162            for (slot, query) in answers.iter_mut().zip(queries) {
163                let text = query.trim();
164                if text.is_empty() {
165                    *slot = Some(0);
166                    continue;
167                }
168                match index.count(text) {
169                    Ok(count) => *slot = count,
170                    Err(err) => tracing::warn!(%err, "the mail index could not answer a count"),
171                }
172            }
173        }
174
175        let remaining: Vec<String> = queries
176            .iter()
177            .zip(&answers)
178            .filter(|(_, answer)| answer.is_none())
179            .map(|(query, _)| query.clone())
180            .collect();
181
182        if !remaining.is_empty() {
183            let counted = self.notmuch.count_batch(&remaining).await?;
184            let mut counted = counted.into_iter();
185            for slot in answers.iter_mut().filter(|slot| slot.is_none()) {
186                *slot = counted.next();
187            }
188        }
189
190        Ok(answers.into_iter().map(|a| a.unwrap_or(0)).collect())
191    }
192
193    async fn thread(&self, id: &ThreadId) -> Result<Thread> {
194        self.notmuch.thread(id).await
195    }
196
197    async fn message(&self, id: &MessageId) -> Result<Message> {
198        self.notmuch.message_with_parts(id).await
199    }
200
201    async fn body(&self, id: &MessageId, options: BodyOptions) -> Result<Body> {
202        self.notmuch
203            .body(id, options.format, options.allow_remote_resources)
204            .await
205    }
206
207    async fn part(&self, id: &MessageId, part: &PartId) -> Result<Part> {
208        self.notmuch.part(id, part).await
209    }
210
211    async fn tag(&self, ops: &[TagOp]) -> Result<Revision> {
212        let revision = self.notmuch.tag(ops).await?;
213        self.freshness.note_write();
214        Ok(revision)
215    }
216
217    async fn sync(
218        &self,
219        accounts: &[AccountId],
220        progress: &dyn ProgressSink,
221    ) -> Result<SyncReport> {
222        let started = Instant::now();
223        let channels = self.channels_for(accounts);
224        let before = self.notmuch.count(&Query::new("*")).await.unwrap_or(0);
225
226        let warnings = mbsync::run(&self.paths, &channels, progress).await?;
227
228        progress.line("indexing new mail");
229        self.notmuch.index_new().await?;
230        self.freshness.note_write();
231
232        let after = self.notmuch.count(&Query::new("*")).await.unwrap_or(before);
233
234        Ok(SyncReport {
235            channels,
236            new_messages: after.saturating_sub(before),
237            duration_ms: started.elapsed().as_millis() as u64,
238            warnings,
239        })
240    }
241
242    async fn send(&self, account: &AccountId, raw: &[u8]) -> Result<()> {
243        let accounts = discovery::accounts(&self.paths);
244        let found = accounts.iter().find(|a| &a.id == account);
245        let msmtp_account = found
246            .and_then(|a| a.msmtp_account.clone())
247            .unwrap_or_else(|| account.to_string());
248
249        // msmtp's only signal for a dead OAuth token is a non-zero exit with
250        // opaque stderr. If this account uses OAuth, name the fix so the
251        // compose pane can show something the user can act on.
252        msmtp::send(&self.paths, &msmtp_account, raw)
253            .await
254            .map_err(|e| match e {
255                Error::ToolFailed { tool, stderr } => {
256                    let profile = found.and_then(|a| discovery::oauth_profile(&self.paths, a));
257                    let stderr = match profile.as_deref() {
258                        Some(profile) => format!("{stderr}\n\n{}", oauth::authorize_hint(profile)),
259                        None => stderr,
260                    };
261                    Error::ToolFailed { tool, stderr }
262                }
263                other => other,
264            })
265    }
266
267    async fn doctor(&self) -> Doctor {
268        crate::doctor::run_with_paths(&self.paths).await
269    }
270}
ecr_store/notmuch_store.rs

ecr_store/
notmuch_store.rs
