diff --git a/src/database/connection.rs b/src/database/connection.rs index 22d8d48..60533cb 100644 --- a/src/database/connection.rs +++ b/src/database/connection.rs @@ -7782,38 +7782,75 @@ impl DatabaseManager { Ok(None) } - /// Get all emails that reference a message-id (in in_reply_to or references fields) - pub async fn get_lore_emails_referencing( + /// Emails that reply to any of `message_ids`: their In-Reply-To or + /// References names one of them, as + /// [`crate::text_utils::header_names_message_id`] decides. Each email is + /// returned once. + /// + /// A query tests `references` for every id by literal containment, which + /// no index serves: it scans that column of the whole table and adds a + /// containment pass per id, about 45 ms per id on a table of a million + /// emails. Passing a whole thread level shares the scan among its ids, + /// but the pass per id remains, so the cost grows with the number of ids. + pub async fn get_lore_emails_referencing_any( &self, - message_id: &str, + message_ids: &[String], ) -> Result> { use arrow::array::AsArray; use futures::TryStreamExt; use std::collections::HashSet; - // Escape SQL string literal for pattern matching - let escaped_message_id = message_id.replace("\\", "\\\\").replace("'", "''"); + // A query costs about one scan of `references` plus one containment + // pass per id, so the chunk size bounds the filter's length more + // than it changes the total cost. + const IDS_PER_QUERY: usize = 64; - let table = self.connection.open_table("lore").execute().await?; - - // Query 1: Find emails where in_reply_to matches - let filter1 = format!("in_reply_to = '{}'", escaped_message_id); - let stream1 = table.query().only_if(&filter1).execute().await?; - let batches1: Vec<_> = stream1.try_collect().await?; + let ids: Vec<&str> = message_ids + .iter() + .map(|id| id.trim()) + .filter(|id| !id.is_empty()) + .collect(); + if ids.is_empty() { + return Ok(Vec::new()); + } - // Query 2: Find emails where references contains the message-id - let filter2 = format!("regexp_like(`references`, '{}')", escaped_message_id); - let stream2 = table.query().only_if(&filter2).execute().await?; - let batches2: Vec<_> = stream2.try_collect().await?; + let table = self.connection.open_table("lore").execute().await?; + let mut batches = Vec::new(); + for chunk in ids.chunks(IDS_PER_QUERY) { + // Ids reach the filter only inside SQL string literals, so the + // quote is the one character to escape. In-Reply-To is compared + // whole, bare and with the leading space a folded header leaves; + // `contains` finds References candidates literally, and the + // exact match is decided below. + let literal = |s: &str| format!("'{}'", s.replace('\'', "''")); + let in_reply_to: Vec = chunk + .iter() + .flat_map(|id| [literal(id), literal(&format!(" {id}"))]) + .collect(); + let references: Vec = chunk + .iter() + .map(|id| format!("contains(`references`, {})", literal(id))) + .collect(); + let filter = format!( + "in_reply_to IN ({}) OR {}", + in_reply_to.join(", "), + references.join(" OR ") + ); - // Combine batches from both queries - let mut batches = batches1; - batches.extend(batches2); + let chunk_batches: Vec<_> = table + .query() + .only_if(&filter) + .execute() + .await? + .try_collect() + .await?; + batches.extend(chunk_batches.into_iter().map(|batch| (chunk, batch))); + } let mut emails = Vec::new(); let mut seen_message_ids = HashSet::new(); - for batch in batches { + for (chunk, batch) in batches { if batch.num_rows() == 0 { continue; } @@ -7867,10 +7904,29 @@ impl DatabaseManager { let message_id = message_ids.value(i).to_string(); // Skip if we've already seen this message_id (deduplication) - if !seen_message_ids.insert(message_id.clone()) { + if seen_message_ids.contains(&message_id) { continue; } + // `contains` also matches inside longer ids; keep exact names. + // For every header form the query matches, a row that names + // an id of another chunk is returned by that chunk's query + // too, so its own chunk's ids suffice. + let in_reply_to = + (!in_reply_tos.is_null(i)).then(|| in_reply_tos.value(i).to_string()); + let references = + (!references_list.is_null(i)).then(|| references_list.value(i).to_string()); + let replies = chunk.iter().any(|id| { + [&in_reply_to, &references] + .into_iter() + .flatten() + .any(|header| crate::text_utils::header_names_message_id(header, id)) + }); + if !replies { + continue; + } + seen_message_ids.insert(message_id.clone()); + // Parse JSON symbols array let symbols_json = symbols_list.value(i); let symbols: Vec = serde_json::from_str(symbols_json).unwrap_or_default(); @@ -7881,17 +7937,9 @@ impl DatabaseManager { date: dates.value(i).to_string(), date_timestamp: Self::get_date_timestamp(date_timestamps, dates, i), message_id, - in_reply_to: if in_reply_tos.is_null(i) { - None - } else { - Some(in_reply_tos.value(i).to_string()) - }, + in_reply_to, subject: subjects.value(i).to_string(), - references: if references_list.is_null(i) { - None - } else { - Some(references_list.value(i).to_string()) - }, + references, recipients: recipients_list.value(i).to_string(), body: bodies.value(i).to_string(), symbols, diff --git a/src/lore_writers.rs b/src/lore_writers.rs index c7e7678..cbef44f 100644 --- a/src/lore_writers.rs +++ b/src/lore_writers.rs @@ -3,13 +3,14 @@ //! These are shared between the CLI (with file output) and MCP (with string buffers) use anyhow::Result; -use std::collections::{HashSet, VecDeque}; +use std::collections::HashSet; use std::io::Write; use base64::engine::general_purpose::STANDARD as BASE64; use base64::Engine as _; use crate::search::LoreSearchOptions; +use crate::text_utils::header_names_message_id; use crate::types::LoreEmailInfo; use crate::DatabaseManager; @@ -458,32 +459,7 @@ pub async fn lore_show_thread_to_writer( } // Step 2: Now walk down from root to collect all messages in thread - let mut all_emails = Vec::new(); - let mut seen_message_ids = HashSet::new(); - let mut to_process = VecDeque::new(); - - // Add root email to seen set and processing queue - seen_message_ids.insert(current_email.message_id.clone()); - to_process.push_back(current_email.message_id.clone()); - all_emails.push(current_email); - - // Recursively find all messages that reply to messages in this thread - while let Some(current_message_id) = to_process.pop_front() { - // Find all emails that reference this message - let referencing_emails = db.get_lore_emails_referencing(¤t_message_id).await?; - - for email in referencing_emails { - // Add to seen set immediately to avoid duplicate processing - if !seen_message_ids.insert(email.message_id.clone()) { - // Already seen this message, skip it - continue; - } - - // Add to processing queue and results - to_process.push_back(email.message_id.clone()); - all_emails.push(email); - } - } + let all_emails = collect_thread_descendants(db, current_email).await?; // Sort emails in thread order (respecting reply structure, then by date) let sorted_emails = crate::search::sort_emails_by_thread_order(&all_emails); @@ -539,9 +515,6 @@ pub async fn lore_show_replies_to_writer( options: &LoreSearchOptions<'_>, writer: &mut dyn Write, ) -> Result<()> { - #[allow(unused)] // verbose is used in conditional compilation - use std::collections::{HashSet, VecDeque}; - if !options.mbox_output { writeln!(writer, "Finding all replies to message: {}", message_id)?; } @@ -562,33 +535,8 @@ pub async fn lore_show_replies_to_writer( }; // Collect all descendants (messages that reply to this message or its descendants) - let mut all_emails = Vec::new(); - let mut seen_message_ids = HashSet::new(); - let mut to_process = VecDeque::new(); - - // Start with the root message let root_message_id = root_email.message_id.clone(); - seen_message_ids.insert(root_email.message_id.clone()); - to_process.push_back(root_email.message_id.clone()); - all_emails.push(root_email); - - // Recursively find all messages that reply to messages in this subthread - while let Some(current_message_id) = to_process.pop_front() { - // Find all emails that reference this message - let referencing_emails = db.get_lore_emails_referencing(¤t_message_id).await?; - - for email in referencing_emails { - // Add to seen set immediately to avoid duplicate processing - if !seen_message_ids.insert(email.message_id.clone()) { - // Already seen this message, skip it - continue; - } - - // Add to processing queue and results - to_process.push_back(email.message_id.clone()); - all_emails.push(email); - } - } + let all_emails = collect_thread_descendants(db, root_email).await?; if all_emails.len() == 1 { // Only the root message, no replies @@ -670,6 +618,81 @@ pub async fn lore_show_replies_to_writer( Ok(()) } +/// `root` and every email below it in its thread, `root` first. +/// +/// Replies to replies count, however deep; ancestors of `root` do not. The +/// walk asks for a whole level at once, through +/// [`DatabaseManager::get_lore_emails_referencing_any`]: each lookup scans +/// the `references` column of the whole lore table, and a level shares that +/// scan instead of paying it once per email. Each id still costs a +/// containment pass, so the walk's cost grows with the size of the thread. +pub async fn collect_thread_descendants( + db: &DatabaseManager, + root: LoreEmailInfo, +) -> Result> { + collect_thread_descendants_with(root, async |ids: &[String]| { + db.get_lore_emails_referencing_any(ids).await + }) + .await +} + +/// [`collect_thread_descendants`] with the reply lookup supplied by the +/// caller (tests count lookups through it): `replies_to_any` returns the +/// emails that reply to any of the ids it is given, and is called once per +/// thread level. +/// +/// Within a level, emails are ordered by the message they answer, in the +/// order those messages were found, those naming it in In-Reply-To before +/// those naming it only in References. An email that answers several +/// messages of the level is placed under the first. Each email appears once, +/// by `message_id`. +#[doc(hidden)] +pub async fn collect_thread_descendants_with( + root: LoreEmailInfo, + mut replies_to_any: F, +) -> Result> +where + F: AsyncFnMut(&[String]) -> Result>, +{ + let mut seen_message_ids = HashSet::new(); + seen_message_ids.insert(root.message_id.clone()); + let mut frontier = vec![root.message_id.trim().to_string()]; + let mut all_emails = vec![root]; + + while !frontier.is_empty() { + let mut found: Vec> = replies_to_any(&frontier) + .await? + .into_iter() + .map(Some) + .collect(); + let mut next_frontier = Vec::new(); + + let headers: [fn(&LoreEmailInfo) -> Option<&str>; 2] = [ + |email| email.in_reply_to.as_deref(), + |email| email.references.as_deref(), + ]; + for parent in &frontier { + for header in headers { + for slot in found.iter_mut() { + let Some(email) = slot.take_if(|email| { + header(email).is_some_and(|h| header_names_message_id(h, parent)) + }) else { + continue; + }; + if seen_message_ids.insert(email.message_id.clone()) { + next_frontier.push(email.message_id.trim().to_string()); + all_emails.push(email); + } + } + } + } + + frontier = next_frontier; + } + + Ok(all_emails) +} + /// Get email by message_id (writer version) pub async fn lore_get_by_message_id_to_writer( db: &DatabaseManager, diff --git a/src/search.rs b/src/search.rs index e30737f..c927dcd 100644 --- a/src/search.rs +++ b/src/search.rs @@ -957,7 +957,7 @@ pub async fn lore_show_thread( message_id: &str, verbose: usize, ) -> Result<()> { - use std::collections::{HashSet, VecDeque}; + use std::collections::HashSet; println!("Finding thread root for message: {}", message_id.cyan()); @@ -1005,32 +1005,7 @@ pub async fn lore_show_thread( println!("Found thread root: {}", root_message_id.bright_black()); // Step 2: Now walk down from root to collect all messages in thread - let mut all_emails = Vec::new(); - let mut seen_message_ids = HashSet::new(); - let mut to_process = VecDeque::new(); - - // Add root email to seen set and processing queue - seen_message_ids.insert(current_email.message_id.clone()); - to_process.push_back(current_email.message_id.clone()); - all_emails.push(current_email); - - // Recursively find all messages that reply to messages in this thread - while let Some(current_message_id) = to_process.pop_front() { - // Find all emails that reference this message - let referencing_emails = db.get_lore_emails_referencing(¤t_message_id).await?; - - for email in referencing_emails { - // Add to seen set immediately to avoid duplicate processing - if !seen_message_ids.insert(email.message_id.clone()) { - // Already seen this message, skip it - continue; - } - - // Add to processing queue and results - to_process.push_back(email.message_id.clone()); - all_emails.push(email); - } - } + let all_emails = crate::lore_writers::collect_thread_descendants(db, current_email).await?; // Sort emails in thread order (respecting reply structure, then by date) // This uses a topological sort based on in_reply_to relationships @@ -1089,8 +1064,6 @@ pub async fn lore_show_replies( message_id: &str, verbose: usize, ) -> Result<()> { - use std::collections::{HashSet, VecDeque}; - println!("Finding all replies to message: {}", message_id.cyan()); // Get the starting message @@ -1107,32 +1080,7 @@ pub async fn lore_show_replies( }; // Collect all descendants (messages that reply to this message or its descendants) - let mut all_emails = Vec::new(); - let mut seen_message_ids = HashSet::new(); - let mut to_process = VecDeque::new(); - - // Start with the root message - seen_message_ids.insert(root_email.message_id.clone()); - to_process.push_back(root_email.message_id.clone()); - all_emails.push(root_email); - - // Recursively find all messages that reply to messages in this subthread - while let Some(current_message_id) = to_process.pop_front() { - // Find all emails that reference this message - let referencing_emails = db.get_lore_emails_referencing(¤t_message_id).await?; - - for email in referencing_emails { - // Add to seen set immediately to avoid duplicate processing - if !seen_message_ids.insert(email.message_id.clone()) { - // Already seen this message, skip it - continue; - } - - // Add to processing queue and results - to_process.push_back(email.message_id.clone()); - all_emails.push(email); - } - } + let all_emails = crate::lore_writers::collect_thread_descendants(db, root_email).await?; if all_emails.len() == 1 { // Only the root message, no replies diff --git a/src/text_utils.rs b/src/text_utils.rs index 455b51a..a8ce79c 100644 --- a/src/text_utils.rs +++ b/src/text_utils.rs @@ -74,6 +74,36 @@ pub fn preprocess_code(code: &str) -> String { code.trim().to_string() } +/// Whether `header` names `message_id` as a whole id. +/// +/// Lore stores In-Reply-To and References as they arrived. A folded +/// header keeps a leading space, In-Reply-To can carry a trailing +/// "(X's message of ...)" comment, and References separates ids with +/// spaces, commas or nothing at all (``). So `message_id` (its +/// surrounding whitespace ignored) matches where it stands between those +/// separators, and never inside a longer id: `` is not named by +/// ``, nor `a@x` by `a@x.y`. +pub fn header_names_message_id(header: &str, message_id: &str) -> bool { + let id = message_id.trim(); + let Some(first) = id.chars().next() else { + return false; + }; + let mut from = 0; + while let Some(offset) = header[from..].find(id) { + let start = from + offset; + let end = start + id.len(); + let before = header[..start].chars().next_back(); + let after = header[end..].chars().next(); + if before.is_none_or(|c| c.is_whitespace() || c == ',' || c == '>') + && after.is_none_or(|c| c.is_whitespace() || c == ',' || c == '<') + { + return true; + } + from = start + first.len_utf8(); + } + false +} + #[cfg(test)] mod tests { use super::*; diff --git a/tests/lore_thread.rs b/tests/lore_thread.rs new file mode 100644 index 0000000..db3715d --- /dev/null +++ b/tests/lore_thread.rs @@ -0,0 +1,313 @@ +// SPDX-License-Identifier: MIT OR Apache-2.0 +// +// Thread walks over the lore table: which emails count as replies, and how +// many lookups a walk costs. +// +// A reply names its parent in In-Reply-To or References. Those headers are +// stored as they arrived, so a message-id is matched literally and as a whole +// id: `+` and `.` in it are ordinary characters, and `` is not named by +// ``. Each lookup scans the `references` column of the whole table, so +// a walk asks for one thread level at a time. +use semcode::lore_writers::{collect_thread_descendants, collect_thread_descendants_with}; +use semcode::text_utils::header_names_message_id; +use semcode::{DatabaseManager, LoreEmailInfo}; +use std::cell::Cell; + +async fn database() -> (tempfile::TempDir, DatabaseManager) { + let dir = tempfile::tempdir().unwrap(); + let db = DatabaseManager::new( + dir.path().join(".semcode.db").to_str().unwrap(), + dir.path().to_string_lossy().into_owned(), + ) + .await + .unwrap(); + db.create_tables().await.unwrap(); + (dir, db) +} + +async fn insert(db: &DatabaseManager, emails: &[LoreEmailInfo]) { + let failed = db.insert_lore_emails(emails).await.unwrap(); + assert!(failed.is_empty(), "emails not stored: {failed:?}"); +} + +fn email(message_id: &str, in_reply_to: Option<&str>, references: Option<&str>) -> LoreEmailInfo { + LoreEmailInfo { + git_commit_sha: format!("sha-{message_id}"), + from: "Someone ".to_string(), + date: "Wed, 11 Nov 2026 12:00:00 +0000".to_string(), + date_timestamp: 1_794_398_400, + message_id: message_id.to_string(), + in_reply_to: in_reply_to.map(str::to_string), + subject: format!("subject of {message_id}"), + references: references.map(str::to_string), + recipients: String::new(), + body: String::new(), + symbols: Vec::new(), + } +} + +fn ids(emails: &[LoreEmailInfo]) -> Vec<&str> { + emails.iter().map(|e| e.message_id.as_str()).collect() +} + +fn sorted(mut ids: Vec<&str>) -> Vec<&str> { + ids.sort_unstable(); + ids +} + +async fn root(db: &DatabaseManager, message_id: &str) -> LoreEmailInfo { + db.get_lore_email_by_message_id(message_id) + .await + .unwrap() + .expect("root is stored") +} + +#[tokio::test] +async fn plus_and_dot_in_a_message_id_are_literal() { + let (_dir, db) = database().await; + let root_id = "<20261111.1+abc@x>"; + insert( + &db, + &[ + email(root_id, None, None), + // Names the root only in References. + email("", None, Some(root_id)), + email("", Some(""), None), + // What the root's id means as a regex: `.` any character, `1+` + // one or more 1s. Not the root. + email("", None, Some("<20261111.1abc@x>")), + ], + ) + .await; + + let thread = collect_thread_descendants(&db, root(&db, root_id).await) + .await + .unwrap(); + assert_eq!(ids(&thread), [root_id, "", ""]); +} + +#[tokio::test] +async fn a_dot_matches_only_a_dot() { + let (_dir, db) = database().await; + insert( + &db, + &[ + email("", None, None), + email("", Some(""), Some("")), + ], + ) + .await; + + let replies = db + .get_lore_emails_referencing_any(&["".to_string()]) + .await + .unwrap(); + assert!(replies.is_empty(), "matched: {:?}", ids(&replies)); + + let thread = collect_thread_descendants(&db, root(&db, "").await) + .await + .unwrap(); + assert_eq!(ids(&thread), [""]); +} + +#[tokio::test] +async fn a_walk_makes_one_lookup_per_level() { + let (_dir, db) = database().await; + let mut emails = vec![ + email("", None, None), + email("", Some(""), Some("")), + ]; + let level2: Vec = (0..40).map(|i| format!("")).collect(); + for id in &level2 { + // Real replies usually list the root in References too, which puts + // them a level up; these name only their parent. + emails.push(email(id, Some(""), Some(""))); + } + insert(&db, &emails).await; + + let lookups = Cell::new(0); + let thread = + collect_thread_descendants_with(root(&db, "").await, async |ids: &[String]| { + lookups.set(lookups.get() + 1); + db.get_lore_emails_referencing_any(ids).await + }) + .await + .unwrap(); + + // Level 1, level 2, and the empty level below it. + assert_eq!(lookups.get(), 3); + assert_eq!(thread.len(), 42); + assert_eq!(ids(&thread[..2]), ["", ""]); + let mut expected: Vec<&str> = level2.iter().map(String::as_str).collect(); + expected.sort_unstable(); + assert_eq!(sorted(ids(&thread[2..])), expected); +} + +#[tokio::test] +async fn quotes_and_backslashes_in_a_message_id_match() { + let (_dir, db) = database().await; + let root_id = r""; + insert( + &db, + &[ + email(root_id, None, None), + email("", Some(root_id), None), + email("", None, Some(root_id)), + ], + ) + .await; + + let thread = collect_thread_descendants(&db, root(&db, root_id).await) + .await + .unwrap(); + assert_eq!( + sorted(ids(&thread)), + ["", "", root_id] + ); +} + +#[tokio::test] +async fn headers_match_as_they_were_stored() { + let (_dir, db) = database().await; + // A Message-ID header folded onto its own line keeps the space. + let folded = email(" ", None, None); + insert( + &db, + &[ + folded.clone(), + email("", Some(""), None), + email("", Some(" "), None), + email("", None, Some(", ")), + email("", None, Some("")), + // A longer id that starts with the root's. + email("", None, Some(" ")), + ], + ) + .await; + + let thread = collect_thread_descendants(&db, folded).await.unwrap(); + assert_eq!( + sorted(ids(&thread)), + [ + " ", + "", + "", + "", + "", + ] + ); +} + +#[tokio::test] +async fn a_lookup_of_many_ids_returns_each_reply_once() { + let (_dir, db) = database().await; + // More ids than one query takes, so the lookup splits them. + let parents: Vec = (0..130).map(|i| format!("")).collect(); + let mut emails: Vec = parents + .iter() + .enumerate() + .map(|(i, parent)| email(&format!(""), Some(parent), None)) + .collect(); + // Answers parents that land in different queries. + emails.push(email( + "", + Some(""), + Some(" "), + )); + insert(&db, &emails).await; + + let replies = db.get_lore_emails_referencing_any(&parents).await.unwrap(); + assert_eq!(replies.len(), 131); + let mut expected: Vec<&str> = emails.iter().map(|e| e.message_id.as_str()).collect(); + expected.sort_unstable(); + assert_eq!(sorted(ids(&replies)), expected); +} + +#[tokio::test] +async fn a_level_lists_in_reply_to_answers_before_references_ones() { + let (_dir, db) = database().await; + insert( + &db, + &[ + email("", None, None), + email("", Some(""), Some("")), + email("", None, Some(" ")), + email("", Some(""), Some("")), + email("", Some(""), Some("")), + ], + ) + .await; + + // `` names the root in References, so it belongs to + // level 1, after the root's In-Reply-To answers. + let thread = collect_thread_descendants(&db, root(&db, "").await) + .await + .unwrap(); + assert_eq!( + ids(&thread), + [ + "", + "", + "", + "", + "", + ] + ); +} + +#[tokio::test] +async fn an_email_answering_two_messages_of_a_level_is_placed_under_the_first() { + let (_dir, db) = database().await; + insert( + &db, + &[ + email("", None, None), + email("", Some(""), Some("")), + email("", Some(""), Some("")), + // Stored in an order that neither table order nor "In-Reply-To + // answers of every parent first" reproduces. + email("", Some(""), Some("")), + // Answers `` by In-Reply-To and `` by References. + email("", Some(""), Some("")), + email("", Some(""), Some("")), + ], + ) + .await; + + // Level 2 follows level 1's order, `` then ``: under `` + // its In-Reply-To answer, then `` by References; under `` + // only what is left. + let thread = collect_thread_descendants(&db, root(&db, "").await) + .await + .unwrap(); + assert_eq!( + ids(&thread), + [ + "", + "", + "", + "", + "", + "", + ] + ); +} + +#[test] +fn header_names_only_whole_message_ids() { + let names = header_names_message_id; + assert!(names("", "")); + assert!(names("", " ")); + assert!(names( + " (Someone's message of \"Wed, 11 Nov 2026\")", + "" + )); + assert!(names("\t", "")); + assert!(!names("", "")); + assert!(!names("", "a@x>")); + // Ids without angle brackets still need a separator on each side. + assert!(names("a@x.y a@x", "a@x")); + assert!(!names("a@x.y", "a@x")); + assert!(!names("", "")); + assert!(!names("", " ")); +}