1mod batch;
2mod json;
3mod query;
4
5pub use json::{split_authors, ShowMessage, ThreadNode};
6pub use query::{and, escape_query_value};
7
8use crate::error::{Error, Result};
9use crate::paths::MailPaths;
10use ecr_core::message::{
11 Address, Body, BodyFormat, Message, MessageId, Part, PartId, Query, TagOp, Thread, ThreadId,
12 ThreadSummary,
13};
14use ecr_core::revision::Revision;
15use std::collections::HashMap;
16use std::io::{BufRead, BufReader};
17use std::path::{Path, PathBuf};
18use std::sync::Arc;
19use tokio::process::Command;
20use tokio::sync::Mutex;
21
22#[derive(Debug, Clone, serde::Serialize)]
24pub struct MailingList {
25 pub id: String,
27 pub name: String,
29 pub count: u32,
30}
31
32fn read_list_id(path: &Path) -> Option<String> {
35 let file = std::fs::File::open(path).ok()?;
36 let mut reader = BufReader::new(file);
37 let mut line = String::new();
38 let mut found: Option<String> = None;
39
40 loop {
41 line.clear();
42 let read = reader.read_line(&mut line).ok()?;
43 if read == 0 {
44 break;
45 }
46 let trimmed = line.trim_end_matches(['\r', '\n']);
47 if trimmed.is_empty() {
48 break;
49 }
50
51 if let Some(value) = found.as_mut() {
53 if trimmed.starts_with([' ', '\t']) {
54 value.push(' ');
55 value.push_str(trimmed.trim());
56 continue;
57 }
58 break;
59 }
60
61 if let Some(rest) = trimmed
62 .get(..8)
63 .filter(|head| head.eq_ignore_ascii_case("list-id:"))
64 .map(|_| &trimmed[8..])
65 {
66 found = Some(rest.trim().to_string());
67 }
68 }
69
70 found.filter(|v| !v.is_empty())
71}
72
73fn split_list_id(raw: &str) -> (String, String) {
76 match (raw.rfind('<'), raw.rfind('>')) {
77 (Some(open), Some(close)) if close > open => {
78 let id = raw[open + 1..close].trim().to_string();
79 let name = raw[..open].trim().trim_matches('"').trim().to_string();
80 let name = if name.is_empty() { id.clone() } else { name };
81 (id, name)
82 }
83 _ => {
84 let id = raw.trim().trim_matches('"').to_string();
85 (id.clone(), id)
86 }
87 }
88}
89
90pub struct Notmuch {
91 paths: Arc<MailPaths>,
92 write_lock: Mutex<()>,
93 parsed: crate::cache::FileCache<Arc<crate::mime::ParsedMessage>>,
96 files: crate::cache::FileCache<PathBuf>,
97}
98
99impl Notmuch {
100 pub fn new(paths: Arc<MailPaths>) -> Self {
101 Self {
102 paths,
103 write_lock: Mutex::new(()),
104 parsed: crate::cache::FileCache::new(256),
105 files: crate::cache::FileCache::new(2048),
106 }
107 }
108
109 pub fn paths(&self) -> &MailPaths {
110 &self.paths
111 }
112
113 fn command(&self) -> Command {
114 let mut command = Command::new(&self.paths.binaries.notmuch);
115 if let Some(config) = &self.paths.notmuch.path {
116 command.arg(format!("--config={}", config.display()));
117 command.env("NOTMUCH_CONFIG", config);
118 }
119 command
120 }
121
122 async fn run(&self, args: &[&str]) -> Result<String> {
123 let output = self
124 .command()
125 .args(args)
126 .output()
127 .await
128 .map_err(|e| match e.kind() {
129 std::io::ErrorKind::NotFound => Error::ToolMissing {
130 tool: crate::tools::NOTMUCH,
131 },
132 _ => Error::Io(e),
133 })?;
134
135 if !output.status.success() {
136 return Err(Error::ToolFailed {
137 tool: crate::tools::NOTMUCH,
138 stderr: String::from_utf8_lossy(&output.stderr).trim().to_string(),
139 });
140 }
141 Ok(String::from_utf8_lossy(&output.stdout).into_owned())
142 }
143
144 async fn run_json<T: serde::de::DeserializeOwned>(&self, args: &[&str]) -> Result<T> {
145 let stdout = self.run(args).await?;
146 serde_json::from_str(&stdout).map_err(|e| Error::ToolFailed {
147 tool: crate::tools::NOTMUCH,
148 stderr: format!("could not parse `notmuch {}` output: {e}", args.join(" ")),
149 })
150 }
151
152 pub async fn revision(&self) -> Result<Revision> {
153 let stdout = self.run(&["count", "--lastmod", "*"]).await?;
154 parse_lastmod(&stdout)
155 }
156
157 pub async fn revision_and_total(&self) -> Result<(Revision, u64)> {
161 let stdout = self
162 .run(&["count", "--lastmod", "--exclude=false", "*"])
163 .await?;
164 let revision = parse_lastmod(&stdout)?;
165 let total = stdout
166 .split_whitespace()
167 .next()
168 .and_then(|c| c.parse().ok())
169 .ok_or_else(|| Error::ToolFailed {
170 tool: crate::tools::NOTMUCH,
171 stderr: format!("unexpected `notmuch count --lastmod` output: {stdout:?}"),
172 })?;
173
174 Ok((revision, total))
175 }
176
177 pub async fn messages_between(&self, from: u64, to: u64) -> Result<Vec<Message>> {
191 let query = format!("lastmod:{from}..{to}");
192
193 let output: json::ShowOutput = self
194 .run_json(&[
195 "show",
196 "--format=json",
197 "--body=false",
198 "--entire-thread=false",
199 "--exclude=false",
200 &query,
201 ])
202 .await?;
203
204 let threads = self.thread_of_each_message(&query).await?;
205
206 Ok(output
207 .flatten()
208 .into_iter()
209 .filter_map(|m| m.into_message())
210 .map(|mut message| {
211 if let Some(thread) = threads.get(message.id.as_str()) {
212 message.thread_id = thread.clone();
213 }
214 message
215 })
216 .collect())
217 }
218
219 async fn thread_of_each_message(&self, query: &str) -> Result<HashMap<String, ThreadId>> {
222 let items: Vec<json::SearchItem> = self
223 .run_json(&[
224 "search",
225 "--format=json",
226 "--output=summary",
227 "--exclude=false",
228 query,
229 ])
230 .await?;
231
232 let mut out = HashMap::new();
233 for item in items {
234 let thread = ThreadId(item.thread.clone());
235 for id in json::matched_ids(&item) {
236 out.insert(id, thread.clone());
237 }
238 }
239 Ok(out)
240 }
241
242 pub async fn count(&self, query: &Query) -> Result<usize> {
243 let stdout = self.run(&["count", query.effective_text()]).await?;
244 stdout.trim().parse().map_err(|_| Error::ToolFailed {
245 tool: crate::tools::NOTMUCH,
246 stderr: format!("`notmuch count` returned {stdout:?}"),
247 })
248 }
249
250 pub async fn count_batch(&self, queries: &[String]) -> Result<Vec<u64>> {
261 use tokio::io::AsyncWriteExt;
262
263 if queries.is_empty() {
264 return Ok(Vec::new());
265 }
266
267 const NOTHING: &str = "tag:__ecr_matches_nothing__";
268 let batch: String = queries
269 .iter()
270 .map(|q| {
271 let line = q.trim();
272 if line.is_empty() {
273 NOTHING
274 } else {
275 line
276 }
277 })
278 .collect::<Vec<_>>()
279 .join("\n");
280
281 let mut child = self
282 .command()
283 .args(["count", "--batch"])
284 .stdin(std::process::Stdio::piped())
285 .stdout(std::process::Stdio::piped())
286 .stderr(std::process::Stdio::piped())
287 .spawn()
288 .map_err(|e| match e.kind() {
289 std::io::ErrorKind::NotFound => Error::ToolMissing {
290 tool: crate::tools::NOTMUCH,
291 },
292 _ => Error::Io(e),
293 })?;
294
295 if let Some(mut stdin) = child.stdin.take() {
296 stdin.write_all(batch.as_bytes()).await?;
297 stdin.write_all(b"\n").await?;
298 stdin.shutdown().await?;
299 }
300
301 let output = child.wait_with_output().await?;
302 if !output.status.success() {
303 return Err(Error::ToolFailed {
304 tool: crate::tools::NOTMUCH,
305 stderr: String::from_utf8_lossy(&output.stderr).trim().to_string(),
306 });
307 }
308
309 let stdout = String::from_utf8_lossy(&output.stdout);
310 let counts: Vec<u64> = stdout
311 .lines()
312 .filter(|line| !line.trim().is_empty())
313 .map(|line| line.trim().parse().unwrap_or(0))
314 .collect();
315
316 if counts.len() != queries.len() {
318 return Err(Error::ToolFailed {
319 tool: crate::tools::NOTMUCH,
320 stderr: format!(
321 "`notmuch count --batch` answered {} of {} queries",
322 counts.len(),
323 queries.len()
324 ),
325 });
326 }
327
328 Ok(counts)
329 }
330
331 pub async fn search_threads(&self, query: &Query) -> Result<Vec<ThreadSummary>> {
332 let limit = format!("--limit={}", query.limit);
333 let offset = format!("--offset={}", query.offset);
334 let items: Vec<json::SearchItem> = self
335 .run_json(&[
336 "search",
337 "--format=json",
338 "--output=summary",
339 "--sort=newest-first",
340 &limit,
341 &offset,
342 query.effective_text(),
343 ])
344 .await?;
345
346 Ok(items
347 .into_iter()
348 .map(json::SearchItem::into_summary)
349 .collect())
350 }
351
352 pub async fn thread(&self, id: &ThreadId) -> Result<Thread> {
353 let messages = self.show(&id.query()).await?;
354 let subject = messages
355 .first()
356 .map(|m| m.subject.clone())
357 .unwrap_or_default();
358
359 Ok(Thread {
360 id: id.clone(),
361 subject,
362 messages,
363 })
364 }
365
366 pub async fn message(&self, id: &MessageId) -> Result<Message> {
367 self.show(&id.query())
368 .await?
369 .into_iter()
370 .next()
371 .ok_or_else(|| Error::MessageNotFound { id: id.to_string() })
372 }
373
374 async fn show(&self, query: &str) -> Result<Vec<Message>> {
375 let output: json::ShowOutput = self
376 .run_json(&[
377 "show",
378 "--format=json",
379 "--body=false",
380 "--entire-thread=true",
381 query,
382 ])
383 .await?;
384
385 let mut messages: Vec<Message> = output
386 .flatten()
387 .into_iter()
388 .filter_map(|m| m.into_message())
389 .collect();
390
391 messages.sort_by_key(|m| m.timestamp);
392 Ok(messages)
393 }
394
395 pub async fn message_file(&self, id: &MessageId) -> Result<PathBuf> {
396 if let Some(cached) = self.files.get(id.as_str(), None) {
399 if cached.is_file() {
400 return Ok(cached);
401 }
402 }
403
404 let stdout = self
405 .run(&["search", "--output=files", "--format=text", &id.query()])
406 .await?;
407
408 let path = stdout
409 .lines()
410 .map(|l| PathBuf::from(l.trim()))
411 .find(|p| p.is_file())
412 .ok_or_else(|| Error::MessageNotFound { id: id.to_string() })?;
413
414 self.files.insert(id.0.clone(), None, path.clone());
415 Ok(path)
416 }
417
418 pub async fn parsed(&self, id: &MessageId) -> Result<Arc<crate::mime::ParsedMessage>> {
419 let path = self.message_file(id).await?;
420 let modified = crate::cache::modified_at(&path);
421 let key = path.to_string_lossy().into_owned();
422
423 if let Some(cached) = self.parsed.get(&key, modified) {
424 return Ok(cached);
425 }
426
427 let raw = tokio::fs::read(&path).await?;
428 let parsed = Arc::new(crate::mime::parse(id.as_str(), &raw)?);
429
430 self.parsed.insert(key, modified, Arc::clone(&parsed));
431 Ok(parsed)
432 }
433
434 pub async fn message_with_parts(&self, id: &MessageId) -> Result<Message> {
435 let mut message = self.message(id).await?;
436 message.parts = self.parsed(id).await?.parts();
437 Ok(message)
438 }
439
440 pub async fn body(
441 &self,
442 id: &MessageId,
443 format: BodyFormat,
444 allow_remote_resources: bool,
445 ) -> Result<Body> {
446 let parsed = self.parsed(id).await?;
447 let ctx = crate::mime::SanitizeContext::new(
448 format!("/api/v1/messages/{id}/parts/"),
449 allow_remote_resources,
450 );
451 Ok(parsed.body(format, &ctx))
452 }
453
454 pub async fn part(&self, id: &MessageId, part: &PartId) -> Result<Part> {
455 self.parsed(id)
456 .await?
457 .part(part)
458 .ok_or_else(|| Error::PartNotFound {
459 id: id.to_string(),
460 part: part.0,
461 })
462 }
463
464 pub async fn address_book(&self, limit: usize) -> Result<crate::address::AddressBook> {
467 use crate::address::{AddressBook, Source};
468
469 let mut book = AddressBook::new();
470
471 let recipients = self
472 .run(&[
473 "address",
474 "--output=recipients",
475 "--deduplicate=address",
476 "tag:sent or tag:draft",
477 ])
478 .await
479 .unwrap_or_default();
480 book.add_lines(&recipients, Source::Recipient);
481
482 let senders = self
483 .run(&["address", "--output=sender", "--deduplicate=address", "*"])
484 .await
485 .unwrap_or_default();
486 book.add_lines(&senders, Source::Sender);
487
488 let _ = limit;
489 Ok(book)
490 }
491
492 pub async fn mailing_lists(&self, scan: usize) -> Result<Vec<MailingList>> {
500 let files = self
501 .run(&[
502 "search",
503 "--output=files",
504 "--limit",
505 &scan.to_string(),
506 "*",
507 ])
508 .await
509 .unwrap_or_default();
510
511 let paths: Vec<PathBuf> = files.lines().map(PathBuf::from).collect();
512
513 let lists = tokio::task::spawn_blocking(move || {
515 let mut seen: HashMap<String, MailingList> = HashMap::new();
516
517 for path in paths {
518 let Some(raw) = read_list_id(&path) else {
519 continue;
520 };
521 let (id, name) = split_list_id(&raw);
522 let entry =
523 seen.entry(id.clone())
524 .or_insert_with(|| MailingList { id, name, count: 0 });
525 entry.count += 1;
526 }
527
528 let mut lists: Vec<MailingList> = seen.into_values().collect();
529 lists.sort_by(|a, b| b.count.cmp(&a.count).then_with(|| a.name.cmp(&b.name)));
530 lists
531 })
532 .await
533 .map_err(|e| Error::ToolFailed {
534 tool: crate::tools::NOTMUCH,
535 stderr: format!("scanning for mailing lists panicked: {e}"),
536 })?;
537
538 Ok(lists)
539 }
540
541 pub async fn indexes_list_id(&self) -> bool {
545 self.run(&["config", "get", "index.header.List"])
546 .await
547 .map(|value| value.trim().eq_ignore_ascii_case("List-Id"))
548 .unwrap_or(false)
549 }
550
551 pub async fn tags(&self) -> Result<Vec<String>> {
553 let stdout = self.run(&["search", "--output=tags", "*"]).await?;
554 Ok(stdout
555 .lines()
556 .map(|l| l.trim().to_string())
557 .filter(|l| !l.is_empty())
558 .collect())
559 }
560
561 pub async fn tag(&self, ops: &[TagOp]) -> Result<Revision> {
562 self.tag_batch(&batch::build(ops)?).await
563 }
564
565 pub async fn tag_batch(&self, batch: &str) -> Result<Revision> {
566 use tokio::io::AsyncWriteExt;
567
568 if batch.trim().is_empty() {
569 return self.revision().await;
570 }
571
572 let _guard = self.write_lock.lock().await;
573
574 let mut child = self
575 .command()
576 .args(["tag", "--batch"])
577 .stdin(std::process::Stdio::piped())
578 .stdout(std::process::Stdio::piped())
579 .stderr(std::process::Stdio::piped())
580 .spawn()
581 .map_err(|e| match e.kind() {
582 std::io::ErrorKind::NotFound => Error::ToolMissing {
583 tool: crate::tools::NOTMUCH,
584 },
585 _ => Error::Io(e),
586 })?;
587
588 if let Some(mut stdin) = child.stdin.take() {
589 stdin.write_all(batch.as_bytes()).await?;
590 stdin.shutdown().await?;
591 }
592
593 let output = child.wait_with_output().await?;
594 if !output.status.success() {
595 return Err(Error::ToolFailed {
596 tool: crate::tools::NOTMUCH,
597 stderr: String::from_utf8_lossy(&output.stderr).trim().to_string(),
598 });
599 }
600
601 drop(_guard);
602 self.revision().await
603 }
604
605 pub async fn index_new(&self) -> Result<Revision> {
606 let _guard = self.write_lock.lock().await;
607 self.run(&["new", "--quiet"]).await?;
608 drop(_guard);
609 self.revision().await
610 }
611}
612
613fn parse_lastmod(stdout: &str) -> Result<Revision> {
614 let line = stdout.trim();
615 let mut fields = line.split_whitespace();
616 let (_count, uuid, lastmod) = (fields.next(), fields.next(), fields.next());
617
618 match (uuid, lastmod) {
619 (Some(uuid), Some(lastmod)) => lastmod
620 .parse()
621 .map(|lastmod| Revision::new(uuid, lastmod))
622 .map_err(|_| Error::ToolFailed {
623 tool: crate::tools::NOTMUCH,
624 stderr: format!("unparseable lastmod in {line:?}"),
625 }),
626 _ => Err(Error::ToolFailed {
627 tool: crate::tools::NOTMUCH,
628 stderr: format!("unexpected `notmuch count --lastmod` output: {line:?}"),
629 }),
630 }
631}
632
633pub fn parse_address_list(raw: &str) -> Vec<Address> {
634 let mut out = Vec::new();
635 let mut current = String::new();
636 let mut in_quotes = false;
637 let mut in_angle = false;
638
639 for ch in raw.chars() {
640 match ch {
641 '"' => {
642 in_quotes = !in_quotes;
643 current.push(ch);
644 }
645 '<' if !in_quotes => {
646 in_angle = true;
647 current.push(ch);
648 }
649 '>' if !in_quotes => {
650 in_angle = false;
651 current.push(ch);
652 }
653 ',' if !in_quotes && !in_angle => {
654 push_address(&mut out, ¤t);
655 current.clear();
656 }
657 _ => current.push(ch),
658 }
659 }
660 push_address(&mut out, ¤t);
661 out
662}
663
664fn push_address(out: &mut Vec<Address>, raw: &str) {
665 let raw = raw.trim();
666 if raw.is_empty() {
667 return;
668 }
669
670 match (raw.find('<'), raw.rfind('>')) {
671 (Some(start), Some(end)) if end > start => {
672 let name = raw[..start].trim().trim_matches('"').trim();
673 let email = raw[start + 1..end].trim();
674 if !email.is_empty() {
675 out.push(Address::new(
676 (!name.is_empty()).then(|| name.to_string()),
677 email,
678 ));
679 }
680 }
681 _ => out.push(Address::new(None, raw)),
682 }
683}
684
685#[cfg(test)]
686mod tests {
687 use super::*;
688
689 #[test]
690 fn parses_the_live_lastmod_line() {
691 let rev = parse_lastmod("45865\tc92ee515-acf5-452e-a148-5941a7e9852f\t227965\n").unwrap();
692 assert_eq!(rev.uuid, "c92ee515-acf5-452e-a148-5941a7e9852f");
693 assert_eq!(rev.lastmod, 227965);
694 }
695
696 #[test]
697 fn rejects_unexpected_lastmod_output() {
698 assert!(parse_lastmod("").is_err());
699 assert!(parse_lastmod("45865").is_err());
700 assert!(parse_lastmod("45865\tuuid\tnot-a-number").is_err());
701 }
702
703 #[test]
704 fn parses_a_plain_address() {
705 let addrs = parse_address_list("alice@example.com");
706 assert_eq!(addrs.len(), 1);
707 assert_eq!(addrs[0].email, "alice@example.com");
708 assert_eq!(addrs[0].name, None);
709 }
710
711 #[test]
712 fn parses_a_named_address() {
713 let addrs = parse_address_list("Google <no-reply@accounts.google.com>");
714 assert_eq!(addrs[0].name.as_deref(), Some("Google"));
715 assert_eq!(addrs[0].email, "no-reply@accounts.google.com");
716 }
717
718 #[test]
719 fn a_comma_inside_a_quoted_name_does_not_split_the_list() {
720 let addrs = parse_address_list("\"Doe, Jane\" <a@b.c>, Other <d@e.f>");
721 assert_eq!(addrs.len(), 2);
722 assert_eq!(addrs[0].name.as_deref(), Some("Doe, Jane"));
723 assert_eq!(addrs[0].email, "a@b.c");
724 assert_eq!(addrs[1].email, "d@e.f");
725 }
726
727 #[test]
728 fn an_empty_header_yields_no_addresses() {
729 assert!(parse_address_list("").is_empty());
730 assert!(parse_address_list(" , ").is_empty());
731 }
732
733 #[test]
734 fn a_list_id_splits_into_its_name_and_its_id() {
735 let (id, name) = split_list_id("Emacs development <emacs-devel.gnu.org>");
736 assert_eq!(id, "emacs-devel.gnu.org");
737 assert_eq!(name, "Emacs development");
738 }
739
740 #[test]
741 fn a_bare_list_id_is_its_own_name() {
742 let (id, name) = split_list_id("<numpy-discussion.python.org>");
743 assert_eq!(id, "numpy-discussion.python.org");
744 assert_eq!(name, "numpy-discussion.python.org");
745
746 let (id, name) = split_list_id("mu-discuss.googlegroups.com");
747 assert_eq!(id, "mu-discuss.googlegroups.com");
748 assert_eq!(name, "mu-discuss.googlegroups.com");
749 }
750
751 #[test]
752 fn a_quoted_list_name_loses_its_quotes() {
753 let (_, name) = split_list_id("\"Culture STIC\" <culture.stic.fr>");
754 assert_eq!(name, "Culture STIC");
755 }
756
757 fn message_with(headers: &str) -> tempfile::NamedTempFile {
758 let file = tempfile::NamedTempFile::new().unwrap();
759 std::fs::write(file.path(), format!("{headers}\n\nThe body.\n")).unwrap();
760 file
761 }
762
763 #[test]
764 fn reads_the_list_id_header() {
765 let file = message_with("From: a@b.c\nList-Id: Emacs <emacs-devel.gnu.org>\nSubject: hi");
766 assert_eq!(
767 read_list_id(file.path()).as_deref(),
768 Some("Emacs <emacs-devel.gnu.org>")
769 );
770 }
771
772 #[test]
773 fn the_header_name_is_matched_regardless_of_case() {
774 let file = message_with("LIST-ID: <x.example.com>");
775 assert_eq!(
776 read_list_id(file.path()).as_deref(),
777 Some("<x.example.com>")
778 );
779 }
780
781 #[test]
782 fn a_folded_list_id_is_rejoined() {
783 let file = message_with("List-Id: A very long list name\n <long.example.com>");
784 assert_eq!(
785 read_list_id(file.path()).as_deref(),
786 Some("A very long list name <long.example.com>")
787 );
788 }
789
790 #[test]
791 fn a_message_without_the_header_yields_nothing() {
792 let file = message_with("From: a@b.c\nSubject: hi");
793 assert_eq!(read_list_id(file.path()), None);
794 }
795
796 #[test]
798 fn the_scan_stops_at_the_end_of_the_headers() {
799 let file = tempfile::NamedTempFile::new().unwrap();
800 std::fs::write(
801 file.path(),
802 "From: a@b.c\n\nList-Id: <not-a-header.example.com>\n",
803 )
804 .unwrap();
805 assert_eq!(read_list_id(file.path()), None);
806 }
807}