From 3d20a82354f997ac7cbe51986dc0454ffca43d20 Mon Sep 17 00:00:00 2001 From: mdecimus <11444311+mdecimus@users.noreply.github.com> Date: Sun, 2 Nov 2025 16:01:05 +0100 Subject: [PATCH] Database schema optimization - part 4 --- crates/common/src/config/telemetry.rs | 4 + crates/common/src/telemetry/tracers/index.rs | 178 ------- crates/common/src/telemetry/tracers/store.rs | 158 ++++-- crates/email/src/message/index/search.rs | 159 ++++-- crates/groupware/src/calendar/index.rs | 55 +- crates/groupware/src/contact/index.rs | 49 +- .../src/management/enterprise/telemetry.rs | 36 +- crates/services/src/housekeeper/mod.rs | 4 +- crates/services/src/task_manager/index.rs | 483 ++++++++++++++---- crates/services/src/task_manager/mod.rs | 25 +- crates/store/src/dispatch/search.rs | 7 +- crates/store/src/search/mod.rs | 44 ++ crates/store/src/write/key.rs | 11 + 13 files changed, 794 insertions(+), 419 deletions(-) delete mode 100644 crates/common/src/telemetry/tracers/index.rs diff --git a/crates/common/src/config/telemetry.rs b/crates/common/src/config/telemetry.rs index 33f21390..f2e0dace 100644 --- a/crates/common/src/config/telemetry.rs +++ b/crates/common/src/config/telemetry.rs @@ -99,6 +99,7 @@ pub struct WebhookTracer { #[cfg(feature = "enterprise")] pub struct StoreTracer { pub store: store::Store, + pub indexed: bool, } // SPDX-SnippetEnd @@ -538,6 +539,9 @@ impl Tracers { lossy: false, typ: TelemetrySubscriberType::StoreTracer(StoreTracer { store: store.clone(), + indexed: config + .property_or_default("tracing.history.indexed", "false") + .unwrap_or(false), }), }; diff --git a/crates/common/src/telemetry/tracers/index.rs b/crates/common/src/telemetry/tracers/index.rs deleted file mode 100644 index 1b19046c..00000000 --- a/crates/common/src/telemetry/tracers/index.rs +++ /dev/null @@ -1,178 +0,0 @@ -fn build_index() { - let mut queue_ids = AHashSet::new(); - let mut values = AHashSet::new(); - - for event in events.iter().chain([span, &event]) { - for (key, value) in &event.keys { - match (key, value) { - (Key::QueueId, Value::UInt(queue_id)) => { - queue_ids.insert(*queue_id); - } - (Key::From | Key::To | Key::Domain | Key::Hostname, Value::String(address)) => { - values.insert(address.clone()); - } - (Key::To, Value::Array(value)) => { - for value in value { - if let Value::String(address) = value { - values.insert(address.clone()); - } - } - } - (Key::RemoteIp, Value::Ipv4(ip)) => { - values.insert(ip.to_string().into()); - } - (Key::RemoteIp, Value::Ipv6(ip)) => { - values.insert(ip.to_string().into()); - } - - _ => {} - } - } - } - // Build index - batch.set( - ValueClass::Telemetry(TelemetryClass::Index { - span_id, - value: (span.inner.typ.code() as u16).to_be_bytes().to_vec(), - }), - vec![], - ); - for queue_id in queue_ids { - batch.set( - ValueClass::Telemetry(TelemetryClass::Index { - span_id, - value: queue_id.to_be_bytes().to_vec(), - }), - vec![], - ); - } - for value in values { - batch.set( - ValueClass::Telemetry(TelemetryClass::Index { - span_id, - value: value.as_bytes().to_vec(), - }), - vec![], - ); - } -} - -/* - - -enum SpanCollector { - Vec(Vec), - HashSet(AHashSet), - Empty, -} - -impl SpanCollector { - fn new(num_params: usize) -> Self { - if num_params == 1 { - Self::Vec(Vec::new()) - } else { - Self::HashSet(AHashSet::new()) - } - } - - fn insert(&mut self, span_id: u64) { - match self { - Self::Vec(vec) => vec.push(span_id), - Self::HashSet(set) => { - set.insert(span_id); - } - _ => unreachable!(), - } - } - - fn into_vec(self) -> Vec { - match self { - Self::Vec(mut vec) => { - vec.sort_unstable_by(|a, b| b.cmp(a)); - vec - } - Self::HashSet(set) => { - let mut vec: Vec = set.into_iter().collect(); - vec.sort_unstable_by(|a, b| b.cmp(a)); - vec - } - Self::Empty => Vec::new(), - } - } - - fn intersect(&mut self, other_span: Self) -> bool { - match (self, other_span) { - (Self::HashSet(set), Self::HashSet(other_set)) => { - set.retain(|span_id| other_set.contains(span_id)); - set.is_empty() - } - _ => unreachable!(), - } - } -} - - let mut spans = SpanCollector::Empty; - let num_params = params.len(); - let todo = "use FTS"; - - for (param_num, param) in params.iter().enumerate() { - let (value, exact_len) = match param { - TracingQuery::EventType(event) => ( - (event.code() as u16).to_be_bytes().to_vec(), - std::mem::size_of::() + U64_LEN, - ), - TracingQuery::QueueId(id) => ( - id.to_be_bytes().to_vec(), - std::mem::size_of::() + U64_LEN, - ), - TracingQuery::Keywords(value) => { - if let Some(value) = value.strip_prefix('"').and_then(|v| v.strip_suffix('"')) { - (value.as_bytes().to_vec(), value.len() + U64_LEN) - } else { - (value.as_bytes().to_vec(), 0) - } - } - }; - - let mut param_spans = SpanCollector::new(num_params); - self.iterate( - IterateParams::new( - ValueKey::from(ValueClass::Telemetry(TelemetryClass::Index { - span_id: 0, - value: value.clone(), - })), - ValueKey::from(ValueClass::Telemetry(TelemetryClass::Index { - span_id: u64::MAX, - value, - })), - ) - .no_values(), - |key, _| { - if exact_len == 0 || key.len() == exact_len { - let span_id = key - .deserialize_be_u64(key.len() - U64_LEN) - .caused_by(trc::location!())?; - - if (from_span_id == 0 || span_id >= from_span_id) - && (to_span_id == 0 || span_id <= to_span_id) - { - param_spans.insert(span_id); - } - } - - Ok(true) - }, - ) - .await - .caused_by(trc::location!())?; - - if param_num == 0 { - spans = param_spans; - } else if spans.intersect(param_spans) { - return Ok(Vec::new()); - } - } - - Ok(spans.into_vec()) - -*/ diff --git a/crates/common/src/telemetry/tracers/store.rs b/crates/common/src/telemetry/tracers/store.rs index 82f2d169..9284238c 100644 --- a/crates/common/src/telemetry/tracers/store.rs +++ b/crates/common/src/telemetry/tracers/store.rs @@ -10,9 +10,11 @@ use crate::config::telemetry::StoreTracer; use ahash::{AHashMap, AHashSet}; +use nlp::language::Language; use std::{future::Future, time::Duration}; use store::{ - Deserialize, Store, ValueKey, + Deserialize, SearchStore, Store, ValueKey, + search::{IndexDocument, SearchField, SearchFilter, SearchQuery, TracingSearchField}, write::{BatchBuilder, SearchIndex, TaskQueueClass, TelemetryClass, ValueClass, now}, }; use trc::{ @@ -51,27 +53,30 @@ pub(crate) fn spawn_store_tracer(builder: SubscriberBuilder, settings: StoreTrac .any(|(k, v)| matches!((k, v), (Key::QueueId, Value::UInt(_)))) { // Serialize events - batch - .set( - ValueClass::Telemetry(TelemetryClass::Span { span_id }), - serialize_events( - [span.as_ref()] - .into_iter() - .chain(events.iter().map(|event| event.as_ref())) - .chain([event.as_ref()].into_iter()), - events.len() + 2, - ), - ) - .with_account_id((span_id >> 32) as u32) // TODO: This is hacky, improve - .with_document(span_id as u32) - .set( - ValueClass::TaskQueue(TaskQueueClass::UpdateIndex { - due: now, - index: SearchIndex::TracingSpan, - is_insert: true, - }), - vec![], - ); + batch.set( + ValueClass::Telemetry(TelemetryClass::Span { span_id }), + serialize_events( + [span.as_ref()] + .into_iter() + .chain(events.iter().map(|event| event.as_ref())) + .chain([event.as_ref()].into_iter()), + events.len() + 2, + ), + ); + + if settings.indexed { + batch + .with_account_id((span_id >> 32) as u32) // TODO: This is hacky, improve + .with_document(span_id as u32) + .set( + ValueClass::TaskQueue(TaskQueueClass::UpdateIndex { + due: now, + index: SearchIndex::TracingSpan, + is_insert: true, + }), + vec![], + ); + } } } } @@ -86,12 +91,6 @@ pub(crate) fn spawn_store_tracer(builder: SubscriberBuilder, settings: StoreTrac }); } -pub enum TracingQuery { - EventType(EventType), - QueueId(u64), - Keywords(String), -} - pub trait TracingStore: Sync + Send { fn get_span( &self, @@ -101,13 +100,11 @@ pub trait TracingStore: Sync + Send { &self, span_id: u64, ) -> impl Future>>> + Send; - fn query_spans( + fn purge_spans( &self, - params: &[TracingQuery], - from_span_id: u64, - to_span_id: u64, - ) -> impl Future>> + Send; - fn purge_spans(&self, period: Duration) -> impl Future> + Send; + period: Duration, + search_store: Option<&SearchStore>, + ) -> impl Future> + Send; } impl TracingStore for Store { @@ -129,16 +126,11 @@ impl TracingStore for Store { .map(|span| span.map(|span| span.0)) } - async fn query_spans( + async fn purge_spans( &self, - params: &[TracingQuery], - from_span_id: u64, - to_span_id: u64, - ) -> trc::Result> { - todo!() - } - - async fn purge_spans(&self, period: Duration) -> trc::Result<()> { + period: Duration, + search_store: Option<&SearchStore>, + ) -> trc::Result<()> { let until_span_id = SnowflakeIdGenerator::from_duration(period).ok_or_else(|| { trc::StoreEvent::UnexpectedError .caused_by(trc::location!()) @@ -154,7 +146,15 @@ impl TracingStore for Store { .await .caused_by(trc::location!())?; - let todo = "delete from index"; + if let Some(search_store) = search_store { + search_store + .unindex( + SearchQuery::new(SearchIndex::TracingSpan) + .with_filter(SearchFilter::lt(SearchField::Id, until_span_id)), + ) + .await + .caused_by(trc::location!())?; + } Ok(()) } @@ -240,3 +240,73 @@ impl Deserialize for RawSpan { Ok(Self(bytes.to_vec())) } } + +pub fn build_span_document( + span_id: u64, + events: Vec>, + index_fields: &AHashSet, +) -> IndexDocument { + let mut document = IndexDocument::with_default_language(Language::None); + + document.index_unsigned(SearchField::Id, span_id); + + for event in events { + for (idx, (key, value)) in event.keys.into_iter().enumerate() { + if idx == 0 + && (index_fields.is_empty() + || index_fields.contains(&TracingSearchField::EventType.into())) + { + document.index_unsigned(TracingSearchField::EventType, event.inner.typ.code()); + } + + match (key, value) { + (Key::QueueId, Value::UInt(queue_id)) => { + if index_fields.is_empty() + || index_fields.contains(&TracingSearchField::QueueId.into()) + { + document.insert_keyword(TracingSearchField::QueueId, queue_id); + } + } + (Key::From | Key::To | Key::Domain | Key::Hostname, Value::String(address)) => { + if index_fields.is_empty() + || index_fields.contains(&TracingSearchField::Address.into()) + { + document.insert_keyword(TracingSearchField::Address, address.into_string()); + } + } + (Key::To, Value::Array(value)) => { + if index_fields.is_empty() + || index_fields.contains(&TracingSearchField::Address.into()) + { + for value in value { + if let Value::String(address) = value { + document.insert_keyword( + TracingSearchField::Address, + address.into_string(), + ); + } + } + } + } + (Key::RemoteIp, Value::Ipv4(ip)) => { + if index_fields.is_empty() + || index_fields.contains(&TracingSearchField::RemoteIp.into()) + { + document.insert_keyword(TracingSearchField::RemoteIp, ip.to_string()); + } + } + (Key::RemoteIp, Value::Ipv6(ip)) => { + if index_fields.is_empty() + || index_fields.contains(&TracingSearchField::RemoteIp.into()) + { + document.insert_keyword(TracingSearchField::RemoteIp, ip.to_string()); + } + } + + _ => {} + } + } + } + + document +} diff --git a/crates/email/src/message/index/search.rs b/crates/email/src/message/index/search.rs index c24d9f01..93d3b9f8 100644 --- a/crates/email/src/message/index/search.rs +++ b/crates/email/src/message/index/search.rs @@ -23,14 +23,29 @@ impl ArchivedMessageMetadata { pub fn index_document( &self, raw_message: &[u8], - index_headers: &AHashSet>, + index_fields: &AHashSet, + index_all_headers: bool, ) -> IndexDocument { let mut language = Language::Unknown; let message_contents = &self.contents[0]; let mut document = IndexDocument::with_default_language(language); - document.index_unsigned(EmailSearchField::ReceivedAt, self.received_at.to_native()); - document.index_unsigned(EmailSearchField::Size, self.size.to_native()); + if index_fields.is_empty() + || index_fields.contains(&SearchField::Email(EmailSearchField::ReceivedAt)) + { + document.index_unsigned( + SearchField::Email(EmailSearchField::ReceivedAt), + self.received_at.to_native(), + ); + } + if index_fields.is_empty() + || index_fields.contains(&SearchField::Email(EmailSearchField::Size)) + { + document.index_unsigned( + SearchField::Email(EmailSearchField::Size), + self.size.to_native(), + ); + } for (part_id, part) in message_contents .parts @@ -43,69 +58,98 @@ impl ArchivedMessageMetadata { language = part_language; for header in part.headers.iter().rev() { - let header_name = HeaderName::from(&header.name); - if !index_headers.is_empty() && !index_headers.contains(&header_name) { - continue; - } - let header_name = match header_name { - HeaderName::Other(name) => Cow::Owned(name.into_owned()), - _ => Cow::Borrowed(header_name.as_static_str()), - }; - match &header.name { ArchivedHeaderName::From => { - header.value.visit_addresses(|_, value| { - document.index_text( - EmailSearchField::From, - value, - Language::Unknown, - ); - }); + if index_fields.is_empty() + || index_fields + .contains(&SearchField::Email(EmailSearchField::From)) + { + header.value.visit_addresses(|_, value| { + document.index_text( + SearchField::Email(EmailSearchField::From), + value, + Language::Unknown, + ); + }); + } } ArchivedHeaderName::To => { - header.value.visit_addresses(|_, value| { - document.index_text(EmailSearchField::To, value, Language::Unknown); - }); + if index_fields.is_empty() + || index_fields.contains(&SearchField::Email(EmailSearchField::To)) + { + header.value.visit_addresses(|_, value| { + document.index_text( + SearchField::Email(EmailSearchField::To), + value, + Language::Unknown, + ); + }); + } } ArchivedHeaderName::Cc => { - header.value.visit_addresses(|_, value| { - document.index_text(EmailSearchField::Cc, value, Language::Unknown); - }); + if index_fields.is_empty() + || index_fields.contains(&SearchField::Email(EmailSearchField::Cc)) + { + header.value.visit_addresses(|_, value| { + document.index_text( + SearchField::Email(EmailSearchField::Cc), + value, + Language::Unknown, + ); + }); + } } ArchivedHeaderName::Bcc => { - header.value.visit_addresses(|_, value| { - document.index_text( - EmailSearchField::Bcc, - value, - Language::Unknown, - ); - }); + if index_fields.is_empty() + || index_fields.contains(&SearchField::Email(EmailSearchField::Bcc)) + { + header.value.visit_addresses(|_, value| { + document.index_text( + SearchField::Email(EmailSearchField::Bcc), + value, + Language::Unknown, + ); + }); + } } ArchivedHeaderName::Subject => { - if let Some(subject) = header.value.as_text() { + if (index_fields.is_empty() + || index_fields + .contains(&SearchField::Email(EmailSearchField::Subject))) + && let Some(subject) = header.value.as_text() + { document.index_text( - EmailSearchField::Subject, + SearchField::Email(EmailSearchField::Subject), subject, part_language, ); } } ArchivedHeaderName::Date => { - if let Some(date) = header.value.as_datetime() { + if (index_fields.is_empty() + || index_fields + .contains(&SearchField::Email(EmailSearchField::SentAt))) + && let Some(date) = header.value.as_datetime() + { document.index_integer( - EmailSearchField::SentAt, + SearchField::Email(EmailSearchField::SentAt), DateTime::from(date).to_timestamp(), ); } } _ => { - header.value.visit_text(|text| { - document.index_text( - EmailSearchField::Header(header_name.clone()), - text, - Language::Unknown, - ); - }); + let field = SearchField::Email(EmailSearchField::Header( + match HeaderName::from(&header.name) { + HeaderName::Other(name) => Cow::Owned(name.into_owned()), + header_name => Cow::Borrowed(header_name.as_static_str()), + }, + )); + + if index_all_headers || index_fields.contains(&field) { + header.value.visit_text(|text| { + document.index_text(field.clone(), text, Language::Unknown); + }); + } } } } @@ -125,16 +169,30 @@ impl ArchivedMessageMetadata { if message_contents.is_html_part(part_id) || message_contents.is_text_part(part_id) { - document.index_text(EmailSearchField::Body, text.as_ref(), part_language); - } else { + if index_fields.is_empty() + || index_fields.contains(&SearchField::Email(EmailSearchField::Body)) + { + document.index_text( + SearchField::Email(EmailSearchField::Body), + text.as_ref(), + part_language, + ); + } + } else if index_fields.is_empty() + || index_fields.contains(&SearchField::Email(EmailSearchField::Attachment)) + { document.index_text( - EmailSearchField::Attachment, + SearchField::Email(EmailSearchField::Attachment), text.as_ref(), part_language, ); } } - ArchivedMetadataPartType::Message(nested_message_id) => { + ArchivedMetadataPartType::Message(nested_message_id) + if index_fields.is_empty() + || index_fields + .contains(&SearchField::Email(EmailSearchField::Attachment)) => + { let nested_message = self.message_id(*nested_message_id); let nested_message_language = nested_message .root_part() @@ -146,7 +204,7 @@ impl ArchivedMessageMetadata { .header_value(&ArchivedHeaderName::Subject) { document.index_text( - EmailSearchField::Attachment, + SearchField::Email(EmailSearchField::Attachment), subject.as_ref(), nested_message_language, ); @@ -169,7 +227,7 @@ impl ArchivedMessageMetadata { _ => unreachable!(), }; document.index_text( - EmailSearchField::Attachment, + SearchField::Email(EmailSearchField::Attachment), text.as_ref(), language, ); @@ -182,7 +240,8 @@ impl ArchivedMessageMetadata { } } - let has_attachment = document.has_field(&SearchField::Email(EmailSearchField::Attachment)); + let has_attachment = + document.has_field(&(SearchField::Email(EmailSearchField::Attachment))); document.index_bool(EmailSearchField::HasAttachment, has_attachment); diff --git a/crates/groupware/src/calendar/index.rs b/crates/groupware/src/calendar/index.rs index 1cb2bbcb..91ebc2c5 100644 --- a/crates/groupware/src/calendar/index.rs +++ b/crates/groupware/src/calendar/index.rs @@ -12,6 +12,7 @@ use crate::calendar::{ ArchivedCalendarEventNotification, ArchivedEventPreferences, CalendarEventNotification, EventPreferences, }; +use ahash::AHashSet; use calcard::icalendar::{ ArchivedICalendarParameterValue, ArchivedICalendarProperty, ArchivedICalendarValue, ICalendarParameterValue, ICalendarProperty, ICalendarValue, @@ -19,7 +20,7 @@ use calcard::icalendar::{ use common::storage::index::{IndexValue, IndexableAndSerializableObject, IndexableObject}; use nlp::language::Language; use store::{ - search::{CalendarSearchField, IndexDocument}, + search::{CalendarSearchField, IndexDocument, SearchField}, write::{IndexPropertyClass, SearchIndex, ValueClass}, xxhash_rust::xxh3, }; @@ -336,10 +337,14 @@ impl ArchivedCalendarEvent { } impl ArchivedCalendarEvent { - pub fn index_document(&self) -> IndexDocument { + pub fn index_document(&self, index_fields: &AHashSet) -> IndexDocument { let mut document = IndexDocument::with_default_language(Language::Unknown); - document.index_integer(CalendarSearchField::Start, self.data.event_range_start()); + if index_fields.is_empty() + || index_fields.contains(&SearchField::Calendar(CalendarSearchField::Start)) + { + document.index_integer(CalendarSearchField::Start, self.data.event_range_start()); + } for component in self .data @@ -349,7 +354,7 @@ impl ArchivedCalendarEvent { .filter(|e| e.component_type.is_scheduling_object()) { for entry in component.entries.iter() { - let field = match entry.name { + let field = SearchField::Calendar(match entry.name { ArchivedICalendarProperty::Summary => CalendarSearchField::Title, ArchivedICalendarProperty::Description => CalendarSearchField::Description, ArchivedICalendarProperty::Location => CalendarSearchField::Location, @@ -357,27 +362,29 @@ impl ArchivedCalendarEvent { ArchivedICalendarProperty::Attendee => CalendarSearchField::Attendee, ArchivedICalendarProperty::Uid => CalendarSearchField::Uid, _ => continue, - }; + }); - for value in entry - .values - .iter() - .filter_map(|v| match v { - ArchivedICalendarValue::Text(v) => Some(v.as_str()), - ArchivedICalendarValue::Uri(uri) => uri.as_str(), - _ => None, - }) - .chain(entry.params.iter().filter_map(|p| match &p.value { - ArchivedICalendarParameterValue::Text(v) => Some(v.as_str()), - ArchivedICalendarParameterValue::Uri(uri) => uri.as_str(), - _ => None, - })) - { - document.index_text( - field, - value.strip_prefix("mailto:").unwrap_or(value), - Language::Unknown, - ); + if index_fields.is_empty() || index_fields.contains(&field) { + for value in entry + .values + .iter() + .filter_map(|v| match v { + ArchivedICalendarValue::Text(v) => Some(v.as_str()), + ArchivedICalendarValue::Uri(uri) => uri.as_str(), + _ => None, + }) + .chain(entry.params.iter().filter_map(|p| match &p.value { + ArchivedICalendarParameterValue::Text(v) => Some(v.as_str()), + ArchivedICalendarParameterValue::Uri(uri) => uri.as_str(), + _ => None, + })) + { + document.index_text( + field.clone(), + value.strip_prefix("mailto:").unwrap_or(value), + Language::Unknown, + ); + } } } } diff --git a/crates/groupware/src/contact/index.rs b/crates/groupware/src/contact/index.rs index eaf1498f..0d8735c0 100644 --- a/crates/groupware/src/contact/index.rs +++ b/crates/groupware/src/contact/index.rs @@ -5,6 +5,7 @@ */ use super::{AddressBook, ArchivedAddressBook, ArchivedContactCard, ContactCard}; +use ahash::AHashSet; use calcard::{ common::IanaString, vcard::{ @@ -15,7 +16,7 @@ use calcard::{ use common::storage::index::{IndexValue, IndexableAndSerializableObject, IndexableObject}; use nlp::language::Language; use store::{ - search::{ContactSearchField, IndexDocument}, + search::{ContactSearchField, IndexDocument, SearchField}, write::SearchIndex, xxhash_rust::xxh3, }; @@ -242,13 +243,17 @@ impl ArchivedContactCard { } impl ArchivedContactCard { - pub fn index_document(&self) -> IndexDocument { + pub fn index_document(&self, index_fields: &AHashSet) -> IndexDocument { let mut document = IndexDocument::with_default_language(Language::Unknown); - document.index_integer(ContactSearchField::Created, self.created.to_native()); + if index_fields.is_empty() + || index_fields.contains(&SearchField::Contact(ContactSearchField::Created)) + { + document.index_integer(ContactSearchField::Created, self.created.to_native()); + } for entry in self.card.entries.iter() { - let field = match entry.name { + let field = SearchField::Contact(match entry.name { ArchivedVCardProperty::N => ContactSearchField::Name, ArchivedVCardProperty::Nickname => ContactSearchField::Nickname, ArchivedVCardProperty::Org => ContactSearchField::Organization, @@ -263,28 +268,30 @@ impl ArchivedContactCard { ArchivedVCardProperty::Uid => ContactSearchField::Uid, ArchivedVCardProperty::Member => ContactSearchField::Member, _ => continue, - }; + }); - for value in entry.values.iter() { - match value { - ArchivedVCardValue::Text(v) => { - document.index_text(field, v, Language::Unknown); - } - ArchivedVCardValue::Kind(v) => { - document.index_text(field, v.as_str(), Language::Unknown); - } - ArchivedVCardValue::Component(v) => { - for item in v.iter() { - document.index_text(field, item, Language::Unknown); + if index_fields.is_empty() || index_fields.contains(&field) { + for value in entry.values.iter() { + match value { + ArchivedVCardValue::Text(v) => { + document.index_text(field.clone(), v, Language::Unknown); } + ArchivedVCardValue::Kind(v) => { + document.index_text(field.clone(), v.as_str(), Language::Unknown); + } + ArchivedVCardValue::Component(v) => { + for item in v.iter() { + document.index_text(field.clone(), item, Language::Unknown); + } + } + _ => (), } - _ => (), } - } - for param in entry.params.iter() { - if let ArchivedVCardParameterValue::Text(value) = ¶m.value { - document.index_text(field, value, Language::Unknown); + for param in entry.params.iter() { + if let ArchivedVCardParameterValue::Text(value) = ¶m.value { + document.index_text(field.clone(), value, Language::Unknown); + } } } } diff --git a/crates/http/src/management/enterprise/telemetry.rs b/crates/http/src/management/enterprise/telemetry.rs index 74ee2f20..e4af9f45 100644 --- a/crates/http/src/management/enterprise/telemetry.rs +++ b/crates/http/src/management/enterprise/telemetry.rs @@ -18,7 +18,7 @@ use common::{ auth::{AccessToken, oauth::GrantType}, telemetry::{ metrics::store::{Metric, MetricsStore}, - tracers::store::{TracingQuery, TracingStore}, + tracers::store::TracingStore, }, }; use directory::{Permission, backend::internal::manage}; @@ -31,7 +31,11 @@ use hyper::{ use mail_parser::DateTime; use serde_json::json; use std::future::Future; -use store::ahash::{AHashMap, AHashSet}; +use store::{ + ahash::{AHashMap, AHashSet}, + search::{SearchField, SearchFilter, SearchQuery, TracingSearchField}, + write::SearchIndex, +}; use trc::{ Collector, DeliveryEvent, EventType, Key, MetricType, QueueEvent, Value, ipc::{bitset::Bitset, subscriber::SubscriberBuilder}, @@ -72,11 +76,12 @@ impl TelemetryApi for Server { let page: usize = params.parse("page").unwrap_or(0); let limit: usize = params.parse("limit").unwrap_or(0); let mut tracing_query = Vec::new(); - if let Some(typ) = params.parse("type") { - tracing_query.push(TracingQuery::EventType(typ)); + tracing_query.push(SearchFilter::And); + if let Some(typ) = params.parse::("type") { + tracing_query.push(SearchFilter::eq(TracingSearchField::EventType, typ.id())); } - if let Some(queue_id) = params.parse("queue_id") { - tracing_query.push(TracingQuery::QueueId(queue_id)); + if let Some(queue_id) = params.parse::("queue_id") { + tracing_query.push(SearchFilter::eq(TracingSearchField::QueueId, queue_id)); } if let Some(query) = params.get("filter") { let mut buf = String::with_capacity(query.len()); @@ -86,14 +91,16 @@ impl TelemetryApi for Server { if in_quote { buf.push(' '); } else if !buf.is_empty() { - tracing_query.push(TracingQuery::Keywords(buf)); + tracing_query + .push(SearchFilter::eq(TracingSearchField::Address, buf)); buf = String::new(); } } else if ch == '"' { buf.push(ch); if in_quote { if !buf.is_empty() { - tracing_query.push(TracingQuery::Keywords(buf)); + tracing_query + .push(SearchFilter::eq(TracingSearchField::Address, buf)); buf = String::new(); } in_quote = false; @@ -105,7 +112,7 @@ impl TelemetryApi for Server { } } if !buf.is_empty() { - tracing_query.push(TracingQuery::Keywords(buf)); + tracing_query.push(SearchFilter::eq(TracingSearchField::Address, buf)); } } let before = params @@ -119,6 +126,11 @@ impl TelemetryApi for Server { .and_then(SnowflakeIdGenerator::from_timestamp) .unwrap_or(0); let values = params.get("values").is_some(); + + tracing_query.push(SearchFilter::lt(SearchField::Id, after)); + tracing_query.push(SearchFilter::gt(SearchField::Id, before)); + tracing_query.push(SearchFilter::End); + let store = &self .core .enterprise @@ -126,7 +138,11 @@ impl TelemetryApi for Server { .and_then(|e| e.trace_store.as_ref()) .ok_or_else(|| manage::unsupported("No tracing store has been configured"))? .store; - let span_ids = store.query_spans(&tracing_query, after, before).await?; + + let span_ids = self + .search_store() + .query(SearchQuery::new(SearchIndex::TracingSpan).with_filters(tracing_query)) + .await?; let (total, span_ids) = if limit > 0 { let offset = page.saturating_sub(1) * limit; diff --git a/crates/services/src/housekeeper/mod.rs b/crates/services/src/housekeeper/mod.rs index e604cb7f..a8acf00b 100644 --- a/crates/services/src/housekeeper/mod.rs +++ b/crates/services/src/housekeeper/mod.rs @@ -735,7 +735,9 @@ impl Purge for Server { // SPDX-License-Identifier: LicenseRef-SEL #[cfg(feature = "enterprise")] if let Some(trace_retention) = trace_retention - && let Err(err) = store.purge_spans(trace_retention).await + && let Err(err) = store + .purge_spans(trace_retention, self.search_store().into()) + .await { trc::error!(err.details("Failed to purge tracing spans")); } diff --git a/crates/services/src/task_manager/index.rs b/crates/services/src/task_manager/index.rs index b8d6384a..dea0f15f 100644 --- a/crates/services/src/task_manager/index.rs +++ b/crates/services/src/task_manager/index.rs @@ -5,26 +5,36 @@ */ use crate::task_manager::{IndexAction, Task}; -use common::Server; -use directory::{Type, backend::internal::manage::ManageDirectory}; -use groupware::cache::GroupwareCache; +use common::{ + Server, + telemetry::tracers::store::{TracingStore, build_span_document}, +}; +use directory::{QueryParams, Type, backend::internal::manage::ManageDirectory}; +use email::message::metadata::MessageMetadata; +use groupware::{cache::GroupwareCache, calendar::CalendarEvent, contact::ContactCard}; +use std::cmp::Ordering; use store::{ IterateParams, SerializeInfallible, U32_LEN, ValueKey, ahash::AHashMap, roaring::RoaringBitmap, + search::{IndexDocument, SearchField, SearchFilter, SearchQuery}, write::{ BatchBuilder, BlobOp, SearchIndex, TaskQueueClass, ValueClass, key::DeserializeBigEndian, now, }, }; -use trc::AddContext; +use trc::{AddContext, TaskQueueEvent}; use types::{ blob_hash::{BLOB_HASH_LEN, BlobHash}, collection::{Collection, SyncCollection}, + field::EmailField, }; pub(crate) trait SearchIndexTask: Sync + Send { - fn index(&self, tasks: &[Task]) -> impl Future + Send; + fn index( + &self, + tasks: &[Task], + ) -> impl Future> + Send; } pub trait ReindexIndexTask: Sync + Send { @@ -36,106 +46,241 @@ pub trait ReindexIndexTask: Sync + Send { ) -> impl Future> + Send; } +const NUM_INDEXES: usize = 5; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum TaskType { + Insert, + Delete, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum TaskStatus { + Success, + Failed, + Ignored, +} + +#[derive(Debug)] +pub(crate) struct IndexTaskResult { + index: SearchIndex, + task_type: TaskType, + status: TaskStatus, +} + impl SearchIndexTask for Server { - async fn index(&self, tasks: &[Task]) -> bool { - todo!() - // Obtain raw message - /*let op_start = Instant::now(); - let raw_message = if let Ok(Some(raw_message)) = self - .blob_store() - .get_blob(hash.as_slice(), 0..usize::MAX) - .await - { - raw_message - } else { - trc::event!( - TaskQueue(TaskQueueEvent::BlobNotFound), - AccountId = account_id, - DocumentId = document_id, - BlobId = hash.as_slice(), - ); - return false; - }; + async fn index(&self, tasks: &[Task]) -> Vec { + let mut results: Vec = Vec::with_capacity(tasks.len()); + let mut batch = BatchBuilder::new(); + let mut document_insertions: [Vec; NUM_INDEXES] = + std::array::from_fn(|_| Vec::new()); + let mut document_deletions: [AHashMap>; NUM_INDEXES] = + std::array::from_fn(|_| AHashMap::new()); - match self - .archive_by_property( - account_id, - Collection::Email, - document_id, - EmailField::Metadata.into(), - ) - .await - { - Ok(Some(metadata_)) => { - match metadata_.unarchive::() { - Ok(metadata) if metadata.blob_hash.0.as_slice() == hash.as_slice() => { - // Index message - /*let document = - FtsDocument::with_default_language(self.core.jmap.default_language) - .with_account_id(account_id) - .with_collection(Collection::Email) - .with_document_id(document_id) - .index_message(metadata, &raw_message); - if let Err(err) = self.core.storage.fts.index(document).await { - trc::error!( - err.account_id(account_id) - .document_id(document_id) - .details("Failed to index email in FTS index") - ); + for task in tasks { + if task.action.is_insert { + let (idx, document) = match task.action.index { + SearchIndex::Email => ( + 0, + build_email_document(self, task.account_id, task.document_id).await, + ), + SearchIndex::Calendar => ( + 1, + build_calendar_document(self, task.account_id, task.document_id).await, + ), + SearchIndex::Contacts => ( + 2, + build_contact_document(self, task.account_id, task.document_id).await, + ), + SearchIndex::File => { + // File indexing not implemented yet + continue; + } + SearchIndex::TracingSpan => ( + 4, + build_tracing_span_document(self, task.account_id, task.document_id).await, + ), + SearchIndex::InMemory => unreachable!(), + }; - return false; - }*/ - - trc::event!( - MessageIngest(MessageIngestEvent::FtsIndex), - AccountId = account_id, - Collection = Collection::Email, - DocumentId = document_id, - Elapsed = op_start.elapsed(), - ); + let result = match document { + Ok(Some(doc)) if !doc.is_empty() => { + document_insertions[idx].push(doc); + TaskStatus::Success } Err(err) => { trc::error!( - err.account_id(account_id) - .document_id(document_id) - .details("Failed to unarchive email metadata") + err.account_id(task.account_id) + .document_id(task.document_id) + .caused_by(trc::location!()) + .ctx(trc::Key::Collection, task.action.index.name()) + .details("Failed to build document for indexing") ); + TaskStatus::Failed } - _ => { - // The message was probably deleted or overwritten trc::event!( - TaskQueue(TaskQueueEvent::MetadataNotFound), - Details = "E-mail blob hash mismatch", - AccountId = account_id, - DocumentId = document_id, + TaskQueue(TaskQueueEvent::TaskIgnored), + Collection = task.action.index.name(), + Reason = "Nothing to index", + AccountId = task.account_id, + DocumentId = task.document_id, ); + TaskStatus::Ignored + } + }; + + results.push(IndexTaskResult { + task_type: TaskType::Insert, + index: task.action.index, + status: result, + }); + } else { + let idx = match task.action.index { + SearchIndex::Email => { + if let Err(err) = delete_email_metadata( + self, + &mut batch, + task.account_id, + task.document_id, + ) + .await + { + trc::error!( + err.account_id(task.account_id) + .document_id(task.document_id) + .caused_by(trc::location!()) + .details("Failed to delete email metadata from index") + ); + results.push(IndexTaskResult { + task_type: TaskType::Delete, + index: task.action.index, + status: TaskStatus::Failed, + }); + continue; + } + 0 + } + SearchIndex::Calendar => 1, + SearchIndex::Contacts => 2, + SearchIndex::File => 3, + SearchIndex::TracingSpan | SearchIndex::InMemory => unreachable!(), + }; + + document_deletions[idx] + .entry(task.account_id) + .or_default() + .push(task.document_id); + + results.push(IndexTaskResult { + task_type: TaskType::Delete, + index: task.action.index, + status: TaskStatus::Success, + }); + } + } + + // Commit deletion batch to data store + if !batch.is_empty() { + if let Err(err) = self.store().write(batch.build_all()).await { + trc::error!( + err.caused_by(trc::location!()) + .details("Failed to commit index deletions to data store") + ); + } + for r in results.iter_mut() { + if r.task_type == TaskType::Delete + && r.status == TaskStatus::Success + && r.index == SearchIndex::Email + { + r.status = TaskStatus::Failed; + } + } + return results; + } + + // Index documents + for (documents, index) in document_insertions.into_iter().zip([ + SearchIndex::Email, + SearchIndex::Calendar, + SearchIndex::Contacts, + SearchIndex::File, + SearchIndex::TracingSpan, + ]) { + if !documents.is_empty() + && let Err(err) = self.search_store().index(index, documents).await + { + trc::error!( + err.caused_by(trc::location!()) + .details("Failed to index documents") + .ctx(trc::Key::Collection, index.name()) + ); + for r in results.iter_mut() { + if r.task_type == TaskType::Delete && r.status == TaskStatus::Success { + r.status = TaskStatus::Failed; } } - - true + return results; } - Err(err) => { + } + + // Delete documents + for (accounts, index) in document_deletions.into_iter().zip([ + SearchIndex::Email, + SearchIndex::Calendar, + SearchIndex::Contacts, + ]) { + let multi_account = match accounts.len().cmp(&1) { + Ordering::Greater => true, + Ordering::Equal => false, + Ordering::Less => continue, + }; + + let mut query = SearchQuery::new(index); + if multi_account { + query.add_filter(SearchFilter::Or); + } + + for (account_id, document_ids) in accounts { + let multi_document = document_ids.len() > 1; + query + .add_filter(SearchFilter::And) + .add_filter(SearchFilter::eq(SearchField::AccountId, account_id)); + + if multi_document { + query.add_filter(SearchFilter::Or); + } + + for document_id in document_ids { + query.add_filter(SearchFilter::eq(SearchField::DocumentId, document_id)); + } + + if multi_document { + query.add_filter(SearchFilter::End); + } + query.add_filter(SearchFilter::End); + } + + if multi_account { + query.add_filter(SearchFilter::End); + } + + if let Err(err) = self.search_store().unindex(query).await { trc::error!( - err.account_id(account_id) - .document_id(document_id) - .caused_by(trc::location!()) - .details("Failed to retrieve email metadata") + err.caused_by(trc::location!()) + .details("Failed to delete documents from index") + .ctx(trc::Key::Collection, index.name()) ); + for r in results.iter_mut() { + if r.task_type == TaskType::Delete && r.status == TaskStatus::Success { + r.status = TaskStatus::Failed; + } + } + return results; + } + } - false - } - _ => { - // The message was probably deleted or overwritten - trc::event!( - TaskQueue(TaskQueueEvent::MetadataNotFound), - Details = "E-mail metadata not found", - AccountId = account_id, - DocumentId = document_id, - ); - true - } - }*/ + results } } @@ -295,3 +440,165 @@ impl ReindexIndexTask for Server { Ok(()) } } + +async fn build_email_document( + server: &Server, + account_id: u32, + document_id: u32, +) -> trc::Result> { + let Some(index_fields) = server.core.jmap.index_fields.get(&SearchIndex::Email) else { + return Ok(None); + }; + + match server + .archive_by_property( + account_id, + Collection::Email, + document_id, + EmailField::Metadata.into(), + ) + .await? + { + Some(metadata_) => { + let metadata = metadata_ + .unarchive::() + .caused_by(trc::location!())?; + + let raw_message = server + .blob_store() + .get_blob(metadata.blob_hash.0.as_slice(), 0..usize::MAX) + .await + .caused_by(trc::location!())? + .ok_or_else(|| { + trc::StoreEvent::NotFound + .into_err() + .details("Blob not found") + })?; + + Ok(Some(metadata.index_document( + &raw_message, + index_fields, + server.core.jmap.index_all_headers, + ))) + } + None => Ok(None), + } +} + +async fn build_calendar_document( + server: &Server, + account_id: u32, + document_id: u32, +) -> trc::Result> { + let Some(index_fields) = server.core.jmap.index_fields.get(&SearchIndex::Calendar) else { + return Ok(None); + }; + + match server + .archive(account_id, Collection::CalendarEvent, document_id) + .await? + { + Some(metadata_) => Ok(Some( + metadata_ + .unarchive::() + .caused_by(trc::location!())? + .index_document(index_fields), + )), + None => Ok(None), + } +} + +async fn build_contact_document( + server: &Server, + account_id: u32, + document_id: u32, +) -> trc::Result> { + let Some(index_fields) = server.core.jmap.index_fields.get(&SearchIndex::Contacts) else { + return Ok(None); + }; + + match server + .archive(account_id, Collection::ContactCard, document_id) + .await? + { + Some(metadata_) => Ok(Some( + metadata_ + .unarchive::() + .caused_by(trc::location!())? + .index_document(index_fields), + )), + None => Ok(None), + } +} + +async fn build_tracing_span_document( + server: &Server, + account_id: u32, + document_id: u32, +) -> trc::Result> { + let Some(index_fields) = server.core.jmap.index_fields.get(&SearchIndex::TracingSpan) else { + return Ok(None); + }; + + let span_id = ((account_id as u64) << 32) | document_id as u64; + let span = server.store().get_span(span_id).await?; + + if !span.is_empty() { + Ok(Some(build_span_document(span_id, span, index_fields))) + } else { + Ok(None) + } +} + +async fn delete_email_metadata( + server: &Server, + batch: &mut BatchBuilder, + account_id: u32, + document_id: u32, +) -> trc::Result<()> { + match server + .archive_by_property( + account_id, + Collection::Email, + document_id, + EmailField::Metadata.into(), + ) + .await? + { + Some(metadata) => { + let tenant_id = server + .core + .storage + .directory + .query(QueryParams::id(account_id).with_return_member_of(false)) + .await + .unwrap_or_default() + .and_then(|p| p.tenant()); + + batch + .with_account_id(account_id) + .with_collection(Collection::Email) + .with_document(document_id); + metadata + .unarchive::() + .caused_by(trc::location!())? + .unindex(batch, account_id, tenant_id); + } + None => { + trc::event!( + TaskQueue(TaskQueueEvent::MetadataNotFound), + Details = "E-mail metadata not found", + AccountId = account_id, + DocumentId = document_id, + ); + } + } + + Ok(()) +} + +impl IndexTaskResult { + pub fn is_done(&self) -> bool { + self.status != TaskStatus::Failed + } +} diff --git a/crates/services/src/task_manager/mod.rs b/crates/services/src/task_manager/mod.rs index d6014737..4f55a167 100644 --- a/crates/services/src/task_manager/mod.rs +++ b/crates/services/src/task_manager/mod.rs @@ -152,9 +152,17 @@ pub fn spawn_task_manager(inner: Arc) { if !locked_batch.is_empty() { let success = server.index(&locked_batch).await; - // Remove entries from queue - if success { + if success.iter().all(|t| t.is_done()) { delete_tasks(&server, &locked_batch).await; + } else { + // Remove successful entries from queue + let mut to_delete = Vec::with_capacity(locked_batch.len()); + for (task, result) in locked_batch.into_iter().zip(success.into_iter()) { + if result.is_done() { + to_delete.push(task); + } + } + delete_tasks(&server, &to_delete).await; } } } @@ -512,6 +520,7 @@ impl TaskQueueManager for Server { _ => { trc::event!( TaskQueue(TaskQueueEvent::TaskIgnored), + Details = event.action.name(), AccountId = event.account_id, DocumentId = event.document_id, ); @@ -554,3 +563,15 @@ async fn delete_tasks(server: &Server, tasks: &[T]) { } } } + +impl TaskAction { + pub fn name(&self) -> &'static str { + match self { + TaskAction::UpdateIndex(_) => "UpdateIndex", + TaskAction::BayesTrain(_) => "BayesTrain", + TaskAction::SendAlarm(_) => "SendAlarm", + TaskAction::SendImip => "SendImip", + TaskAction::MergeThreads(_) => "MergeThreads", + } + } +} diff --git a/crates/store/src/dispatch/search.rs b/crates/store/src/dispatch/search.rs index eb1bddfb..eef7d0c2 100644 --- a/crates/store/src/dispatch/search.rs +++ b/crates/store/src/dispatch/search.rs @@ -9,6 +9,7 @@ use crate::{ SearchStore, backend::elastic::query, search::{IndexDocument, SearchComparator, SearchDocumentId, SearchFilter, SearchQuery}, + write::SearchIndex, }; use trc::AddContext; use types::collection::Collection; @@ -32,7 +33,11 @@ impl SearchStore { .caused_by(trc::location!())*/ } - pub async fn index(&self, documents: Vec) -> trc::Result<()> { + pub async fn index( + &self, + index: SearchIndex, + documents: Vec, + ) -> trc::Result<()> { todo!() /*match self { SearchStore::Store(store) => store.index_insert(document).await, diff --git a/crates/store/src/search/mod.rs b/crates/store/src/search/mod.rs index eb5d1c3b..08242928 100644 --- a/crates/store/src/search/mod.rs +++ b/crates/store/src/search/mod.rs @@ -35,6 +35,7 @@ pub enum SearchField { Calendar(CalendarSearchField), Contact(ContactSearchField), File(FileSearchField), + Tracing(TracingSearchField), } #[derive(Debug, Clone, PartialEq, Eq, Hash)] @@ -86,12 +87,21 @@ pub enum FileSearchField { Content, } +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum TracingSearchField { + EventType, + QueueId, + Address, + RemoteIp, +} + #[derive(Debug, Clone, PartialEq, Eq)] pub enum SearchValue { Text { value: String, language: Language }, Int(i64), Uint(u64), Boolean(bool), + Keywords(Vec), } pub trait SearchDocumentId: Sized { @@ -351,6 +361,29 @@ impl IndexDocument { .insert(field.into(), SearchValue::Uint(value.into())); } + pub fn insert_keyword( + &mut self, + field: impl Into, + keyword: impl Into, + ) { + let search_field = field.into(); + + match self.fields.entry(search_field) { + Entry::Occupied(mut entry) => { + if let SearchValue::Keywords(existing_keywords) = entry.get_mut() { + existing_keywords.push(keyword.into()); + } + } + Entry::Vacant(entry) => { + entry.insert(SearchValue::Keywords(vec![keyword.into()])); + } + } + } + + pub fn is_empty(&self) -> bool { + self.fields.is_empty() + } + pub fn has_field(&self, field: &SearchField) -> bool { self.fields.contains_key(field) } @@ -389,6 +422,11 @@ impl SearchQuery { self } + pub fn add_filter(&mut self, filter: SearchFilter) -> &mut Self { + self.filters.push(filter); + self + } + pub fn with_comparator(mut self, comparator: SearchComparator) -> Self { self.comparators.push(comparator); self @@ -438,6 +476,12 @@ impl From for SearchField { } } +impl From for SearchField { + fn from(field: TracingSearchField) -> Self { + SearchField::Tracing(field) + } +} + impl From for SearchValue { fn from(value: u64) -> Self { SearchValue::Uint(value) diff --git a/crates/store/src/write/key.rs b/crates/store/src/write/key.rs index a9608561..c31d029e 100644 --- a/crates/store/src/write/key.rs +++ b/crates/store/src/write/key.rs @@ -683,6 +683,17 @@ impl SearchIndex { } } + pub fn name(&self) -> &'static str { + match self { + SearchIndex::Email => "email", + SearchIndex::Calendar => "calendar", + SearchIndex::Contacts => "contacts", + SearchIndex::File => "file", + SearchIndex::TracingSpan => "tracing", + SearchIndex::InMemory => "in_memory", + } + } + pub fn try_from_str(value: &str) -> Option { match value { "email" => Some(SearchIndex::Email),