ecr_store/index/
mod.rs

1//! A SQLite mirror of what notmuch knows about each message.
2//!
3//! Every request otherwise costs a notmuch process. The mirror carries message
4//! metadata — id, thread, timestamp, subject, sender, tags — and answers the
5//! queries it can prove it answers *identically*: tags, ids, threads, `*`, and
6//! booleans of those. That is every mailbox the sidebar offers and every count
7//! beside them. Anything else, and any failure at all, falls through to notmuch
8//! — see [`plan`], which also records why text search is not on the list.
9//!
10//! It carries no words, so it is small: a 46k-message maildir is about 9MB.
11//!
12//! notmuch remains the only writer of mail state. Nothing in here is a source
13//! of truth; the file can be deleted at any point and is rebuilt on the next
14//! refresh.
15
16mod freshness;
17mod plan;
18mod relative;
19mod schema;
20mod sync;
21
22use crate::error::Result;
23use crate::paths::MailPaths;
24use ecr_core::message::{Message, Query, ThreadId, ThreadSummary};
25use ecr_core::revision::Revision;
26use rusqlite::{Connection, OptionalExtension};
27use std::collections::{BTreeMap, BTreeSet};
28use std::path::{Path, PathBuf};
29use std::sync::Mutex;
30
31pub use freshness::Freshness;
32pub use sync::{refresh, refresh_incremental, Refreshed};
33
34/// What the index holds, for `ecr doctor`.
35#[derive(Debug, Clone, PartialEq, Eq)]
36pub struct IndexStatus {
37    pub path: Option<PathBuf>,
38    pub revision: Option<Revision>,
39    pub messages: u64,
40    pub bytes: u64,
41}
42
43pub struct MessageIndex {
44    conn: Mutex<Connection>,
45    path: Option<PathBuf>,
46    exclude_tags: Vec<String>,
47}
48
49impl MessageIndex {
50    pub fn file_name() -> &'static str {
51        "index.sqlite3"
52    }
53
54    pub fn path_for(paths: &MailPaths) -> PathBuf {
55        paths.ecr_state_dir.join(Self::file_name())
56    }
57
58    pub fn open(paths: &MailPaths) -> Result<Self> {
59        let path = Self::path_for(paths);
60        if let Some(parent) = path.parent() {
61            std::fs::create_dir_all(parent)?;
62        }
63        Self::open_at(&path, paths.notmuch_config.exclude_tags.clone())
64    }
65
66    pub fn open_at(path: &Path, exclude_tags: Vec<String>) -> Result<Self> {
67        let conn = Connection::open(path)?;
68        schema::prepare(&conn)?;
69
70        Ok(Self {
71            conn: Mutex::new(conn),
72            path: Some(path.to_path_buf()),
73            exclude_tags,
74        })
75    }
76
77    pub fn in_memory(exclude_tags: Vec<String>) -> Result<Self> {
78        let conn = Connection::open_in_memory()?;
79        schema::prepare(&conn)?;
80
81        Ok(Self {
82            conn: Mutex::new(conn),
83            path: None,
84            exclude_tags,
85        })
86    }
87
88    fn with<T>(&self, f: impl FnOnce(&Connection) -> Result<T>) -> Result<T> {
89        let conn = self
90            .conn
91            .lock()
92            .map_err(|_| crate::Error::Index("the index lock was poisoned".into()))?;
93        f(&conn)
94    }
95
96    pub fn revision(&self) -> Result<Option<Revision>> {
97        self.with(|conn| {
98            let uuid: Option<String> = meta(conn, "uuid")?;
99            let lastmod: Option<String> = meta(conn, "lastmod")?;
100            Ok(match (uuid, lastmod.and_then(|v| v.parse().ok())) {
101                (Some(uuid), Some(lastmod)) => Some(Revision { uuid, lastmod }),
102                _ => None,
103            })
104        })
105    }
106
107    pub fn message_count(&self) -> Result<u64> {
108        self.with(|conn| {
109            let count: i64 =
110                conn.query_row("SELECT COUNT(*) FROM messages", [], |row| row.get(0))?;
111            Ok(count as u64)
112        })
113    }
114
115    pub fn status(&self) -> IndexStatus {
116        IndexStatus {
117            path: self.path.clone(),
118            revision: self.revision().ok().flatten(),
119            messages: self.message_count().unwrap_or(0),
120            bytes: self
121                .path
122                .as_ref()
123                .and_then(|p| std::fs::metadata(p).ok())
124                .map(|m| m.len())
125                .unwrap_or(0),
126        }
127    }
128
129    /// `None` when the query is not one the index can answer identically.
130    pub fn search_threads(&self, query: &Query) -> Result<Option<Vec<ThreadSummary>>> {
131        let Some(plan) = plan::plan(query.effective_text(), &self.exclude_tags) else {
132            return Ok(None);
133        };
134
135        self.with(|conn| {
136            let page = thread_page(conn, &plan, query.limit, query.offset)?;
137            if page.is_empty() {
138                return Ok(Some(Vec::new()));
139            }
140            Ok(Some(summaries(conn, &plan, page)?))
141        })
142    }
143
144    pub fn count(&self, text: &str) -> Result<Option<u64>> {
145        let Some(plan) = plan::plan(text, &self.exclude_tags) else {
146            return Ok(None);
147        };
148
149        self.with(|conn| {
150            let sql = format!(
151                "SELECT COUNT(*) FROM messages m WHERE {}",
152                plan.predicate(plan::Shape::Set)
153            );
154            let count: i64 =
155                conn.query_row(&sql, rusqlite::params_from_iter(&plan.params), |row| {
156                    row.get(0)
157                })?;
158            Ok(Some(count as u64))
159        })
160    }
161
162    /// Replaces what the index holds for these messages, in one transaction, and
163    /// moves the watermark with them. A refresh interrupted halfway therefore
164    /// resumes from the last chunk that landed rather than starting over.
165    pub fn apply(&self, messages: &[Message], revision: &Revision) -> Result<()> {
166        self.with(|conn| {
167            let tx = conn.unchecked_transaction()?;
168            for message in messages {
169                upsert(&tx, message)?;
170            }
171            set_meta(&tx, "uuid", &revision.uuid)?;
172            set_meta(&tx, "lastmod", &revision.lastmod.to_string())?;
173            tx.commit()?;
174            Ok(())
175        })
176    }
177
178    pub fn clear(&self) -> Result<()> {
179        self.with(schema::reset)
180    }
181}
182
183fn meta(conn: &Connection, key: &str) -> Result<Option<String>> {
184    Ok(conn
185        .query_row("SELECT value FROM meta WHERE key = ?", [key], |row| {
186            row.get(0)
187        })
188        .optional()?)
189}
190
191fn set_meta(conn: &Connection, key: &str, value: &str) -> Result<()> {
192    conn.execute(
193        "INSERT INTO meta (key, value) VALUES (?, ?)
194         ON CONFLICT (key) DO UPDATE SET value = excluded.value",
195        [key, value],
196    )?;
197    Ok(())
198}
199
200fn upsert(conn: &Connection, message: &Message) -> Result<()> {
201    let author = message
202        .from
203        .first()
204        .map(|a| a.display().to_string())
205        .unwrap_or_default();
206
207    let num: i64 = conn.query_row(
208        "INSERT INTO messages (id, thread, timestamp, subject, author)
209              VALUES (?, ?, ?, ?, ?)
210         ON CONFLICT (id) DO UPDATE SET
211              thread = excluded.thread,
212              timestamp = excluded.timestamp,
213              subject = excluded.subject,
214              author = excluded.author
215         RETURNING num",
216        rusqlite::params![
217            message.id.as_str(),
218            message.thread_id.as_str(),
219            message.timestamp,
220            message.subject,
221            author,
222        ],
223        |row| row.get(0),
224    )?;
225
226    conn.execute("DELETE FROM tags WHERE message = ?", [num])?;
227    let mut insert = conn.prepare_cached("INSERT INTO tags (message, tag) VALUES (?, ?)")?;
228    for tag in &message.tags {
229        insert.execute(rusqlite::params![num, tag])?;
230    }
231
232    Ok(())
233}
234
235struct Page {
236    thread: String,
237    timestamp: i64,
238}
239
240/// The page of threads, newest first.
241///
242/// A thread's sort key is its newest *matched* message, so walking matched
243/// messages newest-first and keeping each thread the first time it appears
244/// produces exactly that order — with no aggregation, and without looking
245/// beyond the page. The obvious `GROUP BY thread ORDER BY MAX(timestamp)` is
246/// the same answer computed over every match in the database: 94ms against a
247/// 46k inbox where this is 2ms, because a page of fifty is fifty rows of work
248/// either way and the aggregate does not know that.
249///
250/// Rows are pulled lazily, so stopping is free and no limit has to be guessed
251/// in advance — a page of fifty threads may take fifty rows or five hundred,
252/// depending on how many messages each thread has.
253fn thread_page(
254    conn: &Connection,
255    plan: &plan::Plan,
256    limit: usize,
257    offset: usize,
258) -> Result<Vec<Page>> {
259    let wanted = offset.saturating_add(limit);
260    if wanted == 0 {
261        return Ok(Vec::new());
262    }
263
264    let sql = format!(
265        "SELECT m.thread, m.timestamp
266           FROM messages m
267          WHERE {}
268          ORDER BY m.timestamp DESC",
269        plan.predicate(plan::Shape::Scan)
270    );
271
272    let mut statement = conn.prepare(&sql)?;
273    let mut rows = statement.query(rusqlite::params_from_iter(&plan.params))?;
274
275    let mut seen: BTreeSet<String> = BTreeSet::new();
276    let mut found: Vec<Page> = Vec::new();
277
278    while let Some(row) = rows.next()? {
279        let timestamp: i64 = row.get(1)?;
280
281        // Two threads whose newest matched message shares a timestamp are
282        // ordered by thread id, and no ordering the timestamp index can supply
283        // does that — asking SQL for it costs a sort of every match. Instead
284        // the walk runs on past the page to the end of the timestamp it ended
285        // in, so the whole tied group is in hand before it is ordered. Rows
286        // arrive in timestamp order, so a group is always contiguous.
287        if found.len() >= wanted && found.last().is_some_and(|p| p.timestamp != timestamp) {
288            break;
289        }
290
291        let thread: String = row.get(0)?;
292        if seen.insert(thread.clone()) {
293            found.push(Page { thread, timestamp });
294        }
295    }
296
297    found.sort_by(|a, b| b.timestamp.cmp(&a.timestamp).then(a.thread.cmp(&b.thread)));
298
299    Ok(found.into_iter().skip(offset).take(limit).collect())
300}
301
302struct Row {
303    thread: String,
304    id: String,
305    subject: String,
306    author: String,
307    matched: bool,
308}
309
310/// notmuch's thread subject is the **newest matched** message's, with one
311/// leading `Re: ` removed once — never repeatedly, and nothing else removed:
312/// `Fw: `, `Fwd: ` and `Undeliverable: ` all stay.
313///
314/// Every word of that had to be measured, and three plausible readings are
315/// wrong. *Newest*, because the search is sorted newest-first and notmuch names
316/// the thread after the message it reached first — "the subject it started
317/// with" is the intuitive rule and it disagrees on any thread that was renamed,
318/// bounced, or forwarded onward. *Matched*, because the same thread answers
319/// differently to different queries; a query that excludes the newest message
320/// names the thread after the newest one that survived. And the header is
321/// otherwise passed through, so a subject that arrives padded keeps its
322/// trailing spaces — tidying it here is a disagreement, not an improvement.
323///
324/// Fixtures cannot catch any of this: a reply to `X` is `Re: X`, so every rule
325/// produces the same string. It was settled by running both paths over a real
326/// 46k maildir and comparing 1,907 thread rows, which is the only thing that
327/// distinguishes them.
328fn thread_subject(newest: &str) -> String {
329    match newest.get(..4) {
330        Some(head) if head.eq_ignore_ascii_case("re: ") => newest[4..].to_string(),
331        _ => newest.to_string(),
332    }
333}
334
335fn summaries(conn: &Connection, plan: &plan::Plan, page: Vec<Page>) -> Result<Vec<ThreadSummary>> {
336    let threads: Vec<&str> = page.iter().map(|p| p.thread.as_str()).collect();
337    let holes = vec!["?"; threads.len()].join(", ");
338
339    // Ordered oldest-first: the thread's subject is its oldest message's, and
340    // notmuch lists authors in the order they wrote.
341    //
342    // The predicate is a *selected expression* rather than a subquery the rows
343    // are tested against, so it is evaluated only for the messages of these
344    // fifty threads. Written as `num IN (SELECT … WHERE <predicate>)` it is
345    // instead evaluated across the whole table to build the set first.
346    let sql = format!(
347        "SELECT m.thread, m.id, m.subject, m.author, ({})
348           FROM messages m
349          WHERE m.thread IN ({holes})
350          ORDER BY m.thread, m.timestamp, m.num",
351        plan.predicate(plan::Shape::Scan)
352    );
353
354    let params = plan
355        .params
356        .iter()
357        .map(|p| p.as_str())
358        .chain(threads.iter().copied())
359        .collect::<Vec<&str>>();
360
361    let mut statement = conn.prepare(&sql)?;
362    let rows = statement
363        .query_map(rusqlite::params_from_iter(params), |row| {
364            Ok(Row {
365                thread: row.get(0)?,
366                id: row.get(1)?,
367                subject: row.get(2)?,
368                author: row.get(3)?,
369                matched: row.get(4)?,
370            })
371        })?
372        .collect::<rusqlite::Result<Vec<_>>>()?;
373
374    let mut by_thread: BTreeMap<String, Vec<Row>> = BTreeMap::new();
375    for row in rows {
376        by_thread.entry(row.thread.clone()).or_default().push(row);
377    }
378
379    let tags = thread_tags(conn, &threads, &holes)?;
380    let now = relative::now();
381
382    Ok(page
383        .into_iter()
384        .map(|entry| {
385            let rows = by_thread.remove(&entry.thread).unwrap_or_default();
386            let (matched, unmatched): (Vec<&Row>, Vec<&Row>) =
387                rows.iter().partition(|row| row.matched);
388
389            ThreadSummary {
390                id: ThreadId(entry.thread.clone()),
391                subject: matched
392                    .last()
393                    .map(|r| thread_subject(&r.subject))
394                    .unwrap_or_default(),
395                authors: authors(&matched, &unmatched),
396                timestamp: entry.timestamp,
397                date_relative: relative::describe(entry.timestamp, now),
398                matched: matched.len(),
399                total: rows.len(),
400                tags: tags.get(&entry.thread).cloned().unwrap_or_default(),
401                newest_message: matched.last().map(|row| row.id.as_str().into()),
402            }
403        })
404        .collect())
405}
406
407/// notmuch's thread tags are the union over every message in the thread,
408/// matched or not — a thread carrying one deleted message reports `deleted`
409/// even when the query excluded it.
410fn thread_tags(
411    conn: &Connection,
412    threads: &[&str],
413    holes: &str,
414) -> Result<BTreeMap<String, BTreeSet<String>>> {
415    let sql = format!(
416        "SELECT m.thread, t.tag
417           FROM messages m JOIN tags t ON t.message = m.num
418          WHERE m.thread IN ({holes})"
419    );
420
421    let mut statement = conn.prepare(&sql)?;
422    let rows = statement.query_map(rusqlite::params_from_iter(threads), |row| {
423        Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
424    })?;
425
426    let mut out: BTreeMap<String, BTreeSet<String>> = BTreeMap::new();
427    for row in rows {
428        let (thread, tag) = row?;
429        out.entry(thread).or_default().insert(tag);
430    }
431    Ok(out)
432}
433
434/// Matched authors first, then the rest, each deduplicated and in date order.
435///
436/// notmuch renders this as `a, b| c` and the caller splits it back apart on
437/// both separators. The index has the real per-message list and could skip the
438/// round trip — but a name with a comma in it survives that and does not
439/// survive notmuch's, so the two paths would disagree on every message from
440/// `Anthropic, PBC`. It goes through the same lossy rendering deliberately;
441/// see `split_authors`.
442fn authors(matched: &[&Row], unmatched: &[&Row]) -> Vec<String> {
443    fn join(rows: &[&Row], seen: &mut BTreeSet<String>) -> String {
444        rows.iter()
445            .filter(|row| !row.author.is_empty() && seen.insert(row.author.clone()))
446            .map(|row| row.author.as_str())
447            .collect::<Vec<_>>()
448            .join(", ")
449    }
450
451    let mut seen = BTreeSet::new();
452    let rendered = format!(
453        "{}|{}",
454        join(matched, &mut seen),
455        join(unmatched, &mut seen)
456    );
457
458    crate::notmuch::split_authors(&rendered)
459}
460
461#[cfg(test)]
462mod tests;
ecr_store/index/mod.rs

ecr_store/index/
mod.rs
