diff --git a/crates/store/src/backend/mysql/search.rs b/crates/store/src/backend/mysql/search.rs index c5c3188..7df89af 100644 --- a/crates/store/src/backend/mysql/search.rs +++ b/crates/store/src/backend/mysql/search.rs @@ -2,6 +2,8 @@ * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC * * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + * + * Modified by Coffey Labs in 2026 for INBUXA. */ use crate::{ @@ -19,7 +21,7 @@ use crate::{ write::SearchIndex, }; use mysql_async::{IsolationLevel, TxOpts, Value, prelude::Queryable}; -use nlp::tokenizers::word::WordTokenizer; +use nlp::{language::Language, tokenizers::word::WordTokenizer}; use std::fmt::Write; impl MysqlStore { @@ -146,6 +148,20 @@ impl MysqlStore { } } +// inbuxa: InnoDB's default full-text stopword list +// (INFORMATION_SCHEMA.INNODB_FT_DEFAULT_STOPWORD) and innodb_ft_min_token_size +// default; words outside these are not in a FULLTEXT index. +const FT_STOPWORDS: &[&str] = &[ + "a", "about", "an", "are", "as", "at", "be", "by", "com", "de", "en", "for", "from", "how", + "i", "in", "is", "it", "la", "of", "on", "or", "that", "the", "this", "to", "was", "what", + "when", "where", "who", "will", "with", "und", "www", +]; +const FT_MIN_TOKEN_SIZE: usize = 3; + +fn is_ft_indexed(word: &str) -> bool { + word.chars().count() >= FT_MIN_TOKEN_SIZE && !FT_STOPWORDS.contains(&word) +} + fn build_filter(query: &mut String, filters: &[SearchFilter]) -> Vec { if filters.is_empty() { return Vec::new(); @@ -171,30 +187,77 @@ fn build_filter(query: &mut String, filters: &[SearchFilter]) -> Vec { if field.is_text() && matches!(op, SearchOperator::Equal | SearchOperator::Contains) { - let (value, mode) = match (value, op) { - (SearchValue::Text { value, .. }, SearchOperator::Equal) => { - (Value::Bytes(format!("{value:?}").into_bytes()), "BOOLEAN") - } - (SearchValue::Text { value, .. }, ..) => { + let (value, mode, unindexed) = match (value, op) { + (SearchValue::Text { value, .. }, SearchOperator::Equal) => ( + Value::Bytes(format!("{value:?}").into_bytes()), + "BOOLEAN", + Vec::new(), + ), + (SearchValue::Text { value, language }, ..) => { let mut text_query = String::with_capacity(value.len() + 1); + let mut unindexed = Vec::new(); for item in WordTokenizer::new(value, MAX_TOKEN_LENGTH) { - if !text_query.is_empty() { - text_query.push(' '); + // inbuxa: InnoDB never indexes stopwords ("com", + // "de", "www", ...) or words under + // innodb_ft_min_token_size, and a required + // (+word) term it has not indexed matches no row, + // so "example.com" or "jo@example.org" found + // nothing. Such words are matched with a + // word-boundary REGEXP instead. + if is_ft_indexed(&item.word) { + if !text_query.is_empty() { + text_query.push(' '); + } + text_query.push('+'); + text_query.push_str(&item.word); + } else { + unindexed.push(item.word); } - text_query.push('+'); - text_query.push_str(&item.word); } - (Value::Bytes(text_query.into_bytes()), "BOOLEAN") + // For language text (bodies, subjects) the unindexed + // words are noise words and only checked when nothing + // else is left to match; keyword text (addresses, + // contact fields) checks every word, as the other + // backends do. + if !text_query.is_empty() && !matches!(language, Language::None) { + unindexed.clear(); + } + + (Value::Bytes(text_query.into_bytes()), "BOOLEAN", unindexed) } _ => { debug_assert!(false, "Invalid search value for text field"); continue; } }; - let _ = write!(query, "MATCH({}) AGAINST(? IN {mode} MODE)", field.column()); - values.push(value); + if unindexed.is_empty() { + let _ = + write!(query, "MATCH({}) AGAINST(? IN {mode} MODE)", field.column()); + values.push(value); + } else { + query.push('('); + let is_empty = matches!(&value, Value::Bytes(v) if v.is_empty()); + if !is_empty { + let _ = write!( + query, + "MATCH({}) AGAINST(? IN {mode} MODE) AND ", + field.column() + ); + values.push(value); + } + for (i, word) in unindexed.iter().enumerate() { + if i > 0 { + query.push_str(" AND "); + } + let _ = write!(query, "{} REGEXP ?", field.column()); + values.push(Value::Bytes( + format!("(^|[^[:alnum:]]){word}([^[:alnum:]]|$)").into_bytes(), + )); + } + query.push(')'); + } } else if let SearchValue::KeyValues(kv) = value { let (key, value) = kv.iter().next().unwrap(); diff --git a/crates/store/src/backend/postgres/search.rs b/crates/store/src/backend/postgres/search.rs index ca083e4..a626894 100644 --- a/crates/store/src/backend/postgres/search.rs +++ b/crates/store/src/backend/postgres/search.rs @@ -2,12 +2,17 @@ * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC * * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + * + * Modified by Coffey Labs in 2026 for INBUXA. */ use crate::{ - backend::postgres::{ - DELETE_CHUNK_SIZE, MIN_DELETE_CHUNK_SIZE, PostgresStore, PsqlSearchField, into_error, - into_pool_error, is_timeout_error, + backend::{ + MAX_TOKEN_LENGTH, + postgres::{ + DELETE_CHUNK_SIZE, MIN_DELETE_CHUNK_SIZE, PostgresStore, PsqlSearchField, into_error, + into_pool_error, is_timeout_error, + }, }, search::{ IndexDocument, SearchComparator, SearchDocumentId, SearchFilter, SearchOperator, @@ -15,7 +20,7 @@ use crate::{ }, write::SearchIndex, }; -use nlp::language::Language; +use nlp::{language::Language, tokenizers::space::SpaceTokenizer}; use std::fmt::Write; use tokio_postgres::{ IsolationLevel, @@ -43,6 +48,19 @@ impl PostgresStore { let primary_keys = index.primary_keys(); let all_fields = index.all_fields(); let fields = document.fields; + // inbuxa: keyword text (addresses, contact fields, ...) is split into + // words before it reaches the text parser, see keyword_terms(). + let keywords = primary_keys + .iter() + .chain(all_fields) + .map(|field| match fields.get(field) { + Some(SearchValue::Text { + value, + language: Language::None, + }) if field.is_text() => Some(keyword_terms(value)), + _ => None, + }) + .collect::>(); let mut values = Vec::with_capacity(fields.len() + 2); let mut query = format!("INSERT INTO {} (", index.psql_table()); @@ -74,7 +92,20 @@ impl PostgresStore { (0, PG_UNSTEMMED_LANG) }; - if field.is_text() { + if let Some(keywords) = &keywords[i] { + let _ = write!(&mut query, "to_tsvector('{language}',{value_ref})"); + values.push(keywords as &(dyn ToSql + Sync)); + if field.sort_column().is_some() { + let value_ref = format!("${}", values.len() + 1); + if text_len > 255 { + let _ = write!(&mut query, ",left({value_ref},255)"); + } else { + let _ = write!(&mut query, ",{value_ref}"); + } + values.push(value as &(dyn ToSql + Sync)); + } + continue; + } else if field.is_text() { let _ = write!(&mut query, "to_tsvector('{language}',{value_ref})"); } else if text_len > 512 { query.push_str("left("); @@ -134,6 +165,7 @@ impl PostgresStore { ) -> trc::Result> { let mut query = format!("SELECT {} FROM {}", R::field().column(), index.psql_table()); let params = self.build_filter(&mut query, filters); + let params = params.iter().map(SqlParam::as_sql).collect::>(); if !sort.is_empty() { build_sort(&mut query, sort); } @@ -155,6 +187,7 @@ impl PostgresStore { let table = filter.index.psql_table(); let mut where_clause = String::new(); let params = self.build_filter(&mut where_clause, &filter.filters); + let params = params.iter().map(SqlParam::as_sql).collect::>(); let conn = self.conn_pool.get().await.map_err(into_pool_error)?; let s = conn .prepare_cached(&format!("DELETE FROM {table}{where_clause}")) @@ -196,7 +229,7 @@ impl PostgresStore { &self, query: &mut String, filters: &'x [SearchFilter], - ) -> Vec<&'x (dyn ToSql + Sync)> { + ) -> Vec> { if filters.is_empty() { return Vec::new(); } @@ -237,6 +270,10 @@ impl PostgresStore { if matches!(language, Language::None) { let _ = write!(query, "@@ {method}('{config}', ${value_pos})"); + if let SearchValue::Text { value, .. } = value { + values.push(SqlParam::Owned(keyword_terms(value))); + continue; + } } else { let _ = write!(query, "@@ ({method}('{config}', ${value_pos})"); for fallback in [PG_FALLBACK_LANG, PG_UNSTEMMED_LANG] { @@ -247,18 +284,18 @@ impl PostgresStore { } query.push(')'); } - values.push(value as &(dyn ToSql + Sync)); + values.push(SqlParam::Ref(value)); } else if let SearchValue::KeyValues(kv) = value { query.push_str(field.column()); query.push(' '); let (key, value) = kv.iter().next().unwrap(); - values.push(key as &(dyn ToSql + Sync)); + values.push(SqlParam::Ref(key)); if !value.is_empty() { let _ = write!(query, "->> ${value_pos} "); op.write_pqsql(query, values.len() + 1); - values.push(value as &(dyn ToSql + Sync)); + values.push(SqlParam::Ref(value)); } else { let _ = write!(query, " ? ${value_pos}"); } @@ -267,7 +304,7 @@ impl PostgresStore { query.push(' '); op.write_pqsql(query, value_pos); - values.push(value as &(dyn ToSql + Sync)); + values.push(SqlParam::Ref(value)); } } SearchFilter::And | SearchFilter::Or => { @@ -321,6 +358,38 @@ impl PostgresStore { } } +// inbuxa: PostgreSQL's text parser keeps "user@example.com" (and host names, +// URLs, file paths, ...) as a single token, so a search for "user" or +// "example.com" never matched an address. Keyword text is split into words the +// same way the built-in index splits it (SpaceTokenizer: lowercase runs of +// alphanumerics) on both the indexing and the query side, so a full address, +// its local part, its domain and the display-name words all match, as they do +// on the other backends. +pub(crate) fn keyword_terms(value: &str) -> String { + let mut terms = String::with_capacity(value.len()); + for token in SpaceTokenizer::new(value, MAX_TOKEN_LENGTH) { + if !terms.is_empty() { + terms.push(' '); + } + terms.push_str(&token); + } + terms +} + +pub(super) enum SqlParam<'x> { + Ref(&'x (dyn ToSql + Sync)), + Owned(String), +} + +impl SqlParam<'_> { + fn as_sql(&self) -> &(dyn ToSql + Sync) { + match self { + SqlParam::Ref(value) => *value, + SqlParam::Owned(value) => value, + } + } +} + fn build_sort(query: &mut String, sort: &[SearchComparator]) { query.push_str(" ORDER BY "); for (i, comparator) in sort.iter().enumerate() { diff --git a/tests/src/store/query.rs b/tests/src/store/query.rs index 9556d43..0f3f6ce 100644 --- a/tests/src/store/query.rs +++ b/tests/src/store/query.rs @@ -128,6 +128,11 @@ pub async fn test(test: &TestServer) { println!("Running trace document tests..."); test_trace_documents(store.clone()).await; + // inbuxa: address fields match by full address, local part, domain and + // display name on every backend + println!("Running address search tests..."); + test_address_search(store.clone()).await; + // Large document insert test println!("Running large document insert tests..."); let mut large_text = String::with_capacity(20 * 1024 * 1024); @@ -972,3 +977,155 @@ async fn test_trace_documents(store: SearchStore) { .unwrap(); } } + +// inbuxa: the message indexer passes each display name and each address of +// From/To/Cc/Bcc as keyword text (Language::None). The built-in index splits +// that text into words, so an address is found by its full form, its local +// part, its domain or a display-name word; PostgreSQL kept the whole address +// as one token and MySQL dropped stopwords such as "com" and words under three +// characters. The expected results below are the built-in (RocksDB/SQLite) +// results and must be the same on every backend. +async fn test_address_search(store: SearchStore) { + const ACCOUNT_ID: u32 = 7; + let messages: [[&[(&str, &str)]; 4]; 5] = [ + // From, To, Cc, Bcc + [ + &[("Amazon.com", "noreply@amazon.com")], + &[("Jane Doe", "jane.doe@example.org")], + &[], + &[], + ], + [ + &[("", "shipment-tracking@amazon.com")], + &[("", "jo@io.de")], + &[("Jane Doe", "jane.doe@example.org")], + &[], + ], + [ + &[("GitHub", "noreply@github.com")], + &[("Jo Li", "jo@io.de")], + &[], + &[("Audit", "audit@example.org")], + ], + [ + &[("Jane Doe", "jane.doe@example.org")], + &[("Amazon Web Services", "aws-marketing@amazon.com")], + &[("Bob", "bob@example.net")], + &[("", "noreply@amazon.com")], + ], + [ + &[("Newsletter", "news@www.example.com")], + &[("", "undisclosed@example.org")], + &[], + &[], + ], + ]; + let fields = [ + EmailSearchField::From, + EmailSearchField::To, + EmailSearchField::Cc, + EmailSearchField::Bcc, + ]; + + let mut documents = Vec::new(); + let mut mask = RoaringBitmap::new(); + for (document_id, message) in messages.iter().enumerate() { + let mut document = IndexDocument::new(SearchIndex::Email) + .with_account_id(ACCOUNT_ID) + .with_document_id(document_id as u32); + for (field, addresses) in fields.iter().zip(message.iter()) { + for (name, address) in addresses.iter() { + if !name.is_empty() { + document.index_text(field.clone(), name, Language::None); + } + document.index_text(field.clone(), address, Language::None); + } + } + document.index_unsigned(EmailSearchField::ReceivedAt, document_id as u64); + documents.push(document); + mask.insert(document_id as u32); + } + store.index(documents).await.unwrap(); + if let SearchStore::ElasticSearch(store) = &store { + store.refresh_index(SearchIndex::Email).await.unwrap(); + } + + for (field, text, expected) in [ + // full address + (EmailSearchField::From, "noreply@amazon.com", vec![0u32]), + (EmailSearchField::To, "jane.doe@example.org", vec![0]), + (EmailSearchField::Cc, "jane.doe@example.org", vec![1]), + (EmailSearchField::Bcc, "noreply@amazon.com", vec![3]), + (EmailSearchField::To, "jo@io.de", vec![1, 2]), + // local part + (EmailSearchField::From, "noreply", vec![0, 2]), + (EmailSearchField::To, "jo", vec![1, 2]), + (EmailSearchField::Cc, "bob", vec![3]), + (EmailSearchField::Bcc, "audit", vec![2]), + // domain + (EmailSearchField::From, "amazon.com", vec![0, 1]), + (EmailSearchField::From, "amazon", vec![0, 1]), + (EmailSearchField::To, "example.org", vec![0, 4]), + (EmailSearchField::To, "io.de", vec![1, 2]), + (EmailSearchField::Cc, "example.net", vec![3]), + (EmailSearchField::Bcc, "example.org", vec![2]), + (EmailSearchField::From, "www.example.com", vec![4]), + (EmailSearchField::From, "com", vec![0, 1, 2, 4]), + // display name + (EmailSearchField::From, "Jane", vec![3]), + (EmailSearchField::From, "jane doe", vec![3]), + (EmailSearchField::To, "Web Services", vec![3]), + (EmailSearchField::To, "Li", vec![2]), + (EmailSearchField::Cc, "Doe", vec![1]), + (EmailSearchField::Bcc, "Audit", vec![2]), + // hyphenated local part + (EmailSearchField::From, "shipment-tracking", vec![1]), + (EmailSearchField::From, "tracking", vec![1]), + // no match + (EmailSearchField::From, "amazon.org", vec![]), + (EmailSearchField::To, "noreply", vec![]), + (EmailSearchField::Bcc, "jane", vec![]), + ] { + let ids = store + .query_account( + SearchQuery::new(SearchIndex::Email) + .with_filters(vec![ + SearchFilter::eq(SearchField::AccountId, ACCOUNT_ID), + SearchFilter::has_keyword(field.clone(), text), + ]) + .with_comparator(SearchComparator::ascending(EmailSearchField::ReceivedAt)) + .with_mask(mask.clone()), + ) + .await + .unwrap(); + assert_eq!(ids, expected, "{field:?} {text:?}"); + } + + // TEXT-style search across all address fields + let ids = store + .query_account( + SearchQuery::new(SearchIndex::Email) + .with_filters(vec![ + SearchFilter::eq(SearchField::AccountId, ACCOUNT_ID), + SearchFilter::Or, + SearchFilter::has_keyword(EmailSearchField::From, "example.org"), + SearchFilter::has_keyword(EmailSearchField::To, "example.org"), + SearchFilter::has_keyword(EmailSearchField::Cc, "example.org"), + SearchFilter::has_keyword(EmailSearchField::Bcc, "example.org"), + SearchFilter::End, + ]) + .with_comparator(SearchComparator::ascending(EmailSearchField::ReceivedAt)) + .with_mask(mask.clone()), + ) + .await + .unwrap(); + assert_eq!(ids, vec![0, 1, 2, 3, 4]); + + store + .unindex( + SearchQuery::new(SearchIndex::Email) + .with_filter(SearchFilter::eq(SearchField::AccountId, ACCOUNT_ID)), + ) + .await + .unwrap(); +}