diff --git a/crates/common/src/config/jmap/settings.rs b/crates/common/src/config/jmap/settings.rs index ae7c7298..bbfa7f6f 100644 --- a/crates/common/src/config/jmap/settings.rs +++ b/crates/common/src/config/jmap/settings.rs @@ -44,6 +44,7 @@ pub struct JmapConfig { pub mail_parse_max_items: usize, pub mail_max_size: usize, pub mail_autoexpunge_after: Option, + pub email_submission_autoexpunge_after: Option, pub contact_parse_max_items: usize, pub calendar_parse_max_items: usize, @@ -276,6 +277,10 @@ impl JmapConfig { .property_or_default::>("email.auto-expunge", "30d") .map(|d| d.map(|d| d.as_secs())) .unwrap_or_default(), + email_submission_autoexpunge_after: config + .property_or_default::>("email-submission.auto-expunge", "3d") + .map(|d| d.map(|d| d.as_secs())) + .unwrap_or_default(), sieve_max_script_name: config .property("sieve.untrusted.limits.name-length") .unwrap_or(512), diff --git a/crates/common/src/core.rs b/crates/common/src/core.rs index c338aa93..228421be 100644 --- a/crates/common/src/core.rs +++ b/crates/common/src/core.rs @@ -354,7 +354,7 @@ impl Server { collection: Collection::Email.into(), document_id: 0, class: ValueClass::IndexProperty(IndexPropertyClass::Integer { - property: EmailField::Stats.into(), + property: EmailField::ReceivedToSize.into(), value: 0, }), }, @@ -363,7 +363,7 @@ impl Server { collection: Collection::Email.into(), document_id: u32::MAX, class: ValueClass::IndexProperty(IndexPropertyClass::Integer { - property: EmailField::Stats.into(), + property: EmailField::ReceivedToSize.into(), value: u64::MAX, }), }, diff --git a/crates/common/src/manager/restore.rs b/crates/common/src/manager/restore.rs index 8b67fce1..e5ad7c22 100644 --- a/crates/common/src/manager/restore.rs +++ b/crates/common/src/manager/restore.rs @@ -13,8 +13,8 @@ use std::{ use store::{ BlobStore, Key, LogKey, SUBSPACE_LOGS, SerializeInfallible, Store, U32_LEN, write::{ - AnyClass, BatchBuilder, BlobOp, DirectoryClass, InMemoryClass, Operation, TaskQueueClass, - ValueClass, ValueOp, key::DeserializeBigEndian, now, + AnyClass, BatchBuilder, BlobOp, DirectoryClass, InMemoryClass, Operation, SearchIndex, + TaskQueueClass, ValueClass, ValueOp, key::DeserializeBigEndian, now, }, }; use store::{ @@ -148,7 +148,7 @@ async fn restore_file(store: Store, blob_store: BlobStore, path: &Path) { batch.set( ValueClass::TaskQueue(TaskQueueClass::UpdateIndex { due, - collection: Collection::Email, + index: SearchIndex::Email, is_insert: true, }), 0u64.serialize(), diff --git a/crates/common/src/storage/index.rs b/crates/common/src/storage/index.rs index e8e37372..1d9fd28a 100644 --- a/crates/common/src/storage/index.rs +++ b/crates/common/src/storage/index.rs @@ -5,7 +5,6 @@ */ use crate::{auth::AccessToken, sharing::notification::ShareNotification}; -use ahash::AHashSet; use rkyv::{ option::ArchivedOption, primitive::{ArchivedU32, ArchivedU64}, @@ -15,8 +14,8 @@ use std::{borrow::Cow, fmt::Debug}; use store::{ Serialize, SerializeInfallible, write::{ - Archive, Archiver, BatchBuilder, BlobOp, DirectoryClass, IntoOperations, TaskQueueClass, - ValueClass, now, + Archive, Archiver, BatchBuilder, BlobOp, DirectoryClass, IntoOperations, SearchIndex, + TaskQueueClass, ValueClass, now, }, }; use types::{ @@ -38,7 +37,8 @@ pub enum IndexValue<'x> { value: IndexItem<'x>, }, SearchIndex { - hashes: AHashSet, + index: SearchIndex, + hash: u64, }, Blob { value: BlobHash, @@ -376,11 +376,11 @@ fn build_index( } } } - IndexValue::SearchIndex { .. } => { + IndexValue::SearchIndex { index, .. } => { batch.set( ValueClass::TaskQueue(TaskQueueClass::UpdateIndex { due: now(), - collection: batch.last_collection().unwrap_or(Collection::None), + index, is_insert: set, }), vec![], @@ -508,20 +508,15 @@ fn merge_index( batch.index(field, new_value.into_owned()); } } - ( - IndexValue::SearchIndex { hashes: old_hashes }, - IndexValue::SearchIndex { hashes: new_hashes }, - ) => { - if old_hashes != new_hashes { - batch.set( - ValueClass::TaskQueue(TaskQueueClass::UpdateIndex { - due: now(), - collection: batch.last_collection().unwrap_or(Collection::None), - is_insert: true, - }), - vec![], - ); - } + (IndexValue::SearchIndex { index, .. }, IndexValue::SearchIndex { .. }) => { + batch.set( + ValueClass::TaskQueue(TaskQueueClass::UpdateIndex { + due: now(), + index, + is_insert: true, + }), + vec![], + ); } ( IndexValue::Property { diff --git a/crates/dav/src/calendar/mod.rs b/crates/dav/src/calendar/mod.rs index 33b51ad1..064d3762 100644 --- a/crates/dav/src/calendar/mod.rs +++ b/crates/dav/src/calendar/mod.rs @@ -23,7 +23,7 @@ use dav_proto::schema::{ use groupware::scheduling::ItipError; use hyper::StatusCode; use trc::AddContext; -use types::{collection::Collection, field::CalendarField}; +use types::{collection::Collection, field::CalendarEventField}; pub(crate) static CALENDAR_CONTAINER_PROPS: [DavProperty; 31] = [ DavProperty::WebDav(WebDavProperty::CreationDate), @@ -99,7 +99,7 @@ pub(crate) async fn assert_is_unique_uid( .document_ids_matching( account_id, Collection::CalendarEvent, - CalendarField::Uid, + CalendarEventField::Uid, uid.as_bytes(), ) .await diff --git a/crates/email/src/message/copy.rs b/crates/email/src/message/copy.rs index 4edfcfe1..1103a5d5 100644 --- a/crates/email/src/message/copy.rs +++ b/crates/email/src/message/copy.rs @@ -5,17 +5,21 @@ */ use super::{ - index::{MAX_SORT_FIELD_LENGTH, TrimTextValue, VisitText}, ingest::{EmailIngest, IngestedEmail}, metadata::{MessageData, MessageMetadata}, }; use crate::{ mailbox::UidMailbox, - message::ingest::{MergeThreadTask, ThreadInfo}, + message::{ + index::extractors::VisitText, + ingest::{MergeThreadTask, ThreadInfo}, + }, }; use common::{Server, auth::ResourceToken, storage::index::ObjectIndexBuilder}; use mail_parser::{HeaderName, HeaderValue, parsers::fields::thread::thread_name}; -use store::write::{BatchBuilder, IndexPropertyClass, TaskQueueClass, ValueClass, now}; +use store::write::{ + BatchBuilder, IndexPropertyClass, SearchIndex, TaskQueueClass, ValueClass, now, +}; use trc::AddContext; use types::{ blob::{BlobClass, BlobId}, @@ -123,8 +127,7 @@ impl EmailCopy for Server { list.first().unwrap().as_ref() } _ => "", - }) - .trim_text(MAX_SORT_FIELD_LENGTH); + }); } _ => (), } @@ -196,7 +199,11 @@ impl EmailCopy for Server { ThreadInfo::serialize(thread_id, &message_ids), ) .set( - ValueClass::TaskQueue(TaskQueueClass::IndexEmail { due: now() }), + ValueClass::TaskQueue(TaskQueueClass::UpdateIndex { + index: SearchIndex::Email, + due: now(), + is_insert: true, + }), MergeThreadTask::new(thread_result).serialize(), ); metadata diff --git a/crates/email/src/message/delete.rs b/crates/email/src/message/delete.rs index ca9afc9c..da024da0 100644 --- a/crates/email/src/message/delete.rs +++ b/crates/email/src/message/delete.rs @@ -11,15 +11,15 @@ use groupware::calendar::storage::ItipAutoExpunge; use std::future::Future; use store::rand::prelude::SliceRandom; use store::write::key::DeserializeBigEndian; -use store::write::{IndexPropertyClass, TaskQueueClass, now}; -use store::{IterateParams, SerializeInfallible, U32_LEN, ValueKey}; +use store::write::{IndexPropertyClass, SearchIndex, TaskQueueClass, now}; +use store::{IterateParams, SerializeInfallible, U32_LEN, U64_LEN, ValueKey}; use store::{ roaring::RoaringBitmap, write::{BatchBuilder, ValueClass}, }; use trc::AddContext; use types::collection::{Collection, VanishedCollection}; -use types::field::{EmailField, Field}; +use types::field::{EmailField, EmailSubmissionField, Field}; pub trait EmailDeletion: Sync + Send { fn emails_delete( @@ -33,6 +33,12 @@ pub trait EmailDeletion: Sync + Send { fn purge_account(&self, account_id: u32) -> impl Future + Send; + fn purge_email_submissions( + &self, + account_id: u32, + hold_period: u64, + ) -> impl Future> + Send; + fn emails_auto_expunge( &self, account_id: u32, @@ -72,7 +78,11 @@ impl EmailDeletion for Server { .custom(ObjectIndexBuilder::<_, ()>::new().with_current(metadata)) .caused_by(trc::location!())? .set( - ValueClass::TaskQueue(TaskQueueClass::UnindexEmail { due }), + ValueClass::TaskQueue(TaskQueueClass::UpdateIndex { + index: SearchIndex::Email, + due, + is_insert: false, + }), 0u64.serialize(), ) .commit_point(); @@ -164,6 +174,16 @@ impl EmailDeletion for Server { ); } + // Delete old e-mail submissions + if let Some(hold_period) = self.core.jmap.email_submission_autoexpunge_after + && let Err(err) = self.purge_email_submissions(account_id, hold_period).await + { + trc::error!( + err.details("Failed to auto-expunge e-mail submissions.") + .account_id(account_id) + ); + } + // Purge changelogs if let Err(err) = self .delete_changes( @@ -218,7 +238,7 @@ impl EmailDeletion for Server { collection: Collection::Email.into(), document_id: 0, class: ValueClass::IndexProperty(IndexPropertyClass::Integer { - property: EmailField::Stats.into(), + property: EmailField::ReceivedToSize.into(), value: 0, }), }, @@ -227,7 +247,7 @@ impl EmailDeletion for Server { collection: Collection::Email.into(), document_id: u32::MAX, class: ValueClass::IndexProperty(IndexPropertyClass::Integer { - property: EmailField::Stats.into(), + property: EmailField::ReceivedToSize.into(), value: now().saturating_sub(hold_period), }), }, @@ -269,6 +289,78 @@ impl EmailDeletion for Server { Ok(()) } + async fn purge_email_submissions(&self, account_id: u32, hold_period: u64) -> trc::Result<()> { + // Filter messages by received date + let mut destroy_ids = Vec::new(); + self.store() + .iterate( + IterateParams::new( + ValueKey { + account_id, + collection: Collection::EmailSubmission.into(), + document_id: 0, + class: ValueClass::IndexProperty(IndexPropertyClass::Integer { + property: EmailSubmissionField::Metadata.into(), + value: 0, + }), + }, + ValueKey { + account_id, + collection: Collection::Email.into(), + document_id: u32::MAX, + class: ValueClass::IndexProperty(IndexPropertyClass::Integer { + property: EmailSubmissionField::Metadata.into(), + value: now().saturating_sub(hold_period), + }), + }, + ) + .ascending() + .no_values(), + |key, _| { + destroy_ids.push(( + key.deserialize_be_u32(key.len() - U32_LEN)?, + key.deserialize_be_u64(key.len() - U32_LEN - U64_LEN)?, + )); + + Ok(true) + }, + ) + .await + .caused_by(trc::location!())?; + + if destroy_ids.is_empty() { + return Ok(()); + } + + trc::event!( + Purge(trc::PurgeEvent::AutoExpunge), + Collection = Collection::EmailSubmission.as_str(), + AccountId = account_id, + Total = destroy_ids.len(), + ); + + // Delete messages + let mut batch = BatchBuilder::new(); + batch + .with_account_id(account_id) + .with_collection(Collection::EmailSubmission); + + for (document_id, send_at) in destroy_ids { + batch + .with_document(document_id) + .clear(EmailSubmissionField::Metadata) + .clear(ValueClass::IndexProperty(IndexPropertyClass::Integer { + property: EmailSubmissionField::Metadata.into(), + value: send_at, + })) + .commit_point(); + } + + self.commit_batch(batch).await?; + + Ok(()) + } + /*async fn emails_purge_tombstoned(&self, account_id: u32) -> trc::Result<()> { // Obtain tombstoned messages let tombstoned_ids = self diff --git a/crates/email/src/message/index.rs b/crates/email/src/message/index.rs deleted file mode 100644 index 502492e3..00000000 --- a/crates/email/src/message/index.rs +++ /dev/null @@ -1,1064 +0,0 @@ -/* - * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC - * - * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL - */ - -use std::borrow::Cow; - -use super::metadata::{ - ArchivedMessageData, ArchivedMessageMetadata, ArchivedMessageMetadataContents, - ArchivedMessageMetadataPart, ArchivedMetadataPartType, DecodedPartContent, MessageData, - MessageMetadata, MessageMetadataPart, -}; -use common::storage::index::{IndexValue, IndexableObject, ObjectIndexBuilder}; -use mail_parser::{ - Addr, Address, ArchivedAddress, ArchivedHeaderName, ArchivedHeaderValue, Group, HeaderName, - HeaderValue, - core::rkyv::ArchivedGetHeader, - decoders::html::html_to_text, - parsers::{fields::thread::thread_name, preview::preview_text}, -}; -use nlp::language::Language; -use rkyv::option::ArchivedOption; -use store::{ - Serialize, SerializeInfallible, - backend::MAX_TOKEN_LENGTH, - write::{Archiver, BatchBuilder, BlobOp, DirectoryClass, IndexPropertyClass, ValueClass}, -}; -use trc::AddContext; -use types::{blob_hash::BlobHash, collection::SyncCollection, field::EmailField}; - -pub const MAX_MESSAGE_PARTS: usize = 1000; -pub const MAX_ID_LENGTH: usize = 100; -pub const MAX_SORT_FIELD_LENGTH: usize = 255; -pub const MAX_STORED_FIELD_LENGTH: usize = 512; -pub const PREVIEW_LENGTH: usize = 256; - -impl MessageMetadata { - #[inline(always)] - pub fn root_part(&self) -> &MessageMetadataPart { - &self.contents[0].parts[0] - } - - pub fn index( - self, - batch: &mut BatchBuilder, - account_id: u32, - tenant_id: Option, - set: bool, - ) -> trc::Result<()> { - if set { - // Serialize metadata - batch.set( - ValueClass::IndexProperty(IndexPropertyClass::Integer { - property: EmailField::Stats.into(), - value: self.received_at, - }), - self.size.serialize(), - ); - } else { - // Delete metadata - batch - .clear(EmailField::Metadata) - .clear(ValueClass::IndexProperty(IndexPropertyClass::Integer { - property: EmailField::Stats.into(), - value: self.received_at, - })); - } - - // Index properties - let quota = if set { - self.size as i64 - } else { - -(self.size as i64) - }; - batch.add(DirectoryClass::UsedQuota(account_id), quota); - if let Some(tenant_id) = tenant_id { - batch.add(DirectoryClass::UsedQuota(tenant_id), quota); - } - - // Link blob - if set { - batch.set( - BlobOp::Link { - hash: self.blob_hash.clone(), - }, - Vec::new(), - ); - } else { - batch.clear(BlobOp::Link { - hash: self.blob_hash.clone(), - }); - } - - if set { - batch.set(EmailField::Metadata, Archiver::new(self).serialize()?); - } - - Ok(()) - } - - /* - - if self.has_attachments { - if set { - batch.tag(EmailField::HasAttachment); - } else { - batch.untag(EmailField::HasAttachment); - } - } - - // Index headers - self.index_headers(batch, set); - - - metadata.index_headers(self, true); - - // Store and index hasAttachment property - if has_attachments { - self.tag(EmailField::HasAttachment); - } - - - fn index_headers(&self, batch: &mut BatchBuilder, set: bool) { - let mut seen_headers = [false; 40]; - for header in self.root_part().headers.iter().rev() { - if matches!(header.name, HeaderName::Other(_)) { - continue; - } - - match header.name { - HeaderName::MessageId => { - header.value.visit_text(|id| { - // Add ids to inverted index - if id.len() < MAX_ID_LENGTH { - if set { - batch.index(EmailField::References, encode_message_id(id)); - } else { - batch.unindex(EmailField::References, encode_message_id(id)); - } - } - }); - } - HeaderName::InReplyTo | HeaderName::References | HeaderName::ResentMessageId => { - header.value.visit_text(|id| { - // Add ids to inverted index - if id.len() < MAX_ID_LENGTH { - if set { - batch.index(EmailField::References, id.serialize()); - } else { - batch.unindex(EmailField::References, id.serialize()); - } - } - }); - } - HeaderName::From | HeaderName::To | HeaderName::Cc | HeaderName::Bcc => { - if !seen_headers[header.name.id() as usize] { - let property = match &header.name { - HeaderName::From => EmailField::From, - HeaderName::To => EmailField::To, - HeaderName::Cc => EmailField::Cc, - HeaderName::Bcc => EmailField::Bcc, - _ => unreachable!(), - }; - let mut sort_text = SortedAddressBuilder::new(); - let mut found_addr = false; - - header.value.visit_addresses(|element, value| { - if !found_addr { - match element { - AddressElement::Name => { - found_addr = !sort_text.push(value); - } - AddressElement::Address => { - sort_text.push(value); - found_addr = true; - } - AddressElement::GroupName => (), - } - } - }); - - // Add address to inverted index - if set { - batch.index(property, sort_text.build()); - } else { - batch.unindex(property, sort_text.build()); - } - seen_headers[header.name.id() as usize] = true; - } - } - HeaderName::Date => { - if !seen_headers[header.name.id() as usize] { - if let HeaderValue::DateTime(datetime) = &header.value { - let value = (datetime.to_timestamp() as u64).serialize(); - if set { - batch.index(EmailField::SentAt, value); - } else { - batch.unindex(EmailField::SentAt, value); - } - } - seen_headers[header.name.id() as usize] = true; - } - } - HeaderName::Subject => { - if !seen_headers[header.name.id() as usize] { - // Index subject - let subject = match &header.value { - HeaderValue::Text(text) => text.clone(), - HeaderValue::TextList(list) if !list.is_empty() => { - list.first().unwrap().clone() - } - _ => "".into(), - }; - - // Index thread name - let thread_name = thread_name(&subject); - let thread_name = if !thread_name.is_empty() { - thread_name.trim_text(MAX_SORT_FIELD_LENGTH) - } else { - "!" - } - .serialize(); - - if set { - batch.index(EmailField::Subject, thread_name); - } else { - batch.unindex(EmailField::Subject, thread_name); - } - - seen_headers[header.name.id() as usize] = true; - } - } - - _ => (), - } - } - - // Add subject to index if missing - if !seen_headers[HeaderName::Subject.id() as usize] { - if set { - batch.index(EmailField::Subject, "!".serialize()); - } else { - batch.unindex(EmailField::Subject, "!".serialize()); - } - } - } - - */ -} - -fn encode_message_id(message_id: &str) -> Vec { - let mut msg_id = Vec::with_capacity(message_id.len() + 1); - msg_id.extend_from_slice(message_id.as_bytes()); - msg_id.push(0); - msg_id -} - -impl ArchivedMessageMetadata { - #[inline(always)] - pub fn root_part(&self) -> &ArchivedMessageMetadataPart { - &self.contents[0].parts[0] - } - - /*pub fn index( - &self, - batch: &mut BatchBuilder, - account_id: u32, - tenant_id: Option, - set: bool, - ) -> trc::Result<()> { - if set { - // Serialize metadata - batch - .index(EmailField::Size, u32::from(self.size).serialize()) - .index( - EmailField::ReceivedAt, - u64::from(self.received_at).serialize(), - ); - } else { - // Delete metadata - batch - .clear(EmailField::Metadata) - .unindex(EmailField::Size, u32::from(self.size).serialize()) - .unindex( - EmailField::ReceivedAt, - u64::from(self.received_at).serialize(), - ); - } - - // Index properties - let quota = if set { - u32::from(self.size) as i64 - } else { - -(u32::from(self.size) as i64) - }; - batch.add(DirectoryClass::UsedQuota(account_id), quota); - if let Some(tenant_id) = tenant_id { - batch.add(DirectoryClass::UsedQuota(tenant_id), quota); - } - - if self.has_attachments { - if set { - batch.tag(EmailField::HasAttachment); - } else { - batch.untag(EmailField::HasAttachment); - } - } - - // Index headers - self.index_headers(batch, set); - - // Link blob - let hash = BlobHash::from(&self.blob_hash); - if set { - batch.set(BlobOp::Link { hash }, Vec::new()); - } else { - batch.clear(BlobOp::Link { hash }); - } - - Ok(()) - } - - fn index_headers(&self, batch: &mut BatchBuilder, set: bool) { - let mut seen_headers = [false; 40]; - for header in self.root_part().headers.iter().rev() { - if matches!(header.name, ArchivedHeaderName::Other(_)) { - continue; - } - - match header.name { - ArchivedHeaderName::MessageId => { - header.value.visit_text(|id| { - // Add ids to inverted index - if id.len() < MAX_ID_LENGTH { - if set { - batch.index(EmailField::References, encode_message_id(id)); - } else { - batch.unindex(EmailField::References, encode_message_id(id)); - } - } - }); - } - ArchivedHeaderName::InReplyTo - | ArchivedHeaderName::References - | ArchivedHeaderName::ResentMessageId => { - header.value.visit_text(|id| { - // Add ids to inverted index - if id.len() < MAX_ID_LENGTH { - if set { - batch.index(EmailField::References, id.serialize()); - } else { - batch.unindex(EmailField::References, id.serialize()); - } - } - }); - } - ArchivedHeaderName::From - | ArchivedHeaderName::To - | ArchivedHeaderName::Cc - | ArchivedHeaderName::Bcc => { - if !seen_headers[header.name.id() as usize] { - let property = match &header.name { - ArchivedHeaderName::From => EmailField::From, - ArchivedHeaderName::To => EmailField::To, - ArchivedHeaderName::Cc => EmailField::Cc, - ArchivedHeaderName::Bcc => EmailField::Bcc, - _ => unreachable!(), - }; - let mut sort_text = SortedAddressBuilder::new(); - let mut found_addr = false; - - header.value.visit_addresses(|element, value| { - if !found_addr { - match element { - AddressElement::Name => { - found_addr = !sort_text.push(value); - } - AddressElement::Address => { - sort_text.push(value); - found_addr = true; - } - AddressElement::GroupName => (), - } - } - }); - - // Add address to inverted index - if set { - batch.index(property, sort_text.build()); - } else { - batch.unindex(property, sort_text.build()); - } - seen_headers[header.name.id() as usize] = true; - } - } - ArchivedHeaderName::Date => { - if !seen_headers[header.name.id() as usize] { - if let ArchivedHeaderValue::DateTime(datetime) = &header.value { - let value = (mail_parser::DateTime::from(datetime).to_timestamp() - as u64) - .serialize(); - if set { - batch.index(EmailField::SentAt, value); - } else { - batch.unindex(EmailField::SentAt, value); - } - } - seen_headers[header.name.id() as usize] = true; - } - } - ArchivedHeaderName::Subject => { - if !seen_headers[header.name.id() as usize] { - // Index subject - let subject = match &header.value { - ArchivedHeaderValue::Text(text) => text.as_str(), - ArchivedHeaderValue::TextList(list) if !list.is_empty() => { - list.first().unwrap().as_str() - } - _ => "", - }; - - // Index thread name - let thread_name = thread_name(subject); - let thread_name = if !thread_name.is_empty() { - thread_name.trim_text(MAX_SORT_FIELD_LENGTH) - } else { - "!" - } - .serialize(); - - if set { - batch.index(EmailField::Subject, thread_name); - } else { - batch.unindex(EmailField::Subject, thread_name); - } - - seen_headers[header.name.id() as usize] = true; - } - } - - _ => (), - } - } - - // Add subject to index if missing - if !seen_headers[HeaderName::Subject.id() as usize] { - if set { - batch.index(EmailField::Subject, "!".serialize()); - } else { - batch.unindex(EmailField::Subject, "!".serialize()); - } - } - } - - */ -} - -impl ArchivedMessageMetadataContents { - pub fn is_html_part(&self, part_id: u16) -> bool { - self.html_body.iter().any(|&id| id == part_id) - } - - pub fn is_text_part(&self, part_id: u16) -> bool { - self.text_body.iter().any(|&id| id == part_id) - } -} -#[derive(Debug)] -pub struct SortedAddressBuilder { - last_is_space: bool, - pub buf: String, -} - -pub(super) trait IndexMessage { - #[allow(clippy::too_many_arguments)] - fn index_message( - &mut self, - account_id: u32, - tenant_id: Option, - message: mail_parser::Message<'_>, - blob_hash: BlobHash, - data: MessageData, - received_at: u64, - ) -> trc::Result<&mut Self>; -} - -impl IndexMessage for BatchBuilder { - fn index_message( - &mut self, - account_id: u32, - tenant_id: Option, - message: mail_parser::Message<'_>, - blob_hash: BlobHash, - data: MessageData, - received_at: u64, - ) -> trc::Result<&mut Self> { - // Index size - self.set( - ValueClass::IndexProperty(IndexPropertyClass::Integer { - property: EmailField::Stats.into(), - value: received_at, - }), - (message.raw_message.len() as u32).serialize(), - ) - .add( - DirectoryClass::UsedQuota(account_id), - message.raw_message.len() as i64, - ); - if let Some(tenant_id) = tenant_id { - self.add( - DirectoryClass::UsedQuota(tenant_id), - message.raw_message.len() as i64, - ); - } - - let mut has_attachments = false; - let mut preview = None; - let preview_part_id = message - .text_body - .first() - .or_else(|| message.html_body.first()) - .copied() - .unwrap_or(u32::MAX); - - for (part_id, part) in message.parts.iter().take(MAX_MESSAGE_PARTS).enumerate() { - let part_id = part_id as u32; - match &part.body { - mail_parser::PartType::Text(text) => { - if part_id == preview_part_id { - preview = - preview_text(text.replace('\r', "").into(), PREVIEW_LENGTH).into(); - } - - if !message.text_body.contains(&part_id) - && !message.html_body.contains(&part_id) - { - has_attachments = true; - } - } - mail_parser::PartType::Html(html) => { - let text = html_to_text(html); - if part_id == preview_part_id { - preview = - preview_text(text.replace('\r', "").into(), PREVIEW_LENGTH).into(); - } - - if !message.text_body.contains(&part_id) - && !message.html_body.contains(&part_id) - { - has_attachments = true; - } - } - mail_parser::PartType::Binary(_) | mail_parser::PartType::Message(_) - if !has_attachments => - { - has_attachments = true; - } - _ => {} - } - } - - // Build metadata - let root_part = message.root_part(); - let metadata = MessageMetadata { - preview: preview.unwrap_or_default().into_owned(), - size: message.raw_message.len() as u32, - raw_headers: message - .raw_message - .as_ref() - .get(root_part.offset_header as usize..root_part.offset_body as usize) - .unwrap_or_default() - .to_vec(), - contents: vec![], - received_at, - has_attachments, - blob_hash, - } - .with_contents(message); - - // Link blob - self.set( - BlobOp::Link { - hash: metadata.blob_hash.clone(), - }, - Vec::new(), - ); - - // Store message data - self.custom(ObjectIndexBuilder::<(), _>::new().with_changes(data)) - .caused_by(trc::location!())?; - - // Store message metadata - self.set( - EmailField::Metadata, - Archiver::new(metadata) - .serialize() - .caused_by(trc::location!())?, - ); - - Ok(self) - } -} - -impl IndexableObject for MessageData { - fn index_values(&self) -> impl Iterator> { - [ - IndexValue::LogItem { - sync_collection: SyncCollection::Email, - prefix: self.thread_id.into(), - }, - IndexValue::LogContainerProperty { - sync_collection: SyncCollection::Thread, - ids: vec![self.thread_id], - }, - IndexValue::LogContainerProperty { - sync_collection: SyncCollection::Email, - ids: self.mailboxes.iter().map(|m| m.mailbox_id).collect(), - }, - ] - .into_iter() - } -} - -impl IndexableObject for &ArchivedMessageData { - fn index_values(&self) -> impl Iterator> { - [ - IndexValue::LogItem { - sync_collection: SyncCollection::Email, - prefix: self.thread_id.to_native().into(), - }, - IndexValue::LogContainerProperty { - sync_collection: SyncCollection::Thread, - ids: vec![self.thread_id.to_native()], - }, - IndexValue::LogContainerProperty { - sync_collection: SyncCollection::Email, - ids: self - .mailboxes - .iter() - .map(|m| m.mailbox_id.to_native()) - .collect(), - }, - ] - .into_iter() - } -} - -pub trait IndexMessageText<'x>: Sized { - fn index_message(self, message: &'x ArchivedMessageMetadata, raw_message: &'x [u8]) -> Self; -} - -/*impl<'x> IndexMessageText<'x> for FtsDocument<'x, mail_parser::HeaderName<'x>> { - fn index_message( - mut self, - message: &'x ArchivedMessageMetadata, - raw_message: &'x [u8], - ) -> Self { - let mut language = Language::Unknown; - let message_contents = &message.contents[0]; - - for (part_id, part) in message_contents - .parts - .iter() - .take(MAX_MESSAGE_PARTS) - .enumerate() - { - let part_language = part.language().unwrap_or(language); - if part_id == 0 { - language = part_language; - - for header in part.headers.iter().rev() { - if matches!(header.name, ArchivedHeaderName::Other(_)) { - continue; - } - // Index hasHeader property - self.index_keyword(Field::Keyword, header.name.as_str().to_ascii_lowercase()); - - match &header.name { - ArchivedHeaderName::MessageId - | ArchivedHeaderName::InReplyTo - | ArchivedHeaderName::References - | ArchivedHeaderName::ResentMessageId => { - header.value.visit_text(|id| { - // Index ids without stemming - if id.len() < MAX_TOKEN_LENGTH { - self.index_keyword( - Field::Header(mail_parser::HeaderName::from(&header.name)), - id.to_string(), - ); - } - }); - } - ArchivedHeaderName::From - | ArchivedHeaderName::To - | ArchivedHeaderName::Cc - | ArchivedHeaderName::Bcc => { - header.value.visit_addresses(|_, value| { - // Index an address name or email without stemming - self.index_tokenized( - Field::Header(mail_parser::HeaderName::from(&header.name)), - value.to_string(), - ); - }); - } - ArchivedHeaderName::Subject => { - // Index subject for FTS - if let Some(subject) = header.value.as_text() { - self.index( - Field::Header(mail_parser::HeaderName::Subject), - subject, - language, - ); - } - } - ArchivedHeaderName::Comments - | ArchivedHeaderName::Keywords - | ArchivedHeaderName::ListId => { - // Index headers - header.value.visit_text(|text| { - self.index_tokenized( - Field::Header(mail_parser::HeaderName::from(&header.name)), - text.to_string(), - ); - }); - } - _ => (), - } - } - } - - let part_id = part_id as u16; - match &part.body { - ArchivedMetadataPartType::Text | ArchivedMetadataPartType::Html => { - let text = match (part.decode_contents(raw_message), &part.body) { - (DecodedPartContent::Text(text), ArchivedMetadataPartType::Text) => text, - (DecodedPartContent::Text(html), ArchivedMetadataPartType::Html) => { - html_to_text(html.as_ref()).into() - } - _ => unreachable!(), - }; - - if message_contents.is_html_part(part_id) - || message_contents.is_text_part(part_id) - { - self.index(Field::Body, text, part_language); - } else { - self.index(Field::Attachment, text, part_language); - } - } - ArchivedMetadataPartType::Message(nested_message_id) => { - let nested_message = message.message_id(*nested_message_id); - let nested_message_language = nested_message - .root_part() - .language() - .unwrap_or(Language::Unknown); - if let Some(ArchivedHeaderValue::Text(subject)) = nested_message - .root_part() - .headers - .header_value(&ArchivedHeaderName::Subject) - { - self.index(Field::Attachment, subject.as_ref(), nested_message_language); - } - - for sub_part in nested_message.parts.iter().take(MAX_MESSAGE_PARTS) { - let language = sub_part.language().unwrap_or(nested_message_language); - match &sub_part.body { - ArchivedMetadataPartType::Text | ArchivedMetadataPartType::Html => { - let text = - match (sub_part.decode_contents(raw_message), &sub_part.body) { - ( - DecodedPartContent::Text(text), - ArchivedMetadataPartType::Text, - ) => text, - ( - DecodedPartContent::Text(html), - ArchivedMetadataPartType::Html, - ) => html_to_text(html.as_ref()).into(), - _ => unreachable!(), - }; - self.index(Field::Attachment, text, language); - } - _ => (), - } - } - } - _ => {} - } - } - self - } -} -*/ - -impl SortedAddressBuilder { - pub fn new() -> Self { - Self { - last_is_space: true, - buf: String::with_capacity(32), - } - } - - pub fn push(&mut self, text: &str) -> bool { - if !text.is_empty() { - if !self.buf.is_empty() { - self.buf.push(' '); - self.last_is_space = true; - } - for ch in text.chars() { - for ch in ch.to_lowercase() { - if self.buf.len() < MAX_SORT_FIELD_LENGTH { - let is_space = ch.is_whitespace(); - if !is_space || !self.last_is_space { - self.buf.push(ch); - self.last_is_space = is_space; - } - } else { - return false; - } - } - } - } - true - } - - pub fn build(self) -> String { - if !self.buf.is_empty() { - self.buf - } else { - "!".to_string() - } - } -} - -impl Default for SortedAddressBuilder { - fn default() -> Self { - Self::new() - } -} - -impl ArchivedMessageMetadataPart { - fn language(&self) -> Option { - self.headers - .header_value(&ArchivedHeaderName::ContentLanguage) - .and_then(|v| { - Language::from_iso_639(match v { - ArchivedHeaderValue::Text(v) => v.as_ref(), - ArchivedHeaderValue::TextList(v) => v.first()?, - _ => { - return None; - } - }) - .unwrap_or(Language::Unknown) - .into() - }) - } -} - -#[derive(Debug, PartialEq, Eq)] -pub enum AddressElement { - Name, - Address, - GroupName, -} - -pub trait VisitText { - fn visit_addresses(&self, visitor: impl FnMut(AddressElement, &str)); - fn visit_text<'x>(&'x self, visitor: impl FnMut(&'x str)); - fn into_visit_text(self, visitor: impl FnMut(String)); -} - -impl VisitText for HeaderValue<'_> { - fn visit_addresses(&self, mut visitor: impl FnMut(AddressElement, &str)) { - match self { - HeaderValue::Address(Address::List(addr_list)) => { - for addr in addr_list { - if let Some(name) = &addr.name { - visitor(AddressElement::Name, name); - } - if let Some(addr) = &addr.address { - visitor(AddressElement::Address, addr); - } - } - } - HeaderValue::Address(Address::Group(groups)) => { - for group in groups { - if let Some(name) = &group.name { - visitor(AddressElement::GroupName, name); - } - - for addr in &group.addresses { - if let Some(name) = &addr.name { - visitor(AddressElement::Name, name); - } - if let Some(addr) = &addr.address { - visitor(AddressElement::Address, addr); - } - } - } - } - _ => (), - } - } - - fn visit_text<'x>(&'x self, mut visitor: impl FnMut(&'x str)) { - match &self { - HeaderValue::Text(text) => { - visitor(text.as_ref()); - } - HeaderValue::TextList(texts) => { - for text in texts { - visitor(text.as_ref()); - } - } - _ => (), - } - } - - fn into_visit_text(self, mut visitor: impl FnMut(String)) { - match self { - HeaderValue::Text(text) => { - visitor(text.into_owned()); - } - HeaderValue::TextList(texts) => { - for text in texts { - visitor(text.into_owned()); - } - } - _ => (), - } - } -} - -pub trait VisitTextArchived { - fn visit_addresses(&self, visitor: impl FnMut(AddressElement, &str)); - fn visit_text(&self, visitor: impl FnMut(&str)); -} - -impl VisitTextArchived for ArchivedHeaderValue<'static> { - fn visit_addresses(&self, mut visitor: impl FnMut(AddressElement, &str)) { - match self { - ArchivedHeaderValue::Address(ArchivedAddress::List(addr_list)) => { - for addr in addr_list.iter() { - if let ArchivedOption::Some(name) = &addr.name { - visitor(AddressElement::Name, name); - } - if let ArchivedOption::Some(addr) = &addr.address { - visitor(AddressElement::Address, addr); - } - } - } - ArchivedHeaderValue::Address(ArchivedAddress::Group(groups)) => { - for group in groups.iter() { - if let ArchivedOption::Some(name) = &group.name { - visitor(AddressElement::GroupName, name); - } - - for addr in group.addresses.iter() { - if let ArchivedOption::Some(name) = &addr.name { - visitor(AddressElement::Name, name); - } - if let ArchivedOption::Some(addr) = &addr.address { - visitor(AddressElement::Address, addr); - } - } - } - } - _ => (), - } - } - - fn visit_text(&self, mut visitor: impl FnMut(&str)) { - match &self { - ArchivedHeaderValue::Text(text) => { - visitor(text.as_ref()); - } - ArchivedHeaderValue::TextList(texts) => { - for text in texts.iter() { - visitor(text.as_ref()); - } - } - _ => (), - } - } -} - -pub trait TrimTextValue { - fn trim_text(self, length: usize) -> Self; -} - -impl TrimTextValue for HeaderValue<'_> { - fn trim_text(self, length: usize) -> Self { - match self { - HeaderValue::Address(Address::List(v)) => { - HeaderValue::Address(Address::List(v.trim_text(length))) - } - HeaderValue::Address(Address::Group(v)) => { - HeaderValue::Address(Address::Group(v.trim_text(length))) - } - HeaderValue::Text(v) => HeaderValue::Text(v.trim_text(length)), - HeaderValue::TextList(v) => HeaderValue::TextList(v.trim_text(length)), - v => v, - } - } -} - -impl TrimTextValue for Addr<'_> { - fn trim_text(self, length: usize) -> Self { - Self { - name: self.name.map(|v| v.trim_text(length)), - address: self.address.map(|v| v.trim_text(length)), - } - } -} - -impl TrimTextValue for Group<'_> { - fn trim_text(self, length: usize) -> Self { - Self { - name: self.name.map(|v| v.trim_text(length)), - addresses: self.addresses.trim_text(length), - } - } -} - -impl TrimTextValue for &str { - fn trim_text(self, length: usize) -> Self { - if self.len() < length { - self - } else { - let mut index = 0; - - for (i, _) in self.char_indices() { - if i > length { - break; - } - index = i; - } - - &self[..index] - } - } -} - -impl TrimTextValue for Cow<'_, str> { - fn trim_text(self, length: usize) -> Self { - if self.len() < length { - self - } else { - let mut result = String::with_capacity(length); - for (i, c) in self.char_indices() { - if i > length { - break; - } - result.push(c); - } - result.into() - } - } -} - -impl TrimTextValue for Vec { - fn trim_text(self, length: usize) -> Self { - self.into_iter().map(|v| v.trim_text(length)).collect() - } -} diff --git a/crates/email/src/message/index/extractors.rs b/crates/email/src/message/index/extractors.rs new file mode 100644 index 00000000..b621cdd9 --- /dev/null +++ b/crates/email/src/message/index/extractors.rs @@ -0,0 +1,250 @@ +/* + * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC + * + * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + */ + +use crate::message::metadata::{ArchivedMessageMetadataContents, ArchivedMessageMetadataPart}; +use mail_parser::{ + Addr, Address, ArchivedAddress, ArchivedHeaderName, ArchivedHeaderValue, Group, HeaderValue, + core::rkyv::ArchivedGetHeader, +}; +use nlp::language::Language; +use rkyv::option::ArchivedOption; +use std::borrow::Cow; + +impl ArchivedMessageMetadataContents { + pub fn is_html_part(&self, part_id: u16) -> bool { + self.html_body.iter().any(|&id| id == part_id) + } + + pub fn is_text_part(&self, part_id: u16) -> bool { + self.text_body.iter().any(|&id| id == part_id) + } +} + +impl ArchivedMessageMetadataPart { + pub fn language(&self) -> Option { + self.headers + .header_value(&ArchivedHeaderName::ContentLanguage) + .and_then(|v| { + Language::from_iso_639(match v { + ArchivedHeaderValue::Text(v) => v.as_ref(), + ArchivedHeaderValue::TextList(v) => v.first()?, + _ => { + return None; + } + }) + .unwrap_or(Language::Unknown) + .into() + }) + } +} + +#[derive(Debug, PartialEq, Eq)] +pub enum AddressElement { + Name, + Address, + GroupName, +} + +pub trait VisitText { + fn visit_addresses(&self, visitor: impl FnMut(AddressElement, &str)); + fn visit_text<'x>(&'x self, visitor: impl FnMut(&'x str)); + fn into_visit_text(self, visitor: impl FnMut(String)); +} + +impl VisitText for HeaderValue<'_> { + fn visit_addresses(&self, mut visitor: impl FnMut(AddressElement, &str)) { + match self { + HeaderValue::Address(Address::List(addr_list)) => { + for addr in addr_list { + if let Some(name) = &addr.name { + visitor(AddressElement::Name, name); + } + if let Some(addr) = &addr.address { + visitor(AddressElement::Address, addr); + } + } + } + HeaderValue::Address(Address::Group(groups)) => { + for group in groups { + if let Some(name) = &group.name { + visitor(AddressElement::GroupName, name); + } + + for addr in &group.addresses { + if let Some(name) = &addr.name { + visitor(AddressElement::Name, name); + } + if let Some(addr) = &addr.address { + visitor(AddressElement::Address, addr); + } + } + } + } + _ => (), + } + } + + fn visit_text<'x>(&'x self, mut visitor: impl FnMut(&'x str)) { + match &self { + HeaderValue::Text(text) => { + visitor(text.as_ref()); + } + HeaderValue::TextList(texts) => { + for text in texts { + visitor(text.as_ref()); + } + } + _ => (), + } + } + + fn into_visit_text(self, mut visitor: impl FnMut(String)) { + match self { + HeaderValue::Text(text) => { + visitor(text.into_owned()); + } + HeaderValue::TextList(texts) => { + for text in texts { + visitor(text.into_owned()); + } + } + _ => (), + } + } +} + +pub trait VisitTextArchived { + fn visit_addresses(&self, visitor: impl FnMut(AddressElement, &str)); + fn visit_text(&self, visitor: impl FnMut(&str)); +} + +impl VisitTextArchived for ArchivedHeaderValue<'static> { + fn visit_addresses(&self, mut visitor: impl FnMut(AddressElement, &str)) { + match self { + ArchivedHeaderValue::Address(ArchivedAddress::List(addr_list)) => { + for addr in addr_list.iter() { + if let ArchivedOption::Some(name) = &addr.name { + visitor(AddressElement::Name, name); + } + if let ArchivedOption::Some(addr) = &addr.address { + visitor(AddressElement::Address, addr); + } + } + } + ArchivedHeaderValue::Address(ArchivedAddress::Group(groups)) => { + for group in groups.iter() { + if let ArchivedOption::Some(name) = &group.name { + visitor(AddressElement::GroupName, name); + } + + for addr in group.addresses.iter() { + if let ArchivedOption::Some(name) = &addr.name { + visitor(AddressElement::Name, name); + } + if let ArchivedOption::Some(addr) = &addr.address { + visitor(AddressElement::Address, addr); + } + } + } + } + _ => (), + } + } + + fn visit_text(&self, mut visitor: impl FnMut(&str)) { + match &self { + ArchivedHeaderValue::Text(text) => { + visitor(text.as_ref()); + } + ArchivedHeaderValue::TextList(texts) => { + for text in texts.iter() { + visitor(text.as_ref()); + } + } + _ => (), + } + } +} + +pub trait TrimTextValue { + fn trim_text(self, length: usize) -> Self; +} + +impl TrimTextValue for HeaderValue<'_> { + fn trim_text(self, length: usize) -> Self { + match self { + HeaderValue::Address(Address::List(v)) => { + HeaderValue::Address(Address::List(v.trim_text(length))) + } + HeaderValue::Address(Address::Group(v)) => { + HeaderValue::Address(Address::Group(v.trim_text(length))) + } + HeaderValue::Text(v) => HeaderValue::Text(v.trim_text(length)), + HeaderValue::TextList(v) => HeaderValue::TextList(v.trim_text(length)), + v => v, + } + } +} + +impl TrimTextValue for Addr<'_> { + fn trim_text(self, length: usize) -> Self { + Self { + name: self.name.map(|v| v.trim_text(length)), + address: self.address.map(|v| v.trim_text(length)), + } + } +} + +impl TrimTextValue for Group<'_> { + fn trim_text(self, length: usize) -> Self { + Self { + name: self.name.map(|v| v.trim_text(length)), + addresses: self.addresses.trim_text(length), + } + } +} + +impl TrimTextValue for &str { + fn trim_text(self, length: usize) -> Self { + if self.len() < length { + self + } else { + let mut index = 0; + + for (i, _) in self.char_indices() { + if i > length { + break; + } + index = i; + } + + &self[..index] + } + } +} + +impl TrimTextValue for Cow<'_, str> { + fn trim_text(self, length: usize) -> Self { + if self.len() < length { + self + } else { + let mut result = String::with_capacity(length); + for (i, c) in self.char_indices() { + if i > length { + break; + } + result.push(c); + } + result.into() + } + } +} + +impl TrimTextValue for Vec { + fn trim_text(self, length: usize) -> Self { + self.into_iter().map(|v| v.trim_text(length)).collect() + } +} diff --git a/crates/email/src/message/index/metadata.rs b/crates/email/src/message/index/metadata.rs new file mode 100644 index 00000000..d5848910 --- /dev/null +++ b/crates/email/src/message/index/metadata.rs @@ -0,0 +1,232 @@ +/* + * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC + * + * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + */ + +use crate::message::{ + index::{IndexMessage, MAX_MESSAGE_PARTS, PREVIEW_LENGTH}, + metadata::{ + ArchivedMessageMetadata, ArchivedMessageMetadataPart, MessageData, MessageMetadata, + MessageMetadataPart, + }, +}; +use common::storage::index::ObjectIndexBuilder; +use mail_parser::{decoders::html::html_to_text, parsers::preview::preview_text}; +use store::{ + Serialize, SerializeInfallible, + write::{Archiver, BatchBuilder, BlobOp, DirectoryClass, IndexPropertyClass, ValueClass}, +}; +use trc::AddContext; +use types::{blob_hash::BlobHash, field::EmailField}; + +impl MessageMetadata { + #[inline(always)] + pub fn root_part(&self) -> &MessageMetadataPart { + &self.contents[0].parts[0] + } + + pub fn index( + self, + batch: &mut BatchBuilder, + account_id: u32, + tenant_id: Option, + set: bool, + ) -> trc::Result<()> { + if set { + // Serialize metadata + batch.set( + ValueClass::IndexProperty(IndexPropertyClass::Integer { + property: EmailField::ReceivedToSize.into(), + value: self.received_at, + }), + self.size.serialize(), + ); + } else { + // Delete metadata + batch + .clear(EmailField::Metadata) + .clear(ValueClass::IndexProperty(IndexPropertyClass::Integer { + property: EmailField::ReceivedToSize.into(), + value: self.received_at, + })); + } + + // Index properties + let quota = if set { + self.size as i64 + } else { + -(self.size as i64) + }; + batch.add(DirectoryClass::UsedQuota(account_id), quota); + if let Some(tenant_id) = tenant_id { + batch.add(DirectoryClass::UsedQuota(tenant_id), quota); + } + + // Link blob + if set { + batch.set( + BlobOp::Link { + hash: self.blob_hash.clone(), + }, + Vec::new(), + ); + } else { + batch.clear(BlobOp::Link { + hash: self.blob_hash.clone(), + }); + } + + if set { + batch.set(EmailField::Metadata, Archiver::new(self).serialize()?); + } + + Ok(()) + } +} + +impl ArchivedMessageMetadata { + #[inline(always)] + pub fn root_part(&self) -> &ArchivedMessageMetadataPart { + &self.contents[0].parts[0] + } + + pub fn unindex(&self, batch: &mut BatchBuilder, account_id: u32, tenant_id: Option) { + // Delete metadata + batch + .clear(EmailField::Metadata) + .clear(ValueClass::IndexProperty(IndexPropertyClass::Integer { + property: EmailField::ReceivedToSize.into(), + value: self.received_at.to_native(), + })); + + // Index properties + let quota = -(u32::from(self.size) as i64); + batch.add(DirectoryClass::UsedQuota(account_id), quota); + if let Some(tenant_id) = tenant_id { + batch.add(DirectoryClass::UsedQuota(tenant_id), quota); + } + + // Unlink blob + batch.clear(BlobOp::Link { + hash: BlobHash::from(&self.blob_hash), + }); + } +} + +impl IndexMessage for BatchBuilder { + fn index_message( + &mut self, + account_id: u32, + tenant_id: Option, + message: mail_parser::Message<'_>, + blob_hash: BlobHash, + data: MessageData, + received_at: u64, + ) -> trc::Result<&mut Self> { + // Index size + self.set( + ValueClass::IndexProperty(IndexPropertyClass::Integer { + property: EmailField::ReceivedToSize.into(), + value: received_at, + }), + (message.raw_message.len() as u32).serialize(), + ) + .add( + DirectoryClass::UsedQuota(account_id), + message.raw_message.len() as i64, + ); + if let Some(tenant_id) = tenant_id { + self.add( + DirectoryClass::UsedQuota(tenant_id), + message.raw_message.len() as i64, + ); + } + + let mut has_attachments = false; + let mut preview = None; + let preview_part_id = message + .text_body + .first() + .or_else(|| message.html_body.first()) + .copied() + .unwrap_or(u32::MAX); + + for (part_id, part) in message.parts.iter().take(MAX_MESSAGE_PARTS).enumerate() { + let part_id = part_id as u32; + match &part.body { + mail_parser::PartType::Text(text) => { + if part_id == preview_part_id { + preview = + preview_text(text.replace('\r', "").into(), PREVIEW_LENGTH).into(); + } + + if !message.text_body.contains(&part_id) + && !message.html_body.contains(&part_id) + { + has_attachments = true; + } + } + mail_parser::PartType::Html(html) => { + let text = html_to_text(html); + if part_id == preview_part_id { + preview = + preview_text(text.replace('\r', "").into(), PREVIEW_LENGTH).into(); + } + + if !message.text_body.contains(&part_id) + && !message.html_body.contains(&part_id) + { + has_attachments = true; + } + } + mail_parser::PartType::Binary(_) | mail_parser::PartType::Message(_) + if !has_attachments => + { + has_attachments = true; + } + _ => {} + } + } + + // Build metadata + let root_part = message.root_part(); + let metadata = MessageMetadata { + preview: preview.unwrap_or_default().into_owned(), + size: message.raw_message.len() as u32, + raw_headers: message + .raw_message + .as_ref() + .get(root_part.offset_header as usize..root_part.offset_body as usize) + .unwrap_or_default() + .to_vec(), + contents: vec![], + received_at, + has_attachments, + blob_hash, + } + .with_contents(message); + + // Link blob + self.set( + BlobOp::Link { + hash: metadata.blob_hash.clone(), + }, + Vec::new(), + ); + + // Store message data + self.custom(ObjectIndexBuilder::<(), _>::new().with_changes(data)) + .caused_by(trc::location!())?; + + // Store message metadata + self.set( + EmailField::Metadata, + Archiver::new(metadata) + .serialize() + .caused_by(trc::location!())?, + ); + + Ok(self) + } +} diff --git a/crates/email/src/message/index/mod.rs b/crates/email/src/message/index/mod.rs new file mode 100644 index 00000000..8d275a7e --- /dev/null +++ b/crates/email/src/message/index/mod.rs @@ -0,0 +1,73 @@ +/* + * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC + * + * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + */ + +use crate::message::metadata::{ArchivedMessageData, MessageData}; +use common::storage::index::{IndexValue, IndexableObject}; +use types::{blob_hash::BlobHash, collection::SyncCollection}; + +pub mod extractors; +pub mod metadata; +pub mod search; + +pub(super) const MAX_MESSAGE_PARTS: usize = 1000; +pub const PREVIEW_LENGTH: usize = 256; + +impl IndexableObject for MessageData { + fn index_values(&self) -> impl Iterator> { + [ + IndexValue::LogItem { + sync_collection: SyncCollection::Email, + prefix: self.thread_id.into(), + }, + IndexValue::LogContainerProperty { + sync_collection: SyncCollection::Thread, + ids: vec![self.thread_id], + }, + IndexValue::LogContainerProperty { + sync_collection: SyncCollection::Email, + ids: self.mailboxes.iter().map(|m| m.mailbox_id).collect(), + }, + ] + .into_iter() + } +} + +impl IndexableObject for &ArchivedMessageData { + fn index_values(&self) -> impl Iterator> { + [ + IndexValue::LogItem { + sync_collection: SyncCollection::Email, + prefix: self.thread_id.to_native().into(), + }, + IndexValue::LogContainerProperty { + sync_collection: SyncCollection::Thread, + ids: vec![self.thread_id.to_native()], + }, + IndexValue::LogContainerProperty { + sync_collection: SyncCollection::Email, + ids: self + .mailboxes + .iter() + .map(|m| m.mailbox_id.to_native()) + .collect(), + }, + ] + .into_iter() + } +} + +pub(super) trait IndexMessage { + #[allow(clippy::too_many_arguments)] + fn index_message( + &mut self, + account_id: u32, + tenant_id: Option, + message: mail_parser::Message<'_>, + blob_hash: BlobHash, + data: MessageData, + received_at: u64, + ) -> trc::Result<&mut Self>; +} diff --git a/crates/email/src/message/index/search.rs b/crates/email/src/message/index/search.rs new file mode 100644 index 00000000..b7649226 --- /dev/null +++ b/crates/email/src/message/index/search.rs @@ -0,0 +1,194 @@ +/* + * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC + * + * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + */ + +use crate::message::{ + index::{MAX_MESSAGE_PARTS, extractors::VisitTextArchived}, + metadata::{ArchivedMessageMetadata, ArchivedMetadataPartType, DecodedPartContent}, +}; +use mail_parser::{ + ArchivedHeaderName, ArchivedHeaderValue, DateTime, HeaderName, core::rkyv::ArchivedGetHeader, + decoders::html::html_to_text, +}; +use nlp::language::Language; +use std::borrow::Cow; +use store::{ + ahash::AHashSet, + search::{EmailSearchField, IndexDocument, SearchField}, +}; + +impl ArchivedMessageMetadata { + pub fn index_document( + &self, + raw_message: &[u8], + index_headers: &AHashSet>, + ) -> IndexDocument { + let mut language = Language::Unknown; + let message_contents = &self.contents[0]; + let mut document = IndexDocument::with_default_language(language); + + document.index_number( + EmailSearchField::ReceivedAt, + self.received_at.to_native() as i64, + ); + document.index_number(EmailSearchField::Size, self.size.to_native()); + + for (part_id, part) in message_contents + .parts + .iter() + .take(MAX_MESSAGE_PARTS) + .enumerate() + { + let part_language = part.language().unwrap_or(language); + if part_id == 0 { + 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, + ); + }); + } + ArchivedHeaderName::To => { + header.value.visit_addresses(|_, value| { + document.index_text(EmailSearchField::To, value, Language::Unknown); + }); + } + ArchivedHeaderName::Cc => { + header.value.visit_addresses(|_, value| { + document.index_text(EmailSearchField::Cc, value, Language::Unknown); + }); + } + ArchivedHeaderName::Bcc => { + header.value.visit_addresses(|_, value| { + document.index_text( + EmailSearchField::Bcc, + value, + Language::Unknown, + ); + }); + } + ArchivedHeaderName::Subject => { + if let Some(subject) = header.value.as_text() { + document.index_text( + EmailSearchField::Subject, + subject, + part_language, + ); + } + } + ArchivedHeaderName::Date => { + if let Some(date) = header.value.as_datetime() { + document.index_number( + EmailSearchField::SentAt, + DateTime::from(date).to_timestamp(), + ); + } + } + _ => { + header.value.visit_text(|text| { + document.index_text( + EmailSearchField::Header(header_name.clone()), + text, + Language::Unknown, + ); + }); + } + } + } + } + + let part_id = part_id as u16; + match &part.body { + ArchivedMetadataPartType::Text | ArchivedMetadataPartType::Html => { + let text = match (part.decode_contents(raw_message), &part.body) { + (DecodedPartContent::Text(text), ArchivedMetadataPartType::Text) => text, + (DecodedPartContent::Text(html), ArchivedMetadataPartType::Html) => { + html_to_text(html.as_ref()).into() + } + _ => unreachable!(), + }; + + 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 { + document.index_text( + EmailSearchField::Attachment, + text.as_ref(), + part_language, + ); + } + } + ArchivedMetadataPartType::Message(nested_message_id) => { + let nested_message = self.message_id(*nested_message_id); + let nested_message_language = nested_message + .root_part() + .language() + .unwrap_or(Language::Unknown); + if let Some(ArchivedHeaderValue::Text(subject)) = nested_message + .root_part() + .headers + .header_value(&ArchivedHeaderName::Subject) + { + document.index_text( + EmailSearchField::Attachment, + subject.as_ref(), + nested_message_language, + ); + } + + for sub_part in nested_message.parts.iter().take(MAX_MESSAGE_PARTS) { + let language = sub_part.language().unwrap_or(nested_message_language); + match &sub_part.body { + ArchivedMetadataPartType::Text | ArchivedMetadataPartType::Html => { + let text = + match (sub_part.decode_contents(raw_message), &sub_part.body) { + ( + DecodedPartContent::Text(text), + ArchivedMetadataPartType::Text, + ) => text, + ( + DecodedPartContent::Text(html), + ArchivedMetadataPartType::Html, + ) => html_to_text(html.as_ref()).into(), + _ => unreachable!(), + }; + document.index_text( + EmailSearchField::Attachment, + text.as_ref(), + language, + ); + } + _ => (), + } + } + } + _ => {} + } + } + + let has_attachment = document.has_field(&SearchField::Email(EmailSearchField::Attachment)); + + document.index_bool(EmailSearchField::HasAttachment, has_attachment); + + document + } +} diff --git a/crates/email/src/message/ingest.rs b/crates/email/src/message/ingest.rs index 2a18d631..894ff859 100644 --- a/crates/email/src/message/ingest.rs +++ b/crates/email/src/message/ingest.rs @@ -4,16 +4,13 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use super::{ - crypto::{EncryptMessage, EncryptMessageError}, - index::{MAX_SORT_FIELD_LENGTH, TrimTextValue}, -}; +use super::crypto::{EncryptMessage, EncryptMessageError}; use crate::{ cache::{MessageCacheFetch, email::MessageCacheAccess}, mailbox::{INBOX_ID, JUNK_ID, UidMailbox}, message::{ crypto::EncryptionParams, - index::{IndexMessage, VisitText}, + index::{IndexMessage, extractors::VisitText}, metadata::MessageData, }, }; @@ -36,8 +33,8 @@ use store::{ IndexKeyPrefix, IterateParams, U32_LEN, ValueKey, ahash::AHashMap, write::{ - BatchBuilder, IndexPropertyClass, TaskQueueClass, ValueClass, key::DeserializeBigEndian, - now, + BatchBuilder, IndexPropertyClass, SearchIndex, TaskQueueClass, ValueClass, + key::DeserializeBigEndian, now, }, }; use trc::{AddContext, MessageIngestEvent}; @@ -427,8 +424,7 @@ impl EmailIngest for Server { list.first().unwrap().as_ref() } _ => "", - }) - .trim_text(MAX_SORT_FIELD_LENGTH); + }); } _ => (), } @@ -654,7 +650,11 @@ impl EmailIngest for Server { ThreadInfo::serialize(thread_id, &message_ids), ) .set( - ValueClass::TaskQueue(TaskQueueClass::IndexEmail { due }), + ValueClass::TaskQueue(TaskQueueClass::UpdateIndex { + index: SearchIndex::Email, + due, + is_insert: true, + }), MergeThreadTask::new(thread_result).serialize(), ); diff --git a/crates/email/src/submission/index.rs b/crates/email/src/submission/index.rs index 851389c7..1ce87f8f 100644 --- a/crates/email/src/submission/index.rs +++ b/crates/email/src/submission/index.rs @@ -6,30 +6,27 @@ use super::{ArchivedEmailSubmission, EmailSubmission}; use common::storage::index::{IndexValue, IndexableAndSerializableObject, IndexableObject}; +use store::{ + U32_LEN, + write::{IndexPropertyClass, ValueClass, key::KeySerializer}, +}; use types::{collection::SyncCollection, field::EmailSubmissionField}; impl IndexableObject for EmailSubmission { fn index_values(&self) -> impl Iterator> { [ - IndexValue::Index { - field: EmailSubmissionField::UndoStatus.into(), - value: self.undo_status.as_index().into(), - }, - IndexValue::Index { - field: EmailSubmissionField::EmailId.into(), - value: self.email_id.into(), - }, - IndexValue::Index { - field: EmailSubmissionField::ThreadId.into(), - value: self.thread_id.into(), - }, - IndexValue::Index { - field: EmailSubmissionField::IdentityId.into(), - value: self.identity_id.into(), - }, - IndexValue::Index { - field: EmailSubmissionField::SendAt.into(), - value: self.send_at.into(), + IndexValue::Property { + field: ValueClass::IndexProperty(IndexPropertyClass::Integer { + property: EmailSubmissionField::Metadata.into(), + value: self.send_at, + }), + value: KeySerializer::new(U32_LEN * 3 + 1) + .write(self.email_id) + .write(self.thread_id) + .write(self.identity_id) + .write(self.undo_status.as_index()) + .finalize() + .into(), }, IndexValue::LogItem { sync_collection: SyncCollection::EmailSubmission, @@ -43,25 +40,18 @@ impl IndexableObject for EmailSubmission { impl IndexableObject for &ArchivedEmailSubmission { fn index_values(&self) -> impl Iterator> { [ - IndexValue::Index { - field: EmailSubmissionField::UndoStatus.into(), - value: self.undo_status.as_index().into(), - }, - IndexValue::Index { - field: EmailSubmissionField::EmailId.into(), - value: self.email_id.into(), - }, - IndexValue::Index { - field: EmailSubmissionField::ThreadId.into(), - value: self.thread_id.into(), - }, - IndexValue::Index { - field: EmailSubmissionField::IdentityId.into(), - value: self.identity_id.into(), - }, - IndexValue::Index { - field: EmailSubmissionField::SendAt.into(), - value: self.send_at.into(), + IndexValue::Property { + field: ValueClass::IndexProperty(IndexPropertyClass::Integer { + property: EmailSubmissionField::Metadata.into(), + value: self.send_at.to_native(), + }), + value: KeySerializer::new(U32_LEN * 3 + 1) + .write(self.email_id.to_native()) + .write(self.thread_id.to_native()) + .write(self.identity_id.to_native()) + .write(self.undo_status.as_index()) + .finalize() + .into(), }, IndexValue::LogItem { sync_collection: SyncCollection::EmailSubmission, diff --git a/crates/email/src/submission/mod.rs b/crates/email/src/submission/mod.rs index 5c1d79d0..ed59eaed 100644 --- a/crates/email/src/submission/mod.rs +++ b/crates/email/src/submission/mod.rs @@ -86,11 +86,11 @@ impl UndoStatus { } } - pub fn as_index(&self) -> &'static str { + pub fn as_index(&self) -> u8 { match self { - UndoStatus::Pending => "p", - UndoStatus::Final => "f", - UndoStatus::Canceled => "c", + UndoStatus::Pending => b'p', + UndoStatus::Final => b'f', + UndoStatus::Canceled => b'c', } } } @@ -104,11 +104,11 @@ impl ArchivedUndoStatus { } } - pub fn as_index(&self) -> &'static str { + pub fn as_index(&self) -> u8 { match self { - ArchivedUndoStatus::Pending => "p", - ArchivedUndoStatus::Final => "f", - ArchivedUndoStatus::Canceled => "c", + ArchivedUndoStatus::Pending => b'p', + ArchivedUndoStatus::Final => b'f', + ArchivedUndoStatus::Canceled => b'c', } } } diff --git a/crates/groupware/src/cache/calcard.rs b/crates/groupware/src/cache/calcard.rs index cc2c596a..222dd8d8 100644 --- a/crates/groupware/src/cache/calcard.rs +++ b/crates/groupware/src/cache/calcard.rs @@ -9,7 +9,7 @@ use crate::{ DavResourceName, RFC_3986, calendar::{ ArchivedCalendar, ArchivedCalendarEvent, Calendar, CalendarEvent, SCHEDULE_INBOX_ID, - SCHEDULE_OUTBOX_ID, + SCHEDULE_OUTBOX_ID, storage::ItipAutoExpunge, }, contact::{AddressBook, ArchivedAddressBook, ArchivedContactCard, ContactCard}, }; @@ -192,11 +192,7 @@ pub(super) async fn build_scheduling_resources( .unwrap_or_else(|| format!("_{account_id}")); let item_ids = server - .document_ids( - account_id, - Collection::CalendarEventNotification, - CalendarNotificationField::CreatedToId, - ) + .itip_ids(account_id) .await .caused_by(trc::location!())?; diff --git a/crates/groupware/src/calendar/index.rs b/crates/groupware/src/calendar/index.rs index 16621008..dfe1df3e 100644 --- a/crates/groupware/src/calendar/index.rs +++ b/crates/groupware/src/calendar/index.rs @@ -17,8 +17,10 @@ use calcard::icalendar::{ ICalendarParameterValue, ICalendarProperty, ICalendarValue, }; use common::storage::index::{IndexValue, IndexableAndSerializableObject, IndexableObject}; +use nlp::language::Language; use store::{ - write::{IndexPropertyClass, ValueClass}, + search::{CalendarSearchField, IndexDocument}, + write::{IndexPropertyClass, SearchIndex, ValueClass}, xxhash_rust::xxh3, }; use types::{acl::AclGrant, collection::SyncCollection, field::CalendarNotificationField}; @@ -76,7 +78,11 @@ impl IndexableObject for CalendarEvent { fn index_values(&self) -> impl Iterator> { [ IndexValue::SearchIndex { - hashes: self.hashes().collect(), + index: SearchIndex::Calendar, + hash: self + .hashes() + .chain([self.data.event_range_start() as u64]) + .fold(0, |acc, hash| acc ^ hash), }, IndexValue::Quota { used: self.dead_properties.size() as u32 @@ -98,7 +104,11 @@ impl IndexableObject for &ArchivedCalendarEvent { fn index_values(&self) -> impl Iterator> { [ IndexValue::SearchIndex { - hashes: self.hashes().collect(), + index: SearchIndex::Calendar, + hash: self + .hashes() + .chain([self.data.event_range_start() as u64]) + .fold(0, |acc, hash| acc ^ hash), }, IndexValue::Quota { used: self.dead_properties.size() as u32 @@ -263,6 +273,7 @@ impl CalendarEvent { | ICalendarProperty::Comment | ICalendarProperty::Attendee | ICalendarProperty::Organizer + | ICalendarProperty::Uid ) }) }) @@ -302,6 +313,7 @@ impl ArchivedCalendarEvent { | ArchivedICalendarProperty::Comment | ArchivedICalendarProperty::Attendee | ArchivedICalendarProperty::Organizer + | ArchivedICalendarProperty::Uid ) }) }) @@ -322,3 +334,54 @@ impl ArchivedCalendarEvent { .map(|v| xxh3::xxh3_64(v.as_bytes())) } } + +impl ArchivedCalendarEvent { + pub fn index_document(&self) -> IndexDocument { + let mut document = IndexDocument::with_default_language(Language::Unknown); + + document.index_number(CalendarSearchField::Start, self.data.event_range_start()); + + for component in self + .data + .event + .components + .iter() + .filter(|e| e.component_type.is_scheduling_object()) + { + for entry in component.entries.iter() { + let field = match entry.name { + ArchivedICalendarProperty::Summary => CalendarSearchField::Title, + ArchivedICalendarProperty::Description => CalendarSearchField::Description, + ArchivedICalendarProperty::Location => CalendarSearchField::Location, + ArchivedICalendarProperty::Organizer => CalendarSearchField::Owner, + 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, + ); + } + } + } + + document + } +} diff --git a/crates/groupware/src/calendar/storage.rs b/crates/groupware/src/calendar/storage.rs index 0a45f006..1cf79ef5 100644 --- a/crates/groupware/src/calendar/storage.rs +++ b/crates/groupware/src/calendar/storage.rs @@ -18,10 +18,10 @@ use crate::{ use calcard::common::timezone::Tz; use common::{Server, auth::AccessToken, storage::index::ObjectIndexBuilder}; use store::{ - IndexKey, IterateParams, SerializeInfallible, U16_LEN, U32_LEN, U64_LEN, + IterateParams, U16_LEN, U32_LEN, U64_LEN, ValueKey, roaring::RoaringBitmap, write::{ - Archive, BatchBuilder, TaskQueueClass, ValueClass, + Archive, BatchBuilder, IndexPropertyClass, TaskQueueClass, ValueClass, key::{DeserializeBigEndian, KeySerializer}, now, }, @@ -33,6 +33,8 @@ use types::{ }; pub trait ItipAutoExpunge: Sync + Send { + fn itip_ids(&self, account_id: u32) -> impl Future> + Send; + fn itip_auto_expunge( &self, account_id: u32, @@ -41,33 +43,71 @@ pub trait ItipAutoExpunge: Sync + Send { } impl ItipAutoExpunge for Server { - async fn itip_auto_expunge(&self, account_id: u32, hold_period: u64) -> trc::Result<()> { - let mut destroy_ids = RoaringBitmap::new(); + async fn itip_ids(&self, account_id: u32) -> trc::Result { + let mut document_ids = RoaringBitmap::new(); self.store() .iterate( IterateParams::new( - IndexKey { + ValueKey { account_id, collection: Collection::CalendarEventNotification.into(), document_id: 0, - field: CalendarNotificationField::CreatedToId.into(), - key: 0u64.serialize(), + class: ValueClass::IndexProperty(IndexPropertyClass::Integer { + property: CalendarNotificationField::CreatedToId.into(), + value: 0, + }), }, - IndexKey { + ValueKey { account_id, collection: Collection::CalendarEventNotification.into(), - document_id: u32::MAX, - field: CalendarNotificationField::CreatedToId.into(), - key: now().saturating_sub(hold_period).serialize(), + document_id: 0, + class: ValueClass::IndexProperty(IndexPropertyClass::Integer { + property: CalendarNotificationField::CreatedToId.into(), + value: u64::MAX, + }), }, ) .no_values() .ascending(), |key, _| { - destroy_ids.insert( - key.deserialize_be_u32(key.len() - U32_LEN) - .caused_by(trc::location!())?, - ); + document_ids.insert(key.deserialize_be_u32(key.len() - U32_LEN)?); + + Ok(true) + }, + ) + .await + .caused_by(trc::location!()) + .map(|_| document_ids) + } + + async fn itip_auto_expunge(&self, account_id: u32, hold_period: u64) -> trc::Result<()> { + let mut destroy_ids = RoaringBitmap::new(); + self.store() + .iterate( + IterateParams::new( + ValueKey { + account_id, + collection: Collection::CalendarEventNotification.into(), + document_id: 0, + class: ValueClass::IndexProperty(IndexPropertyClass::Integer { + property: CalendarNotificationField::CreatedToId.into(), + value: 0, + }), + }, + ValueKey { + account_id, + collection: Collection::CalendarEventNotification.into(), + document_id: 0, + class: ValueClass::IndexProperty(IndexPropertyClass::Integer { + property: CalendarNotificationField::CreatedToId.into(), + value: now().saturating_sub(hold_period), + }), + }, + ) + .no_values() + .ascending(), + |key, _| { + destroy_ids.insert(key.deserialize_be_u32(key.len() - U32_LEN)?); Ok(true) }, diff --git a/crates/groupware/src/contact/index.rs b/crates/groupware/src/contact/index.rs index 19942f5b..9dc17cd0 100644 --- a/crates/groupware/src/contact/index.rs +++ b/crates/groupware/src/contact/index.rs @@ -5,9 +5,20 @@ */ use super::{AddressBook, ArchivedAddressBook, ArchivedContactCard, ContactCard}; -use calcard::vcard::{ArchivedVCardProperty, VCardProperty}; +use calcard::{ + common::IanaString, + vcard::{ + ArchivedVCardParameterValue, ArchivedVCardProperty, ArchivedVCardValue, + VCardParameterValue, VCardProperty, + }, +}; use common::storage::index::{IndexValue, IndexableAndSerializableObject, IndexableObject}; -use store::xxhash_rust::xxh3; +use nlp::language::Language; +use store::{ + search::{ContactSearchField, IndexDocument}, + write::SearchIndex, + xxhash_rust::xxh3, +}; use types::{acl::AclGrant, collection::SyncCollection, field::ContactField}; use utils::sanitize_email; @@ -86,7 +97,8 @@ impl IndexableObject for ContactCard { value: self.emails().next().into(), }, IndexValue::SearchIndex { - hashes: self.hashes().collect(), + index: SearchIndex::Contacts, + hash: self.hashes().fold(0, |acc, hash| acc ^ hash), }, IndexValue::Quota { used: self.dead_properties.size() as u32 @@ -115,7 +127,8 @@ impl IndexableObject for &ArchivedContactCard { value: self.emails().next().into(), }, IndexValue::SearchIndex { - hashes: self.hashes().collect(), + index: SearchIndex::Contacts, + hash: self.hashes().fold(0, |acc, hash| acc ^ hash), }, IndexValue::Quota { used: self.dead_properties.size() as u32 @@ -154,9 +167,23 @@ impl ContactCard { | VCardProperty::Note | VCardProperty::Nickname | VCardProperty::Email + | VCardProperty::Kind + | VCardProperty::Uid + | VCardProperty::Member + | VCardProperty::Impp + | VCardProperty::Socialprofile + | VCardProperty::Tel ) }) - .flat_map(|e| e.values.iter().filter_map(|v| v.as_text())) + .flat_map(|e| { + e.values + .iter() + .filter_map(|v| v.as_text()) + .chain(e.params.iter().filter_map(|p| match &p.value { + VCardParameterValue::Text(v) => Some(v.as_str()), + _ => None, + })) + }) .map(|v| xxh3::xxh3_64(v.as_bytes())) } @@ -185,9 +212,23 @@ impl ArchivedContactCard { | ArchivedVCardProperty::Note | ArchivedVCardProperty::Nickname | ArchivedVCardProperty::Email + | ArchivedVCardProperty::Kind + | ArchivedVCardProperty::Uid + | ArchivedVCardProperty::Member + | ArchivedVCardProperty::Impp + | ArchivedVCardProperty::Socialprofile + | ArchivedVCardProperty::Tel ) }) - .flat_map(|e| e.values.iter().filter_map(|v| v.as_text())) + .flat_map(|e| { + e.values + .iter() + .filter_map(|v| v.as_text()) + .chain(e.params.iter().filter_map(|p| match &p.value { + ArchivedVCardParameterValue::Text(v) => Some(v.as_str()), + _ => None, + })) + }) .map(|v| xxh3::xxh3_64(v.as_bytes())) } @@ -199,3 +240,55 @@ impl ArchivedContactCard { }) } } + +impl ArchivedContactCard { + pub fn index_document(&self) -> IndexDocument { + let mut document = IndexDocument::with_default_language(Language::Unknown); + + document.index_number(ContactSearchField::Created, self.created.to_native()); + + for entry in self.card.entries.iter() { + let field = match entry.name { + ArchivedVCardProperty::N => ContactSearchField::Name, + ArchivedVCardProperty::Nickname => ContactSearchField::Nickname, + ArchivedVCardProperty::Org => ContactSearchField::Organization, + ArchivedVCardProperty::Email => ContactSearchField::Email, + ArchivedVCardProperty::Tel => ContactSearchField::Phone, + ArchivedVCardProperty::Impp | ArchivedVCardProperty::Socialprofile => { + ContactSearchField::OnlineService + } + ArchivedVCardProperty::Adr => ContactSearchField::Address, + ArchivedVCardProperty::Note => ContactSearchField::Note, + ArchivedVCardProperty::Kind => ContactSearchField::Kind, + 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); + } + } + _ => (), + } + } + + for param in entry.params.iter() { + if let ArchivedVCardParameterValue::Text(value) = ¶m.value { + document.index_text(field, value, Language::Unknown); + } + } + } + + document + } +} diff --git a/crates/imap/src/op/expunge.rs b/crates/imap/src/op/expunge.rs index 20f62913..1978e082 100644 --- a/crates/imap/src/op/expunge.rs +++ b/crates/imap/src/op/expunge.rs @@ -22,7 +22,7 @@ use std::{sync::Arc, time::Instant}; use store::{ SerializeInfallible, roaring::RoaringBitmap, - write::{BatchBuilder, TaskQueueClass, ValueClass, now}, + write::{BatchBuilder, SearchIndex, TaskQueueClass, ValueClass, now}, }; use trc::AddContext; use types::{ @@ -190,12 +190,16 @@ impl SessionData { ); if metadata.inner.mailboxes.len() == 1 { - // Tombstone message + // Delete message batch .custom(ObjectIndexBuilder::<_, ()>::new().with_current(metadata)) .caused_by(trc::location!())? .set( - ValueClass::TaskQueue(TaskQueueClass::UnindexEmail { due }), + ValueClass::TaskQueue(TaskQueueClass::UpdateIndex { + index: SearchIndex::Email, + due, + is_insert: false, + }), 0u64.serialize(), ) .commit_point(); diff --git a/crates/imap/src/op/status.rs b/crates/imap/src/op/status.rs index 2e5a80b2..aca7f83e 100644 --- a/crates/imap/src/op/status.rs +++ b/crates/imap/src/op/status.rs @@ -305,7 +305,7 @@ impl SessionData { collection: Collection::Email.into(), document_id: 0, class: ValueClass::IndexProperty(IndexPropertyClass::Integer { - property: EmailField::Stats.into(), + property: EmailField::ReceivedToSize.into(), value: 0, }), }, @@ -314,7 +314,7 @@ impl SessionData { collection: Collection::Email.into(), document_id: u32::MAX, class: ValueClass::IndexProperty(IndexPropertyClass::Integer { - property: EmailField::Stats.into(), + property: EmailField::ReceivedToSize.into(), value: u64::MAX, }), }, diff --git a/crates/jmap/src/calendar_event/mod.rs b/crates/jmap/src/calendar_event/mod.rs index aa7e549c..761ce14f 100644 --- a/crates/jmap/src/calendar_event/mod.rs +++ b/crates/jmap/src/calendar_event/mod.rs @@ -8,7 +8,7 @@ use calcard::jscalendar::JSCalendarProperty; use common::Server; use jmap_proto::error::set::SetError; use trc::AddContext; -use types::{collection::Collection, field::CalendarField, id::Id}; +use types::{collection::Collection, field::CalendarEventField, id::Id}; pub mod copy; pub mod get; @@ -67,7 +67,7 @@ pub(super) async fn assert_is_unique_uid( .document_exists( account_id, Collection::CalendarEvent, - CalendarField::Uid, + CalendarEventField::Uid, uid.as_bytes(), ) .await diff --git a/crates/jmap/src/calendar_event/query.rs b/crates/jmap/src/calendar_event/query.rs index b92f689b..e2a44fa1 100644 --- a/crates/jmap/src/calendar_event/query.rs +++ b/crates/jmap/src/calendar_event/query.rs @@ -4,7 +4,7 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use crate::changes::state::JmapCacheState; +use crate::{api::query::QueryResponseBuilder, changes::state::JmapCacheState}; use calcard::{common::timezone::Tz, jscalendar::JSCalendarDateTime}; use chrono::offset::TimeZone; use common::{Server, auth::AccessToken}; @@ -14,15 +14,17 @@ use jmap_proto::{ object::calendar_event::{self, CalendarEventComparator, CalendarEventFilter}, request::MaybeInvalid, }; -use nlp::tokenizers::word::WordTokenizer; +use nlp::language::Language; use std::{cmp::Ordering, sync::Arc}; -use store::{backend::MAX_TOKEN_LENGTH, roaring::RoaringBitmap, search::SearchFilter}; +use store::{ + roaring::RoaringBitmap, + search::{CalendarSearchField, SearchComparator, SearchFilter}, +}; use trc::AddContext; use types::{ TimeRange, acl::Acl, collection::{Collection, SyncCollection}, - field::CalendarField, }; pub trait CalendarEventQuery: Sync + Send { @@ -44,10 +46,7 @@ impl CalendarEventQuery for Server { let cache = self .fetch_dav_resources(access_token, account_id, SyncCollection::Calendar) .await?; - let filter_mask = (access_token.is_shared(account_id)) - .then(|| cache.shared_items(access_token, [Acl::ReadItems], true)); let default_tz = request.arguments.time_zone.unwrap_or(Tz::UTC); - let expand_recurrences = request.arguments.expand_recurrences.unwrap_or(false); let mut filter: Option = None; let mut did_filter_by_time = false; @@ -73,15 +72,73 @@ impl CalendarEventQuery for Server { ))) } CalendarEventFilter::Uid(uid) => { - filters.push(SearchFilter::eq(CalendarField::Uid, uid.into_bytes())) + filters.push(SearchFilter::eq(CalendarSearchField::Uid, uid)); } CalendarEventFilter::Text(value) => { - for token in WordTokenizer::new(&value, MAX_TOKEN_LENGTH) { - filters.push(SearchFilter::eq( - CalendarField::Text, - token.word.into_owned().into_bytes(), - )); - } + let (text, language) = + Language::detect(value, self.core.jmap.default_language); + filters.push(SearchFilter::Or); + filters.push(SearchFilter::has_text( + CalendarSearchField::Title, + text.clone(), + language, + )); + filters.push(SearchFilter::has_text( + CalendarSearchField::Description, + text.clone(), + language, + )); + filters.push(SearchFilter::has_text( + CalendarSearchField::Location, + text.clone(), + language, + )); + filters.push(SearchFilter::has_text( + CalendarSearchField::Owner, + text.clone(), + language, + )); + filters.push(SearchFilter::has_text( + CalendarSearchField::Attendee, + text, + language, + )); + filters.push(SearchFilter::End); + } + CalendarEventFilter::Title(title) => { + filters.push(SearchFilter::has_text_detect( + CalendarSearchField::Title, + title, + self.core.jmap.default_language, + )); + } + CalendarEventFilter::Description(description) => { + filters.push(SearchFilter::has_text_detect( + CalendarSearchField::Description, + description, + self.core.jmap.default_language, + )); + } + CalendarEventFilter::Location(location) => { + filters.push(SearchFilter::has_text_detect( + CalendarSearchField::Location, + location, + self.core.jmap.default_language, + )); + } + CalendarEventFilter::Owner(owner) => { + filters.push(SearchFilter::has_text( + CalendarSearchField::Owner, + owner, + Language::None, + )); + } + CalendarEventFilter::Attendee(attendee) => { + filters.push(SearchFilter::has_text( + CalendarSearchField::Attendee, + attendee, + Language::None, + )); } CalendarEventFilter::After(_) | CalendarEventFilter::Before(_) => { if let Some(filter) = &filter @@ -105,29 +162,74 @@ impl CalendarEventQuery for Server { .details(unsupported.into_string())); } }, - - Filter::And | Filter::Or | Filter::Not | Filter::Close => { - filters.push(cond.into()); + Filter::And => { + filters.push(SearchFilter::And); + } + Filter::Or => { + filters.push(SearchFilter::Or); + } + Filter::Not => { + filters.push(SearchFilter::Not); + } + Filter::Close => { + filters.push(SearchFilter::End); } } } - let mut result_set = self - .filter(account_id, Collection::CalendarEvent, filters) + let expand_recurrences = request.arguments.expand_recurrences.unwrap_or(false); + let comparators = if !expand_recurrences { + request + .sort + .take() + .unwrap_or_default() + .into_iter() + .map(|comparator| match comparator.property { + CalendarEventComparator::Start | CalendarEventComparator::RecurrenceId => { + Ok(SearchComparator::field( + CalendarSearchField::Start, + comparator.is_ascending, + )) + } + CalendarEventComparator::Uid => Ok(SearchComparator::field( + CalendarSearchField::Uid, + comparator.is_ascending, + )), + CalendarEventComparator::Created | CalendarEventComparator::Updated => { + Err(trc::JmapEvent::UnsupportedSort + .into_err() + .details(comparator.property.into_string().into_owned())) + } + CalendarEventComparator::_T(other) => Err(trc::JmapEvent::UnsupportedSort + .into_err() + .details(other.to_string())), + }) + .collect::, _>>()? + } else { + vec![] + }; + let results = self + .search_store() + .query(account_id, Collection::CalendarEvent, filters, comparators) .await?; - if let Some(filter_mask) = filter_mask { - result_set.apply_mask(filter_mask); - } + let mut response = QueryResponseBuilder::new( + results.len(), + self.core.jmap.query_max_results, + cache.get_state(false), + &request, + ); - let num_results = result_set.results.len() as usize; - if num_results > 0 { + if !results.is_empty() { // Extract comparators let comparators = request .sort .as_deref() .filter(|s| !s.is_empty()) .unwrap_or_default(); + let filter_mask = (access_token.is_shared(account_id)) + .then(|| cache.shared_items(access_token, [Acl::ReadItems], true)); + if expand_recurrences { let Some(time_range) = filter.filter(|f| f.start != i64::MIN && f.end != i64::MAX) else { @@ -136,12 +238,19 @@ impl CalendarEventQuery for Server { )); }; let max_instances = self.core.groupware.max_ical_instances; - let mut results = Vec::with_capacity(result_set.results.len() as usize); + let mut expanded_results = Vec::with_capacity(results.len() as usize); let has_uid_comparator = comparators .iter() .any(|c| matches!(c.property, CalendarEventComparator::Uid)); - for document_id in result_set.results { + for document_id in results { + if filter_mask + .as_ref() + .is_some_and(|filter_ids| !filter_ids.contains(document_id)) + { + continue; + } + let Some(_calendar_event) = self .archive(account_id, Collection::CalendarEvent, document_id) .await? @@ -171,14 +280,14 @@ impl CalendarEventQuery for Server { .expand(default_tz, time_range) .unwrap_or_default() { - if results.len() < max_instances { - results.push(SearchResult { + if expanded_results.len() < max_instances { + expanded_results.push(SearchResult { created: calendar_event.created.to_native().to_be_bytes(), updated: calendar_event.modified.to_native().to_be_bytes(), start: expansion.start.to_be_bytes(), uid: uid.clone(), document_id, - expansion_id: expansion.expansion_id, + expansion_id: expansion.expansion_id.into(), }); } else { return Err(trc::JmapEvent::InvalidArguments.into_err().details( @@ -189,8 +298,8 @@ impl CalendarEventQuery for Server { } // Sort results - if !results.is_empty() { - results.sort_by(|a, b| { + if !expanded_results.is_empty() { + expanded_results.sort_by(|a, b| { for comparator in comparators { let ordering = a .get_property(&comparator.property) @@ -208,69 +317,30 @@ impl CalendarEventQuery for Server { } Ordering::Equal }); - } - // Add results - let (mut response, paginate) = self - .build_query_response(results.len(), cache.get_state(false), &request) - .await?; - if let Some(mut paginate) = paginate { - for result in results { - if !paginate.add(result.expansion_id + 1, result.document_id) { + // Add results + for result in expanded_results { + if !response.add(result.expansion_id.unwrap() + 1, result.document_id) { break; } } - response.update_results(paginate.build())?; } - - Ok(response) } else { - let mut comparators_ = Vec::with_capacity(comparators.len()); - - for comparator in comparators { - comparators_.push(match &comparator.property { - CalendarEventComparator::Uid => { - SearchComparator::field(CalendarField::Uid, comparator.is_ascending) - } - CalendarEventComparator::Start => { - SearchComparator::field(CalendarField::Start, comparator.is_ascending) - } - CalendarEventComparator::Created => { - SearchComparator::field(CalendarField::Created, comparator.is_ascending) - } - CalendarEventComparator::Updated => { - SearchComparator::field(CalendarField::Updated, comparator.is_ascending) - } - unsupported => { - return Err(trc::JmapEvent::UnsupportedSort - .into_err() - .details(unsupported.clone().into_string())); - } - }); - } - - // Sort results - let (response, paginate) = self - .build_query_response(num_results, cache.get_state(false), &request) - .await?; - if let Some(paginate) = paginate { - self.sort(result_set, comparators_, paginate, response) - .await - } else { - Ok(response) + for document_id in results { + if filter_mask + .as_ref() + .is_some_and(|filter_ids| !filter_ids.contains(document_id)) + { + continue; + } + if !response.add(0, document_id) { + break; + } } } - } else { - let (response, _) = self - .build_query_response( - result_set.results.len() as usize, - cache.get_state(false), - &request, - ) - .await?; - - Ok(response) } + + response.build() } } @@ -282,7 +352,7 @@ fn local_timestamp(dt: &JSCalendarDateTime, tz: Tz) -> Option { #[derive(Debug)] struct SearchResult { - expansion_id: u32, + expansion_id: Option, document_id: u32, start: [u8; std::mem::size_of::()], created: [u8; std::mem::size_of::()], diff --git a/crates/jmap/src/calendar_event_notification/query.rs b/crates/jmap/src/calendar_event_notification/query.rs index e7f652cb..a630d38e 100644 --- a/crates/jmap/src/calendar_event_notification/query.rs +++ b/crates/jmap/src/calendar_event_notification/query.rs @@ -4,21 +4,28 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use crate::{ changes::state::JmapCacheState}; +use crate::{api::query::QueryResponseBuilder, changes::state::JmapCacheState}; use common::{Server, auth::AccessToken}; use groupware::cache::GroupwareCache; use jmap_proto::{ - method::query::{Comparator, Filter, QueryRequest, QueryResponse}, + method::query::{Filter, QueryRequest, QueryResponse}, object::calendar_event_notification::{ CalendarEventNotification, CalendarEventNotificationComparator, CalendarEventNotificationFilter, }, request::IntoValid, }; -use store::{SerializeInfallible, query}; +use store::{ + IterateParams, U32_LEN, U64_LEN, ValueKey, + ahash::AHashSet, + roaring::RoaringBitmap, + search::SearchFilter, + write::{IndexPropertyClass, ValueClass, key::DeserializeBigEndian}, +}; +use trc::AddContext; use types::{ collection::{Collection, SyncCollection}, - field::CalendarField, + field::CalendarNotificationField, }; pub trait CalendarEventNotificationQuery: Sync + Send { @@ -29,6 +36,12 @@ pub trait CalendarEventNotificationQuery: Sync + Send { ) -> impl Future> + Send; } +struct Notification { + document_id: u32, + created: u64, + event_id: u32, +} + impl CalendarEventNotificationQuery for Server { async fn calendar_event_notification_query( &self, @@ -44,36 +57,73 @@ impl CalendarEventNotificationQuery for Server { SyncCollection::CalendarEventNotification, ) .await?; + let mut notifications = Vec::with_capacity(16); + + self.store() + .iterate( + IterateParams::new( + ValueKey { + account_id, + collection: Collection::CalendarEventNotification.into(), + document_id: 0, + class: ValueClass::IndexProperty(IndexPropertyClass::Integer { + property: CalendarNotificationField::CreatedToId.into(), + value: 0, + }), + }, + ValueKey { + account_id, + collection: Collection::CalendarEventNotification.into(), + document_id: 0, + class: ValueClass::IndexProperty(IndexPropertyClass::Integer { + property: CalendarNotificationField::CreatedToId.into(), + value: u64::MAX, + }), + }, + ) + .ascending(), + |key, value| { + notifications.push(Notification { + document_id: key.deserialize_be_u32(key.len() - U32_LEN)?, + created: key.deserialize_be_u64(key.len() - U32_LEN - U64_LEN)?, + event_id: value.deserialize_be_u32(0)?, + }); + + Ok(true) + }, + ) + .await + .caused_by(trc::location!())?; for cond in std::mem::take(&mut request.filter) { match cond { Filter::Property(cond) => match cond { CalendarEventNotificationFilter::Before(before) => { - filters.push(SearchFilter::lt( - CalendarField::Created, - (before.timestamp() as u64).serialize(), - )) + let before = before.timestamp() as u64; + filters.push(SearchFilter::is_in_set(RoaringBitmap::from_iter( + notifications + .iter() + .filter_map(|n| (n.created < before).then_some(n.document_id)), + ))) } CalendarEventNotificationFilter::After(after) => { - filters.push(SearchFilter::gt( - CalendarField::Created, - (after.timestamp() as u64).serialize(), - )) + let after = after.timestamp() as u64; + filters.push(SearchFilter::is_in_set(RoaringBitmap::from_iter( + notifications + .iter() + .filter_map(|n| (n.created > after).then_some(n.document_id)), + ))) } CalendarEventNotificationFilter::CalendarEventIds(ids) => { - let has_many = ids.len() > 1; - if has_many { - filters.push(SearchFilter::Or); - } - for id in ids.into_valid() { - filters.push(SearchFilter::eq( - CalendarField::EventId, - id.document_id().serialize(), - )); - } - if has_many { - filters.push(SearchFilter::End); - } + let ids = ids + .into_valid() + .map(|id| id.document_id()) + .collect::>(); + filters.push(SearchFilter::is_in_set(RoaringBitmap::from_iter( + notifications + .iter() + .filter_map(|n| ids.contains(&n.event_id).then_some(n.document_id)), + ))) } unsupported => { return Err(trc::JmapEvent::UnsupportedFilter @@ -81,49 +131,67 @@ impl CalendarEventNotificationQuery for Server { .details(unsupported.into_string())); } }, - - Filter::And | Filter::Or | Filter::Not | Filter::Close => { - filters.push(cond.into()); + Filter::And => { + filters.push(SearchFilter::And); + } + Filter::Or => { + filters.push(SearchFilter::Or); + } + Filter::Not => { + filters.push(SearchFilter::Not); + } + Filter::Close => { + filters.push(SearchFilter::End); } } } - let result_set = self - .filter(account_id, Collection::CalendarEventNotification, filters) - .await?; + // Parse sort criteria + let mut is_ascending = true; + for comparator in request.sort.take().unwrap_or_default() { + match comparator.property { + CalendarEventNotificationComparator::Created => { + is_ascending = comparator.is_ascending; + } + CalendarEventNotificationComparator::_T(unsupported) => { + return Err(trc::JmapEvent::UnsupportedSort + .into_err() + .details(unsupported)); + } + }; + } + if !is_ascending { + notifications.reverse(); + } - let (response, paginate) = self - .build_query_response( - result_set.results.len() as usize, - cache.get_state(false), - &request, + let results = self + .search_store() + .query( + account_id, + Collection::CalendarEventNotification, + filters, + vec![], ) .await?; - if let Some(paginate) = paginate { - // Parse sort criteria - let mut comparators = Vec::with_capacity(request.sort.as_ref().map_or(1, |s| s.len())); - for comparator in request.sort.filter(|s| !s.is_empty()).unwrap_or_else(|| { - vec![Comparator::descending( - CalendarEventNotificationComparator::Created, - )] - }) { - comparators.push(match comparator.property { - CalendarEventNotificationComparator::Created => { - SearchComparator::field(CalendarField::Created, comparator.is_ascending) - } - CalendarEventNotificationComparator::_T(unsupported) => { - return Err(trc::JmapEvent::UnsupportedSort - .into_err() - .details(unsupported)); - } - }); - } + let mut response = QueryResponseBuilder::new( + results.len(), + self.core.jmap.query_max_results, + cache.get_state(false), + &request, + ); - // Sort results - self.sort(result_set, comparators, paginate, response).await - } else { - Ok(response) + if !results.is_empty() { + let results = results.into_iter().collect::>(); + for notification in notifications { + if results.contains(¬ification.document_id) + && !response.add(0, notification.document_id) + { + break; + } + } } + + response.build() } } diff --git a/crates/jmap/src/contact/mod.rs b/crates/jmap/src/contact/mod.rs index 52e6fde4..0f5cb5fa 100644 --- a/crates/jmap/src/contact/mod.rs +++ b/crates/jmap/src/contact/mod.rs @@ -7,7 +7,6 @@ use calcard::jscontact::JSContactProperty; use common::{DavName, DavResources, Server}; use jmap_proto::error::set::SetError; -use store::SearchFilter; use trc::AddContext; use types::{collection::Collection, field::ContactField, id::Id}; @@ -26,15 +25,15 @@ pub(super) async fn assert_is_unique_uid( ) -> trc::Result>>> { if let Some(uid) = uid { let hits = server - .store() - .filter( + .document_ids_matching( account_id, Collection::ContactCard, - vec![Filter::eq(ContactField::Uid, uid.as_bytes().to_vec())], + ContactField::Uid, + uid.as_bytes(), ) .await .caused_by(trc::location!())?; - if !hits.results.is_empty() { + if !hits.is_empty() { for document_id in resources .paths .iter() @@ -44,7 +43,7 @@ pub(super) async fn assert_is_unique_uid( }) .map(|path| resources.resources[path.resource_idx].document_id) { - if hits.results.contains(document_id) { + if hits.contains(document_id) { return Ok(Err(SetError::invalid_properties() .with_property(JSContactProperty::Uid) .with_description(format!( diff --git a/crates/jmap/src/contact/query.rs b/crates/jmap/src/contact/query.rs index 4b2bff78..3e46ec97 100644 --- a/crates/jmap/src/contact/query.rs +++ b/crates/jmap/src/contact/query.rs @@ -4,24 +4,24 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ +use crate::{api::query::QueryResponseBuilder, changes::state::JmapCacheState}; use common::{Server, auth::AccessToken}; use groupware::cache::GroupwareCache; use jmap_proto::{ - method::query::{Comparator, Filter, QueryRequest, QueryResponse}, + method::query::{Filter, QueryRequest, QueryResponse}, object::contact::{ContactCard, ContactCardComparator, ContactCardFilter}, request::MaybeInvalid, }; -use nlp::tokenizers::word::WordTokenizer; -use store::{SerializeInfallible, backend::MAX_TOKEN_LENGTH, query, roaring::RoaringBitmap}; +use store::{ + roaring::RoaringBitmap, + search::{ContactSearchField, SearchComparator, SearchFilter}, +}; use types::{ acl::Acl, collection::{Collection, SyncCollection}, - field::ContactField, }; use utils::sanitize_email; -use crate::{ changes::state::JmapCacheState}; - pub trait ContactCardQuery: Sync + Send { fn contact_card_query( &self, @@ -52,93 +52,188 @@ impl ContactCardQuery for Server { cache.children_ids(id.document_id()), ))) } - ContactCardFilter::Uid(uid) => { - filters.push(SearchFilter::eq(ContactField::Uid, uid.into_bytes())) + ContactCardFilter::Name(value) + | ContactCardFilter::NameGiven(value) + | ContactCardFilter::NameSurname(value) + | ContactCardFilter::NameSurname2(value) => { + filters.push(SearchFilter::has_unknown_text( + ContactSearchField::Name, + value, + )); + } + ContactCardFilter::Nickname(value) => { + filters.push(SearchFilter::has_unknown_text( + ContactSearchField::Nickname, + value, + )); + } + ContactCardFilter::Organization(value) => { + filters.push(SearchFilter::has_unknown_text( + ContactSearchField::Organization, + value, + )); + } + ContactCardFilter::Phone(value) => { + filters.push(SearchFilter::has_unknown_text( + ContactSearchField::Phone, + value, + )); + } + ContactCardFilter::OnlineService(value) => { + filters.push(SearchFilter::has_unknown_text( + ContactSearchField::OnlineService, + value, + )); + } + ContactCardFilter::Address(value) => { + filters.push(SearchFilter::has_unknown_text( + ContactSearchField::Address, + value, + )); + } + ContactCardFilter::Note(value) => { + filters.push(SearchFilter::has_text_detect( + ContactSearchField::Note, + value, + self.core.jmap.default_language, + )); + } + ContactCardFilter::HasMember(value) => { + filters.push(SearchFilter::has_unknown_text( + ContactSearchField::Member, + value, + )); + } + ContactCardFilter::Kind(value) => { + filters.push(SearchFilter::eq(ContactSearchField::Kind, value)); + } + ContactCardFilter::Uid(value) => { + filters.push(SearchFilter::eq(ContactSearchField::Uid, value)) + } + ContactCardFilter::Email(email) => { + filters.push(SearchFilter::has_unknown_text( + ContactSearchField::Email, + sanitize_email(&email).unwrap_or(email), + )) } - ContactCardFilter::Email(email) => filters.push(SearchFilter::eq( - ContactField::Email, - sanitize_email(&email).unwrap_or(email).into_bytes(), - )), ContactCardFilter::Text(value) => { - for token in WordTokenizer::new(&value, MAX_TOKEN_LENGTH) { - filters.push(SearchFilter::eq( - ContactField::Text, - token.word.into_owned().into_bytes(), - )); - } + filters.push(SearchFilter::Or); + filters.push(SearchFilter::has_unknown_text( + ContactSearchField::Name, + value.clone(), + )); + filters.push(SearchFilter::has_unknown_text( + ContactSearchField::Nickname, + value.clone(), + )); + filters.push(SearchFilter::has_unknown_text( + ContactSearchField::Organization, + value.clone(), + )); + filters.push(SearchFilter::has_unknown_text( + ContactSearchField::Email, + value.clone(), + )); + filters.push(SearchFilter::has_unknown_text( + ContactSearchField::Phone, + value.clone(), + )); + filters.push(SearchFilter::has_unknown_text( + ContactSearchField::OnlineService, + value.clone(), + )); + filters.push(SearchFilter::has_unknown_text( + ContactSearchField::Address, + value.clone(), + )); + filters.push(SearchFilter::has_text_detect( + ContactSearchField::Note, + value, + self.core.jmap.default_language, + )); + filters.push(SearchFilter::End); } ContactCardFilter::CreatedBefore(before) => filters.push(SearchFilter::lt( - ContactField::Created, - (before.timestamp() as u64).serialize(), + ContactSearchField::Created, + before.timestamp(), )), ContactCardFilter::CreatedAfter(after) => filters.push(SearchFilter::gt( - ContactField::Created, - (after.timestamp() as u64).serialize(), + ContactSearchField::Created, + after.timestamp(), )), - ContactCardFilter::UpdatedBefore(before) => filters.push(SearchFilter::lt( - ContactField::Updated, - (before.timestamp() as u64).serialize(), + /*ContactCardFilter::UpdatedBefore(before) => filters.push(SearchFilter::lt( + ContactSearchField::Updated, + before.timestamp(), )), ContactCardFilter::UpdatedAfter(after) => filters.push(SearchFilter::gt( - ContactField::Updated, - (after.timestamp() as u64).serialize(), - )), + ContactSearchField::Updated, + after.timestamp(), + )),*/ unsupported => { return Err(trc::JmapEvent::UnsupportedFilter .into_err() .details(unsupported.into_string())); } }, - - Filter::And | Filter::Or | Filter::Not | Filter::Close => { - filters.push(cond.into()); + Filter::And => { + filters.push(SearchFilter::And); + } + Filter::Or => { + filters.push(SearchFilter::Or); + } + Filter::Not => { + filters.push(SearchFilter::Not); + } + Filter::Close => { + filters.push(SearchFilter::End); } } } - let mut result_set = self - .filter(account_id, Collection::ContactCard, filters) + let comparators = request + .sort + .take() + .unwrap_or_default() + .into_iter() + .map(|comparator| match comparator.property { + ContactCardComparator::Created => Ok(SearchComparator::field( + ContactSearchField::Created, + comparator.is_ascending, + )), + /*ContactCardComparator::Updated => Ok(SearchComparator::field( + ContactSearchField::Updated, + comparator.is_ascending, + )),*/ + other => Err(trc::JmapEvent::UnsupportedSort + .into_err() + .details(other.into_string())), + }) + .collect::, _>>()?; + + let results = self + .search_store() + .query(account_id, Collection::ContactCard, filters, comparators) .await?; - if let Some(filter_mask) = filter_mask { - result_set.apply_mask(filter_mask); - } + let mut response = QueryResponseBuilder::new( + results.len(), + self.core.jmap.query_max_results, + cache.get_state(false), + &request, + ); - let (response, paginate) = self - .build_query_response( - result_set.results.len() as usize, - cache.get_state(false), - &request, - ) - .await?; - - if let Some(paginate) = paginate { - // Parse sort criteria - let mut comparators = Vec::with_capacity(request.sort.as_ref().map_or(1, |s| s.len())); - for comparator in request - .sort - .filter(|s| !s.is_empty()) - .unwrap_or_else(|| vec![Comparator::descending(ContactCardComparator::Updated)]) + for document_id in results { + if filter_mask + .as_ref() + .is_some_and(|filter_ids| !filter_ids.contains(document_id)) { - comparators.push(match comparator.property { - ContactCardComparator::Created => { - SearchComparator::field(ContactField::Created, comparator.is_ascending) - } - ContactCardComparator::Updated => { - SearchComparator::field(ContactField::Updated, comparator.is_ascending) - } - unsupported => { - return Err(trc::JmapEvent::UnsupportedSort - .into_err() - .details(unsupported.into_string())); - } - }); + continue; + } + if !response.add(0, document_id) { + break; } - - // Sort results - self.sort(result_set, comparators, paginate, response).await - } else { - Ok(response) } + + response.build() } } diff --git a/crates/jmap/src/file/query.rs b/crates/jmap/src/file/query.rs index e785858e..c41cb879 100644 --- a/crates/jmap/src/file/query.rs +++ b/crates/jmap/src/file/query.rs @@ -4,7 +4,7 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use crate::{ changes::state::JmapCacheState}; +use crate::{api::query::QueryResponseBuilder, changes::state::JmapCacheState}; use common::{Server, auth::AccessToken}; use groupware::cache::GroupwareCache; use jmap_proto::{ @@ -12,7 +12,7 @@ use jmap_proto::{ object::file_node::{FileNode, FileNodeFilter}, request::MaybeInvalid, }; -use store::{query, roaring::RoaringBitmap}; +use store::{roaring::RoaringBitmap, search::SearchFilter}; use types::{ acl::Acl, collection::{Collection, SyncCollection}, @@ -122,63 +122,51 @@ impl FileNodeQuery for Server { .details(unsupported.into_string())); } }, - - Filter::And | Filter::Or | Filter::Not | Filter::Close => { - filters.push(cond.into()); + Filter::And => { + filters.push(SearchFilter::And); + } + Filter::Or => { + filters.push(SearchFilter::Or); + } + Filter::Not => { + filters.push(SearchFilter::Not); + } + Filter::Close => { + filters.push(SearchFilter::End); } } } - let mut result_set = self - .filter(account_id, Collection::FileNode, filters) - .await?; - - if let Some(filter_mask) = filter_mask { - result_set.apply_mask(filter_mask); + if request.sort.as_ref().is_some_and(|s| !s.is_empty()) { + return Err(trc::JmapEvent::UnsupportedSort + .into_err() + .details("Sorting is not supported on FileNode")); } - let (response, paginate) = self - .build_query_response( - result_set.results.len() as usize, - cache.get_state(false), - &request, - ) + let results = self + .search_store() + .query(account_id, Collection::FileNode, filters, vec![]) .await?; - if let Some(paginate) = paginate { - // Parse sort criteria - /*let mut comparators = Vec::with_capacity(request.sort.as_ref().map_or(1, |s| s.len())); - for comparator in request - .sort - .filter(|s| !s.is_empty()) - .unwrap_or_else(|| vec![Comparator::descending(FileNodeComparator::Updated)]) + let mut response = QueryResponseBuilder::new( + results.len(), + self.core.jmap.query_max_results, + cache.get_state(false), + &request, + ); + + for document_id in results { + if filter_mask + .as_ref() + .is_some_and(|filter_ids| !filter_ids.contains(document_id)) { - comparators.push(match comparator.property { - FileNodeComparator::Created => { - SearchComparator::field(ContactField::Created, comparator.is_ascending) - } - FileNodeComparator::Updated => { - SearchComparator::field(ContactField::Updated, comparator.is_ascending) - } - unsupported => { - return Err(trc::JmapEvent::UnsupportedSort - .into_err() - .details(unsupported.into_string())); - } - }); - }*/ - - if request.sort.is_some_and(|s| !s.is_empty()) { - return Err(trc::JmapEvent::UnsupportedSort - .into_err() - .details("Sorting is not supported on FileNode")); + continue; + } + if !response.add(0, document_id) { + break; } - - // Sort results - self.sort(result_set, Default::default(), paginate, response) - .await - } else { - Ok(response) } + + response.build() } } diff --git a/crates/jmap/src/mailbox/query.rs b/crates/jmap/src/mailbox/query.rs index 99b2742d..cbfbb80d 100644 --- a/crates/jmap/src/mailbox/query.rs +++ b/crates/jmap/src/mailbox/query.rs @@ -108,7 +108,7 @@ impl MailboxQuery for Server { .items .iter() .filter(|mailbox| { - !matches!(mailbox.role, SpecialUse::None) == has_role + matches!(mailbox.role, SpecialUse::None) != has_role }) .map(|m| m.document_id) .collect::(), diff --git a/crates/jmap/src/principal/query.rs b/crates/jmap/src/principal/query.rs index ecdb7a16..8e336df6 100644 --- a/crates/jmap/src/principal/query.rs +++ b/crates/jmap/src/principal/query.rs @@ -4,7 +4,6 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use crate::JmapMethods; use common::{Server, auth::AccessToken}; use directory::{Permission, QueryParams, Type, backend::internal::manage::ManageDirectory}; use http_proto::HttpSessionData; @@ -14,10 +13,12 @@ use jmap_proto::{ types::state::State, }; use std::future::Future; -use store::{query::ResultSet, roaring::RoaringBitmap}; +use store::{roaring::RoaringBitmap, search::SearchFilter}; use trc::AddContext; use types::collection::Collection; +use crate::api::query::QueryResponseBuilder; + pub trait PrincipalQuery: Sync + Send { fn principal_query( &self, @@ -42,12 +43,6 @@ impl PrincipalQuery for Server { .details("The administrator has disabled directory queries.".to_string())); } - let mut result_set = ResultSet { - account_id: request.account_id.document_id(), - collection: Collection::Principal, - results: RoaringBitmap::new(), - }; - let mut is_set = true; let principal_ids = self .store() .list_principals( @@ -70,6 +65,7 @@ impl PrincipalQuery for Server { .map(|p| p.id()) .collect::(); + let mut filters = Vec::with_capacity(request.filter.len()); for cond in std::mem::take(&mut request.filter) { match cond { Filter::Property(cond) => match cond { @@ -81,74 +77,52 @@ impl PrincipalQuery for Server { .query(QueryParams::name(name.as_str()).with_return_member_of(false)) .await? { - if is_set || result_set.results.contains(principal.id()) { - result_set.results = - RoaringBitmap::from_sorted_iter([principal.id()]).unwrap(); - } else { - result_set.results = RoaringBitmap::new(); - } - } else { - result_set.results = RoaringBitmap::new(); + filters.push(SearchFilter::is_in_set( + RoaringBitmap::from_sorted_iter([principal.id()]).unwrap(), + )); } - is_set = false; } PrincipalFilter::Email(email) => { - let mut ids = RoaringBitmap::new(); if let Some(id) = self .email_to_id(self.directory(), &email, session.session_id) .await? { - ids.insert(id); - } - if is_set { - result_set.results = ids; - is_set = false; - } else { - result_set.results &= ids; + filters.push(SearchFilter::is_in_set( + RoaringBitmap::from_sorted_iter([id]).unwrap(), + )); } } PrincipalFilter::AccountIds(ids) => { - let ids = ids - .into_iter() - .filter_map(|id| { - let id = id.document_id(); - if principal_ids.contains(id) { - Some(id) - } else { - None - } - }) - .collect::(); - if is_set { - result_set.results = ids; - is_set = false; - } else { - result_set.results &= ids; - } + filters.push(SearchFilter::is_in_set( + ids.into_iter() + .filter_map(|id| { + let id = id.document_id(); + if principal_ids.contains(id) { + Some(id) + } else { + None + } + }) + .collect::(), + )); } PrincipalFilter::Text(text) => { - let ids = self - .store() - .list_principals( - Some(text.as_str()), - access_token.tenant.map(|t| t.id), - &[], - false, - 0, - 0, - ) - .await? - .items - .into_iter() - .map(|p| p.id()) - .collect::(); - - if is_set { - result_set.results = ids; - is_set = false; - } else { - result_set.results &= ids; - } + filters.push(SearchFilter::is_in_set( + self.store() + .list_principals( + Some(text.as_str()), + access_token.tenant.map(|t| t.id), + &[], + false, + 0, + 0, + ) + .await? + .items + .into_iter() + .map(|p| p.id()) + .collect::(), + )); } PrincipalFilter::Type(principal_type) => { let typ = match principal_type { @@ -159,28 +133,22 @@ impl PrincipalQuery for Server { PrincipalType::Other => Type::Other, }; - let ids = self - .store() - .list_principals( - None, - access_token.tenant.map(|t| t.id), - &[typ], - false, - 0, - 0, - ) - .await? - .items - .into_iter() - .map(|p| p.id()) - .collect::(); - - if is_set { - result_set.results = ids; - is_set = false; - } else { - result_set.results &= ids; - } + filters.push(SearchFilter::is_in_set( + self.store() + .list_principals( + None, + access_token.tenant.map(|t| t.id), + &[typ], + false, + 0, + 0, + ) + .await? + .items + .into_iter() + .map(|p| p.id()) + .collect::(), + )); } other => { return Err(trc::JmapEvent::UnsupportedFilter @@ -188,28 +156,39 @@ impl PrincipalQuery for Server { .details(other.to_string())); } }, - Filter::And | Filter::Or | Filter::Not | Filter::Close => { - return Err(trc::JmapEvent::UnsupportedFilter - .into_err() - .details("Logical operators are not supported")); + Filter::And => { + filters.push(SearchFilter::And); + } + Filter::Or => { + filters.push(SearchFilter::Or); + } + Filter::Not => { + filters.push(SearchFilter::Not); + } + Filter::Close => { + filters.push(SearchFilter::End); } } } - if is_set { - result_set.results = principal_ids; - } else { - result_set.results &= principal_ids; - } - - let (response, paginate) = self - .build_query_response(result_set.results.len() as usize, State::Initial, &request) + let results = self + .search_store() + .query(u32::MAX, Collection::Principal, filters, vec![]) .await?; - if let Some(paginate) = paginate { - self.sort(result_set, Vec::new(), paginate, response).await - } else { - Ok(response) + let mut response = QueryResponseBuilder::new( + results.len(), + self.core.jmap.query_max_results, + State::Initial, + &request, + ); + + for document_id in results { + if principal_ids.contains(document_id) && !response.add(0, document_id) { + break; + } } + + response.build() } } diff --git a/crates/jmap/src/share_notification/query.rs b/crates/jmap/src/share_notification/query.rs index 0c2c39de..06429e65 100644 --- a/crates/jmap/src/share_notification/query.rs +++ b/crates/jmap/src/share_notification/query.rs @@ -4,18 +4,15 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use std::time::Duration; - -use crate::{ UpdateResults}; +use crate::api::query::QueryResponseBuilder; use common::{Server, sharing::notification::ShareNotification}; use jmap_proto::{ method::query::{Filter, QueryRequest, QueryResponse}, object::share_notification::{self, ShareNotificationFilter}, types::state::State, }; -use store::{ - Deserialize, IterateParams, LogKey, U64_LEN, query::ResultSet, write::key::DeserializeBigEndian, -}; +use std::time::Duration; +use store::{Deserialize, IterateParams, LogKey, U64_LEN, write::key::DeserializeBigEndian}; use trc::AddContext; use types::{ collection::{Collection, SyncCollection}, @@ -79,12 +76,6 @@ impl ShareNotificationQuery for Server { } let mut results = Vec::new(); - let mut result_set = ResultSet { - account_id, - collection: Collection::None, - results: Default::default(), - }; - self.store() .iterate( IterateParams::new( @@ -113,7 +104,6 @@ impl ShareNotificationQuery for Server { } } - result_set.results.insert(results.len() as u32); results.push(Id::from(change_id)); Ok(true) @@ -122,20 +112,19 @@ impl ShareNotificationQuery for Server { .await .caused_by(trc::location!())?; - let (mut response, paginate) = self - .build_query_response(result_set.results.len() as usize, State::Initial, &request) - .await?; + let mut response = QueryResponseBuilder::new( + results.len(), + self.core.jmap.query_max_results, + State::Initial, + &request, + ); - if let Some(mut paginate) = paginate { - for result in results { - if !paginate.add_id(result) { - break; - } + for id in results { + if !response.add_id(id) { + break; } - - response.update_results(paginate.build())?; } - Ok(response) + response.build() } } diff --git a/crates/jmap/src/sieve/query.rs b/crates/jmap/src/sieve/query.rs index 6f981d88..95778530 100644 --- a/crates/jmap/src/sieve/query.rs +++ b/crates/jmap/src/sieve/query.rs @@ -4,18 +4,19 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use crate::changes::state::StateManager; +use crate::{api::query::QueryResponseBuilder, changes::state::StateManager}; use common::Server; use email::sieve::ingest::SieveScriptIngest; use jmap_proto::{ - method::query::{Comparator, Filter, QueryRequest, QueryResponse}, + method::query::{Filter, QueryRequest, QueryResponse}, object::sieve::{Sieve, SieveComparator, SieveFilter}, }; use std::future::Future; use store::{ - query::{self}, - roaring::RoaringBitmap, + IndexKeyPrefix, IterateParams, U32_LEN, ahash::AHashSet, roaring::RoaringBitmap, + search::SearchFilter, write::key::DeserializeBigEndian, }; +use trc::AddContext; use types::{ collection::{Collection, SyncCollection}, field::SieveField, @@ -48,73 +49,149 @@ impl SieveScriptQuery for Server { None }; + let mut document_ids = RoaringBitmap::new(); + let mut names = Vec::new(); + self.store() + .iterate( + IterateParams::new( + IndexKeyPrefix { + account_id, + collection: Collection::SieveScript.into(), + field: SieveField::Name.into(), + }, + IndexKeyPrefix { + account_id, + collection: Collection::SieveScript.into(), + field: u8::from(Collection::SieveScript) + 1, + }, + ) + .no_values(), + |key, _| { + let document_id = key.deserialize_be_u32(key.len() - U32_LEN)?; + + names.push(( + document_id, + key.get(IndexKeyPrefix::len()..key.len() - U32_LEN) + .and_then(|v| std::str::from_utf8(v).ok()) + .unwrap_or_default() + .to_string(), + )); + + document_ids.insert(document_id); + + Ok(true) + }, + ) + .await + .caused_by(trc::location!())?; + for cond in std::mem::take(&mut request.filter) { match cond { Filter::Property(cond) => match cond { SieveFilter::Name(name) => { - filters.push(SearchFilter::contains(SieveField::Name, &name)) + let name = name.to_lowercase(); + + filters.push(SearchFilter::is_in_set(RoaringBitmap::from_iter( + names + .iter() + .filter_map(|(id, n)| (n.contains(&name)).then_some(*id)) + .collect::>(), + ))); } SieveFilter::IsActive(is_active) => { - if !is_active { - filters.push(SearchFilter::Not); - } - filters.push(SearchFilter::is_in_set(RoaringBitmap::from_iter( - active_script_id, - ))); - if !is_active { - filters.push(SearchFilter::End); + let active_script_id = active_script_id.unwrap(); + + if is_active { + filters.push(SearchFilter::is_in_set(RoaringBitmap::from_iter([ + active_script_id, + ]))); + } else { + let mut inactive_set = document_ids.clone(); + inactive_set.remove(active_script_id); + filters.push(SearchFilter::is_in_set(inactive_set)); } } SieveFilter::_T(other) => { return Err(trc::JmapEvent::UnsupportedFilter.into_err().details(other)); } }, - - Filter::And | Filter::Or | Filter::Not | Filter::Close => { - filters.push(cond.into()); + Filter::And => { + filters.push(SearchFilter::And); + } + Filter::Or => { + filters.push(SearchFilter::Or); + } + Filter::Not => { + filters.push(SearchFilter::Not); + } + Filter::Close => { + filters.push(SearchFilter::End); } } } - let result_set = self - .filter(account_id, Collection::SieveScript, filters) - .await?; + // Parse sort criteria + let mut sort_by_active = None; + for comparator in request + .sort + .take() + .filter(|s| !s.is_empty()) + .unwrap_or_default() + { + match comparator.property { + SieveComparator::Name => { + if !comparator.is_ascending { + names.reverse(); + } + } + SieveComparator::IsActive => { + sort_by_active = Some(comparator.is_ascending); + } + SieveComparator::_T(other) => { + return Err(trc::JmapEvent::UnsupportedSort.into_err().details(other)); + } + }; + } - let (response, paginate) = self - .build_query_response( - result_set.results.len() as usize, - self.get_state(account_id, SyncCollection::SieveScript) - .await?, - &request, - ) - .await?; + let mut results = self + .search_store() + .query(account_id, Collection::SieveScript, filters, vec![]) + .await? + .into_iter() + .collect::>(); - if let Some(paginate) = paginate { - // Parse sort criteria - let mut comparators = Vec::with_capacity(request.sort.as_ref().map_or(1, |s| s.len())); - for comparator in request - .sort - .filter(|s| !s.is_empty()) - .unwrap_or_else(|| vec![Comparator::descending(SieveComparator::Name)]) + let mut response = QueryResponseBuilder::new( + results.len(), + self.core.jmap.query_max_results, + self.get_state(account_id, SyncCollection::SieveScript) + .await?, + &request, + ); + + if !results.is_empty() { + if matches!(sort_by_active, Some(true)) + && results.remove(&active_script_id.unwrap_or_default()) + && !response.add(0, active_script_id.unwrap()) { - comparators.push(match comparator.property { - SieveComparator::Name => { - SearchComparator::field(SieveField::Name, comparator.is_ascending) - } - SieveComparator::IsActive => SearchComparator::set( - RoaringBitmap::from_iter(active_script_id), - comparator.is_ascending, - ), - SieveComparator::_T(other) => { - return Err(trc::JmapEvent::UnsupportedSort.into_err().details(other)); - } - }); + return response.build(); } - // Sort results - self.sort(result_set, comparators, paginate, response).await - } else { - Ok(response) + let mut last_id = None; + for (document_id, _) in names { + if results.contains(&document_id) { + if sort_by_active.is_some() && Some(document_id) == active_script_id { + last_id = Some(document_id); + } else if !response.add(0, document_id) { + return response.build(); + } + } + } + + if let Some(active_id) = last_id { + response.add(0, active_id); + } } + + response.build() } } diff --git a/crates/jmap/src/sieve/set.rs b/crates/jmap/src/sieve/set.rs index ddcd682d..5295163b 100644 --- a/crates/jmap/src/sieve/set.rs +++ b/crates/jmap/src/sieve/set.rs @@ -404,13 +404,13 @@ impl SieveScriptSet for Server { .as_ref() .is_none_or(|(_, obj)| obj.inner.name != value.as_ref()) && let Some(id) = self - .filter( + .document_ids_matching( ctx.resource_token.account_id, Collection::SieveScript, - vec![Filter::eq(SieveField::Name, value.as_bytes().to_vec())], + SieveField::Name, + value.as_bytes(), ) .await? - .results .min() { return Ok(Err(SetError::already_exists() diff --git a/crates/jmap/src/submission/get.rs b/crates/jmap/src/submission/get.rs index 937e62a2..16284675 100644 --- a/crates/jmap/src/submission/get.rs +++ b/crates/jmap/src/submission/get.rs @@ -19,7 +19,11 @@ use jmap_tools::{Key, Map, Value}; use smtp::queue::{ArchivedError, ArchivedErrorDetails, ArchivedStatus, Message, spool::SmtpSpool}; use smtp_proto::ArchivedResponse; use std::future::Future; -use store::rkyv::option::ArchivedOption; +use store::{ + IterateParams, U32_LEN, ValueKey, + rkyv::option::ArchivedOption, + write::{IndexPropertyClass, ValueClass, key::DeserializeBigEndian, now}, +}; use trc::AddContext; use types::{ collection::{Collection, SyncCollection}, @@ -54,21 +58,45 @@ impl EmailSubmissionGet for Server { EmailSubmissionProperty::MdnBlobIds, ]); let account_id = request.account_id.document_id(); - let email_submission_ids = self - .document_ids( - account_id, - Collection::EmailSubmission, - EmailSubmissionField::EmailId, - ) - .await?; let ids = if let Some(ids) = ids { ids } else { - email_submission_ids - .iter() - .take(self.core.jmap.get_max_objects) - .map(Into::into) - .collect::>() + let mut ids = Vec::with_capacity(16); + + self.store() + .iterate( + IterateParams::new( + ValueKey { + account_id, + collection: Collection::CalendarEventNotification.into(), + document_id: 0, + class: ValueClass::IndexProperty(IndexPropertyClass::Integer { + property: EmailSubmissionField::Metadata.into(), + value: now() - (3 * 86400), + }), + }, + ValueKey { + account_id, + collection: Collection::CalendarEventNotification.into(), + document_id: 0, + class: ValueClass::IndexProperty(IndexPropertyClass::Integer { + property: EmailSubmissionField::Metadata.into(), + value: u64::MAX, + }), + }, + ) + .ascending() + .no_values(), + |key, _| { + ids.push(Id::from(key.deserialize_be_u32(key.len() - U32_LEN)?)); + + Ok(ids.len() < self.core.jmap.get_max_objects) + }, + ) + .await + .caused_by(trc::location!())?; + + ids }; let mut response = GetResponse { account_id: request.account_id.into(), @@ -83,10 +111,6 @@ impl EmailSubmissionGet for Server { for id in ids { // Obtain the email_submission object let document_id = id.document_id(); - if !email_submission_ids.contains(document_id) { - response.not_found.push(id); - continue; - } let submission_ = if let Some(submission) = self .archive(account_id, Collection::EmailSubmission, document_id) .await? diff --git a/crates/jmap/src/submission/query.rs b/crates/jmap/src/submission/query.rs index ecc52b76..073c1c95 100644 --- a/crates/jmap/src/submission/query.rs +++ b/crates/jmap/src/submission/query.rs @@ -4,19 +4,23 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use crate::changes::state::StateManager; +use crate::{api::query::QueryResponseBuilder, changes::state::StateManager}; use common::Server; use email::submission::UndoStatus; use jmap_proto::{ - method::query::{Comparator, Filter, QueryRequest, QueryResponse}, + method::query::{Filter, QueryRequest, QueryResponse}, object::email_submission::{self, EmailSubmissionComparator, EmailSubmissionFilter}, request::IntoValid, }; use std::future::Future; use store::{ - SerializeInfallible, - query::{self}, + IterateParams, U32_LEN, U64_LEN, ValueKey, + ahash::AHashSet, + roaring::RoaringBitmap, + search::SearchFilter, + write::{IndexPropertyClass, ValueClass, key::DeserializeBigEndian, now}, }; +use trc::AddContext; use types::{ collection::{Collection, SyncCollection}, field::EmailSubmissionField, @@ -29,123 +33,213 @@ pub trait EmailSubmissionQuery: Sync + Send { ) -> impl Future> + Send; } +struct Submission { + document_id: u32, + send_at: u64, + email_id: u32, + thread_id: u32, + identity_id: u32, + undo_status: u8, +} + impl EmailSubmissionQuery for Server { async fn email_submission_query( &self, mut request: QueryRequest, ) -> trc::Result { let account_id = request.account_id.document_id(); - let mut filters = Vec::with_capacity(request.filter.len()); + let mut submissions = Vec::with_capacity(16); + + self.store() + .iterate( + IterateParams::new( + ValueKey { + account_id, + collection: Collection::CalendarEventNotification.into(), + document_id: 0, + class: ValueClass::IndexProperty(IndexPropertyClass::Integer { + property: EmailSubmissionField::Metadata.into(), + value: now() - (3 * 86400), + }), + }, + ValueKey { + account_id, + collection: Collection::CalendarEventNotification.into(), + document_id: 0, + class: ValueClass::IndexProperty(IndexPropertyClass::Integer { + property: EmailSubmissionField::Metadata.into(), + value: u64::MAX, + }), + }, + ) + .ascending(), + |key, value| { + submissions.push(Submission { + document_id: key.deserialize_be_u32(key.len() - U32_LEN)?, + send_at: key.deserialize_be_u64(key.len() - U32_LEN - U64_LEN)?, + email_id: value.deserialize_be_u32(0)?, + thread_id: value.deserialize_be_u32(U32_LEN)?, + identity_id: value.deserialize_be_u32(U32_LEN + U32_LEN)?, + undo_status: value.last().copied().unwrap(), + }); + + Ok(true) + }, + ) + .await + .caused_by(trc::location!())?; + + let mut filters = Vec::with_capacity(request.filter.len()); for cond in std::mem::take(&mut request.filter) { match cond { Filter::Property(cond) => match cond { EmailSubmissionFilter::IdentityIds(ids) => { - filters.push(SearchFilter::Or); - for id in ids.into_valid() { - filters.push(SearchFilter::eq( - EmailSubmissionField::IdentityId, - id.document_id().serialize(), - )); - } - filters.push(SearchFilter::End); + let ids = ids + .into_valid() + .map(|id| id.document_id()) + .collect::>(); + + filters.push(SearchFilter::is_in_set(RoaringBitmap::from_iter( + submissions + .iter() + .filter(|s| ids.contains(&s.identity_id)) + .map(|s| s.document_id), + ))); } EmailSubmissionFilter::EmailIds(ids) => { - filters.push(SearchFilter::Or); - for id in ids.into_valid() { - filters.push(SearchFilter::eq( - EmailSubmissionField::EmailId, - id.id().serialize(), - )); - } - filters.push(SearchFilter::End); + let ids = ids + .into_valid() + .map(|id| id.document_id()) + .collect::>(); + + filters.push(SearchFilter::is_in_set(RoaringBitmap::from_iter( + submissions + .iter() + .filter(|s| ids.contains(&s.email_id)) + .map(|s| s.document_id), + ))); } EmailSubmissionFilter::ThreadIds(ids) => { - filters.push(SearchFilter::Or); - for id in ids.into_valid() { - filters.push(SearchFilter::eq( - EmailSubmissionField::ThreadId, - id.document_id().serialize(), - )); - } - filters.push(SearchFilter::End); + let ids = ids + .into_valid() + .map(|id| id.document_id()) + .collect::>(); + + filters.push(SearchFilter::is_in_set(RoaringBitmap::from_iter( + submissions + .iter() + .filter(|s| ids.contains(&s.thread_id)) + .map(|s| s.document_id), + ))); } EmailSubmissionFilter::UndoStatus(undo_status) => { - filters.push(SearchFilter::eq( - EmailSubmissionField::UndoStatus, - match undo_status { - email_submission::UndoStatus::Pending => UndoStatus::Pending, - email_submission::UndoStatus::Final => UndoStatus::Final, - email_submission::UndoStatus::Canceled => UndoStatus::Canceled, - } - .as_index() - .serialize(), - )) + let undo_status = match undo_status { + email_submission::UndoStatus::Pending => UndoStatus::Pending, + email_submission::UndoStatus::Final => UndoStatus::Final, + email_submission::UndoStatus::Canceled => UndoStatus::Canceled, + } + .as_index(); + + filters.push(SearchFilter::is_in_set(RoaringBitmap::from_iter( + submissions + .iter() + .filter(|s| s.undo_status == undo_status) + .map(|s| s.document_id), + ))); + } + EmailSubmissionFilter::Before(before) => { + let before = before.timestamp() as u64; + + filters.push(SearchFilter::is_in_set(RoaringBitmap::from_iter( + submissions + .iter() + .filter(|s| s.send_at < before) + .map(|s| s.document_id), + ))); + } + EmailSubmissionFilter::After(after) => { + let after = after.timestamp() as u64; + + filters.push(SearchFilter::is_in_set(RoaringBitmap::from_iter( + submissions + .iter() + .filter(|s| s.send_at > after) + .map(|s| s.document_id), + ))); } - EmailSubmissionFilter::Before(before) => filters.push(SearchFilter::lt( - EmailSubmissionField::SendAt, - (before.timestamp() as u64).serialize(), - )), - EmailSubmissionFilter::After(after) => filters.push(SearchFilter::gt( - EmailSubmissionField::SendAt, - (after.timestamp() as u64).serialize(), - )), EmailSubmissionFilter::_T(other) => { return Err(trc::JmapEvent::UnsupportedFilter.into_err().details(other)); } }, - - Filter::And | Filter::Or | Filter::Not | Filter::Close => { - filters.push(cond.into()); + Filter::And => { + filters.push(SearchFilter::And); + } + Filter::Or => { + filters.push(SearchFilter::Or); + } + Filter::Not => { + filters.push(SearchFilter::Not); + } + Filter::Close => { + filters.push(SearchFilter::End); } } } - let result_set = self - .filter(account_id, Collection::EmailSubmission, filters) - .await?; + let results = self + .search_store() + .query(account_id, Collection::ContactCard, filters, vec![]) + .await? + .into_iter() + .collect::>(); - let (response, paginate) = self - .build_query_response( - result_set.results.len() as usize, - self.get_state(account_id, SyncCollection::EmailSubmission) - .await?, - &request, - ) - .await?; + let mut response = QueryResponseBuilder::new( + results.len(), + self.core.jmap.query_max_results, + self.get_state(account_id, SyncCollection::EmailSubmission) + .await?, + &request, + ); - if let Some(paginate) = paginate { - // Parse sort criteria - let mut comparators = Vec::with_capacity(request.sort.as_ref().map_or(1, |s| s.len())); - for comparator in request - .sort - .filter(|s| !s.is_empty()) - .unwrap_or_else(|| vec![Comparator::descending(EmailSubmissionComparator::SentAt)]) - { - comparators.push(match comparator.property { - EmailSubmissionComparator::EmailId => SearchComparator::field( - EmailSubmissionField::EmailId, - comparator.is_ascending, - ), - EmailSubmissionComparator::ThreadId => SearchComparator::field( - EmailSubmissionField::ThreadId, - comparator.is_ascending, - ), - EmailSubmissionComparator::SentAt => SearchComparator::field( - EmailSubmissionField::SendAt, - comparator.is_ascending, - ), + if !results.is_empty() { + if let Some(comparator) = request.sort.take().unwrap_or_default().into_iter().next() { + match comparator.property { + EmailSubmissionComparator::EmailId => { + if comparator.is_ascending { + submissions.sort_by_key(|s| s.email_id); + } else { + submissions.sort_by_key(|s| u32::MAX - s.email_id); + } + } + EmailSubmissionComparator::ThreadId => { + if comparator.is_ascending { + submissions.sort_by_key(|s| s.thread_id); + } else { + submissions.sort_by_key(|s| u32::MAX - s.thread_id); + } + } + EmailSubmissionComparator::SentAt => { + if !comparator.is_ascending { + submissions.reverse(); + } + } EmailSubmissionComparator::_T(other) => { return Err(trc::JmapEvent::UnsupportedSort.into_err().details(other)); } - }); + } } - // Sort results - self.sort(result_set, comparators, paginate, response).await - } else { - Ok(response) + for submission in submissions { + if results.contains(&submission.document_id) + && !response.add(0, submission.document_id) + { + break; + } + } } + + response.build() } } diff --git a/crates/jmap/src/thread/get.rs b/crates/jmap/src/thread/get.rs index 343a0d26..82119215 100644 --- a/crates/jmap/src/thread/get.rs +++ b/crates/jmap/src/thread/get.rs @@ -14,17 +14,9 @@ use jmap_proto::{ }; use jmap_tools::Map; use std::future::Future; -use store::{ - ahash::AHashMap, - query::{Comparator, ResultSet, sort::Pagination}, - roaring::RoaringBitmap, -}; +use store::{ahash::AHashMap, roaring::RoaringBitmap}; use trc::AddContext; -use types::{ - collection::{Collection, SyncCollection}, - field::EmailField, - id::Id, -}; +use types::{collection::SyncCollection, id::Id}; pub trait ThreadGet: Sync + Send { fn thread_get( @@ -83,23 +75,11 @@ impl ThreadGet for Server { let mut thread: Map<'_, ThreadProperty, ThreadValue> = Map::with_capacity(2).with_key_value(ThreadProperty::Id, id); if add_email_ids { - let doc_count = document_ids.len() as usize; - let todo = " sorted as vec![Comparator::ascending(EmailField::ReceivedAt)],"; thread.insert_unchecked( ThreadProperty::EmailIds, - self.core - .storage - .data - .sort( - ResultSet::new(account_id, Collection::Email, document_ids), - vec![], - Pagination::new(doc_count, 0, None, 0), - ) - .await - .caused_by(trc::location!())? - .ids + document_ids .into_iter() - .map(|id| Id::from_parts(thread_id, id.document_id())) + .map(|id| Id::from_parts(thread_id, id)) .collect::>(), ); } diff --git a/crates/jmap/src/vacation/get.rs b/crates/jmap/src/vacation/get.rs index 5c348e47..12618b71 100644 --- a/crates/jmap/src/vacation/get.rs +++ b/crates/jmap/src/vacation/get.rs @@ -4,7 +4,7 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use crate::{ changes::state::StateManager}; +use crate::changes::state::StateManager; use common::Server; use email::sieve::{SieveScript, ingest::SieveScriptIngest}; use jmap_proto::{ @@ -17,7 +17,6 @@ use jmap_proto::{ }; use jmap_tools::{Map, Value}; use std::future::Future; -use store::SearchFilter; use trc::AddContext; use types::{ collection::{Collection, SyncCollection}, @@ -162,12 +161,13 @@ impl VacationResponseGet for Server { } async fn get_vacation_sieve_script_id(&self, account_id: u32) -> trc::Result> { - self.filter( + self.document_ids_matching( account_id, Collection::SieveScript, - vec![Filter::eq(SieveField::Name, "vacation".as_bytes().to_vec())], + SieveField::Name, + "vacation".as_bytes(), ) .await - .map(|r| r.results.min()) + .map(|r| r.min()) } } diff --git a/crates/pop3/src/mailbox.rs b/crates/pop3/src/mailbox.rs index defa4058..390df363 100644 --- a/crates/pop3/src/mailbox.rs +++ b/crates/pop3/src/mailbox.rs @@ -66,7 +66,7 @@ impl Session { collection: Collection::Email.into(), document_id: 0, class: ValueClass::IndexProperty(IndexPropertyClass::Integer { - property: EmailField::Stats.into(), + property: EmailField::ReceivedToSize.into(), value: 0, }), }, @@ -75,7 +75,7 @@ impl Session { collection: Collection::Email.into(), document_id: u32::MAX, class: ValueClass::IndexProperty(IndexPropertyClass::Integer { - property: EmailField::Stats.into(), + property: EmailField::ReceivedToSize.into(), value: u64::MAX, }), }, diff --git a/crates/services/src/task_manager/fts.rs b/crates/services/src/task_manager/fts.rs index 0014a98d..b6e2a8ce 100644 --- a/crates/services/src/task_manager/fts.rs +++ b/crates/services/src/task_manager/fts.rs @@ -6,13 +6,16 @@ use common::Server; use directory::{Type, backend::internal::manage::ManageDirectory}; -use email::message::{index::IndexMessageText, metadata::MessageMetadata}; +use email::message::metadata::MessageMetadata; use std::time::Instant; use store::{ IterateParams, SerializeInfallible, U32_LEN, ValueKey, ahash::AHashMap, roaring::RoaringBitmap, - write::{BatchBuilder, BlobOp, TaskQueueClass, ValueClass, key::DeserializeBigEndian, now}, + write::{ + BatchBuilder, BlobOp, SearchIndex, TaskQueueClass, ValueClass, key::DeserializeBigEndian, + now, + }, }; use trc::{AddContext, MessageIngestEvent, TaskQueueEvent}; use types::{ @@ -418,7 +421,11 @@ impl FtsIndexTask for Server { for document_id in document_ids { batch.with_document(document_id).set( - ValueClass::TaskQueue(TaskQueueClass::IndexEmail { due }), + ValueClass::TaskQueue(TaskQueueClass::UpdateIndex { + due, + index: SearchIndex::Email, + is_insert: true, + }), 0u64.serialize(), ); diff --git a/crates/services/src/task_manager/mod.rs b/crates/services/src/task_manager/mod.rs index 55c94c59..66e6b0c3 100644 --- a/crates/services/src/task_manager/mod.rs +++ b/crates/services/src/task_manager/mod.rs @@ -18,6 +18,7 @@ use std::time::Duration; use std::{sync::Arc, time::Instant}; use store::rand; use store::rand::seq::SliceRandom; +use store::write::SearchIndex; use store::{ IterateParams, U16_LEN, U32_LEN, U64_LEN, ValueKey, ahash::AHashMap, @@ -47,8 +48,7 @@ pub struct Task { #[derive(Debug, Clone, Hash, PartialEq, Eq)] pub enum TaskAction { - Index, - Unindex, + UpdateIndex { index: SearchIndex, is_insert: bool }, BayesTrain { learn_spam: bool }, SendAlarm { alarm: CalendarAlarm }, SendImip, @@ -102,17 +102,13 @@ pub fn spawn_task_manager(inner: Arc) { // Lock task if server.try_lock_task(&task).await { let success = match &task.action { - TaskAction::Index => { + TaskAction::UpdateIndex { index, is_insert } => { let todo = "implement"; /*server .fts_index(task.account_id, task.document_id, hash) .await*/ true } - TaskAction::Unindex => { - let todo = "implement"; - true - } TaskAction::BayesTrain { learn_spam } => { let todo = "implement"; /*server @@ -211,14 +207,20 @@ impl TaskQueueManager for Server { account_id: 0, collection: 0, document_id: 0, - class: ValueClass::TaskQueue(TaskQueueClass::IndexEmail { due: 0 }), + class: ValueClass::TaskQueue(TaskQueueClass::UpdateIndex { + due: 0, + index: SearchIndex::Email, + is_insert: true, + }), }; let to_key = ValueKey:: { account_id: u32::MAX, collection: u8::MAX, document_id: u32::MAX, - class: ValueClass::TaskQueue(TaskQueueClass::IndexEmail { + class: ValueClass::TaskQueue(TaskQueueClass::UpdateIndex { due: now_timestamp + QUEUE_REFRESH_INTERVAL, + index: SearchIndex::Email, + is_insert: true, }), }; @@ -286,7 +288,9 @@ impl TaskQueueManager for Server { let roles = &self.core.network.roles; for event in tasks { let tx = match &event.action { - TaskAction::Index { .. } if roles.fts_indexing.is_enabled_for_hash(&event) => { + TaskAction::UpdateIndex { .. } + if roles.fts_indexing.is_enabled_for_hash(&event) => + { &ipc.tx_fts } TaskAction::BayesTrain { .. } @@ -379,12 +383,15 @@ impl Task { fn lock_key(&self) -> Vec { match &self.action { - TaskAction::Index => KeySerializer::new((U32_LEN * 2) + U64_LEN + 1) - .write(0u8) - .write(self.due) - .write_leb128(self.account_id) - .write_leb128(self.document_id) - .finalize(), + TaskAction::UpdateIndex { index, .. } => { + KeySerializer::new((U32_LEN * 2) + U64_LEN + 2) + .write(0u8) + .write(self.due) + .write_leb128(self.account_id) + .write_leb128(self.document_id) + .write(index.to_u8()) + .finalize() + } TaskAction::BayesTrain { .. } => KeySerializer::new((U32_LEN * 2) + 1) .write(1u8) .write_leb128(self.account_id) @@ -402,18 +409,12 @@ impl Task { .write_leb128(self.account_id) .write_leb128(self.document_id) .finalize(), - TaskAction::Unindex => KeySerializer::new((U32_LEN * 2) + U64_LEN + 1) - .write(4u8) - .write(self.due) - .write_leb128(self.account_id) - .write_leb128(self.document_id) - .finalize(), } } fn lock_expiry(&self) -> u64 { match self.action { - TaskAction::Index | TaskAction::Unindex => INDEX_EXPIRY, + TaskAction::UpdateIndex { .. } => INDEX_EXPIRY, TaskAction::BayesTrain { .. } => BAYES_LOCK_EXPIRY, TaskAction::SendAlarm { .. } | TaskAction::SendImip => ALARM_EXPIRY, } @@ -422,8 +423,11 @@ impl Task { fn value_classes(&self) -> impl Iterator { [ Some(ValueClass::TaskQueue(match &self.action { - TaskAction::Index => TaskQueueClass::IndexEmail { due: self.due }, - TaskAction::Unindex => TaskQueueClass::UnindexEmail { due: self.due }, + TaskAction::UpdateIndex { index, is_insert } => TaskQueueClass::UpdateIndex { + due: self.due, + index: *index, + is_insert: *is_insert, + }, TaskAction::BayesTrain { learn_spam } => TaskQueueClass::BayesTrain { due: self.due, learn_spam: *learn_spam, @@ -456,8 +460,14 @@ impl Task { account_id: key.deserialize_be_u32(U64_LEN)?, document_id: key.deserialize_be_u32(U64_LEN + U32_LEN + 1)?, action: match key.get(U64_LEN + U32_LEN) { - Some(0) => TaskAction::Index, - Some(7) => TaskAction::Unindex, + Some(v @ (7 | 8)) => TaskAction::UpdateIndex { + index: key + .last() + .copied() + .and_then(SearchIndex::try_from_u8) + .ok_or_else(|| trc::Error::corrupted_key(key, None, trc::location!()))?, + is_insert: *v == 7, + }, Some(v @ (1 | 2)) => TaskAction::BayesTrain { learn_spam: *v == 1, }, diff --git a/crates/store/src/search/mod.rs b/crates/store/src/search/mod.rs index aacfe3e7..eb19819f 100644 --- a/crates/store/src/search/mod.rs +++ b/crates/store/src/search/mod.rs @@ -8,10 +8,10 @@ pub mod index; pub mod local; pub mod query; +use ahash::AHashMap; use nlp::language::Language; use roaring::RoaringBitmap; -use std::borrow::Cow; -use types::collection::Collection; +use std::{borrow::Cow, collections::hash_map::Entry}; #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum SearchOperator { @@ -24,7 +24,7 @@ pub enum SearchOperator { Exists, } -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq, Hash)] pub enum SearchField { Email(EmailSearchField), Calendar(CalendarSearchField), @@ -32,7 +32,7 @@ pub enum SearchField { File(FileSearchField), } -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq, Hash)] pub enum EmailSearchField { From, To, @@ -48,17 +48,34 @@ pub enum EmailSearchField { Header(Cow<'static, str>), } -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] pub enum CalendarSearchField { - Summary, + Title, + Description, + Location, + Owner, + Attendee, + Start, + Uid, } -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] pub enum ContactSearchField { + Created, + Member, + Kind, Name, + Nickname, + Organization, + Email, + Phone, + OnlineService, + Address, + Note, + Uid, } -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] pub enum FileSearchField { Name, Content, @@ -95,18 +112,11 @@ pub enum SearchComparator { #[derive(Debug)] pub struct IndexDocument { pub(crate) account_id: u32, - pub(crate) collection: Collection, pub(crate) document_id: u32, - pub(crate) fields: Vec, + pub(crate) fields: AHashMap, pub(crate) default_language: Language, } -#[derive(Debug)] -pub struct IndexField { - pub(crate) field: SearchField, - pub(crate) value: SearchValue, -} - impl SearchFilter { pub fn cond( field: impl Into, @@ -214,10 +224,16 @@ impl SearchFilter { } } + #[inline(always)] pub fn has_english_text(field: impl Into, text: impl Into) -> Self { Self::has_text(field, text, Language::English) } + #[inline(always)] + pub fn has_unknown_text(field: impl Into, text: impl Into) -> Self { + Self::has_text(field, text, Language::Unknown) + } + pub fn is_in_set(set: RoaringBitmap) -> Self { SearchFilter::DocumentSet(set) } @@ -257,11 +273,10 @@ impl SearchComparator { impl IndexDocument { pub fn with_default_language(default_language: Language) -> Self { Self { - fields: vec![], + fields: Default::default(), default_language, account_id: 0, document_id: 0, - collection: Collection::None, } } @@ -275,16 +290,39 @@ impl IndexDocument { self } - pub fn with_collection(mut self, collection: Collection) -> Self { - self.collection = collection; - self + pub fn index_text(&mut self, field: impl Into, value: &str, language: Language) { + match self.fields.entry(field.into()) { + Entry::Occupied(mut entry) => { + if let SearchValue::Text { + value: existing_value, + .. + } = entry.get_mut() + { + existing_value.push(' '); + existing_value.push_str(value); + } + } + Entry::Vacant(entry) => { + entry.insert(SearchValue::Text { + value: value.to_string(), + language, + }); + } + } } - pub fn index(&mut self, field: impl Into, value: impl Into) { - self.fields.push(IndexField { - field: field.into(), - value: value.into(), - }); + pub fn index_bool(&mut self, field: impl Into, value: bool) { + self.fields + .insert(field.into(), SearchValue::Boolean(value)); + } + + pub fn index_number>(&mut self, field: impl Into, value: N) { + self.fields + .insert(field.into(), SearchValue::Number(value.into())); + } + + pub fn has_field(&self, field: &SearchField) -> bool { + self.fields.contains_key(field) } } diff --git a/crates/store/src/write/key.rs b/crates/store/src/write/key.rs index fe6f2945..91b6eeb8 100644 --- a/crates/store/src/write/key.rs +++ b/crates/store/src/write/key.rs @@ -15,7 +15,8 @@ use crate::{ SUBSPACE_QUEUE_EVENT, SUBSPACE_QUEUE_MESSAGE, SUBSPACE_QUOTA, SUBSPACE_REPORT_IN, SUBSPACE_REPORT_OUT, SUBSPACE_SETTINGS, SUBSPACE_TASK_QUEUE, SUBSPACE_TELEMETRY_INDEX, SUBSPACE_TELEMETRY_METRIC, SUBSPACE_TELEMETRY_SPAN, U16_LEN, U32_LEN, U64_LEN, ValueKey, - WITH_SUBSPACE, write::IndexPropertyClass, + WITH_SUBSPACE, + write::{IndexPropertyClass, SearchIndex}, }; use std::convert::TryInto; use types::{blob_hash::BLOB_HASH_LEN, collection::SyncCollection}; @@ -279,15 +280,15 @@ impl ValueClass { .write(document_id), ValueClass::TaskQueue(task) => match task { TaskQueueClass::UpdateIndex { - collection, + index, is_insert, due, } => serializer .write(*due) .write(account_id) .write(if *is_insert { 7u8 } else { 8u8 }) - .write(u8::from(*collection)) - .write(document_id), + .write(document_id) + .write(index.to_u8()), TaskQueueClass::BayesTrain { due, learn_spam } => serializer .write(*due) .write(account_id) @@ -657,3 +658,26 @@ impl Deserialize for ReportEvent { }) } } + +impl SearchIndex { + pub fn to_u8(&self) -> u8 { + match self { + SearchIndex::Email => 0, + SearchIndex::Calendar => 1, + SearchIndex::Contacts => 2, + SearchIndex::File => 3, + SearchIndex::DeliveryHistory => 4, + } + } + + pub fn try_from_u8(value: u8) -> Option { + match value { + 0 => Some(SearchIndex::Email), + 1 => Some(SearchIndex::Calendar), + 2 => Some(SearchIndex::Contacts), + 3 => Some(SearchIndex::File), + 4 => Some(SearchIndex::DeliveryHistory), + _ => None, + } + } +} diff --git a/crates/store/src/write/mod.rs b/crates/store/src/write/mod.rs index 02bdde16..97a60f17 100644 --- a/crates/store/src/write/mod.rs +++ b/crates/store/src/write/mod.rs @@ -192,7 +192,7 @@ pub enum IndexPropertyClass { pub enum TaskQueueClass { UpdateIndex { due: u64, - collection: Collection, + index: SearchIndex, is_insert: bool, }, BayesTrain { @@ -211,6 +211,15 @@ pub enum TaskQueueClass { }, } +#[derive(Debug, PartialEq, Clone, Copy, Eq, Hash)] +pub enum SearchIndex { + Email, + Calendar, + Contacts, + File, + DeliveryHistory, +} + #[derive(Debug, PartialEq, Clone, Eq, Hash)] pub struct AnyClass { pub subspace: u8, diff --git a/crates/types/src/field.rs b/crates/types/src/field.rs index 1cddbcd0..8137f78b 100644 --- a/crates/types/src/field.rs +++ b/crates/types/src/field.rs @@ -40,7 +40,7 @@ pub enum CalendarNotificationField { pub enum EmailField { Archive, Metadata, - Stats, + ReceivedToSize, Threading, } @@ -63,11 +63,7 @@ pub enum SieveField { #[repr(u8)] pub enum EmailSubmissionField { Archive, - UndoStatus, - EmailId, - ThreadId, - IdentityId, - SendAt, + Metadata, } #[derive(Clone, Copy, Debug, PartialEq, Eq)] @@ -115,7 +111,7 @@ impl From for u8 { match value { EmailField::Metadata => 71, EmailField::Threading => 90, - EmailField::Stats => 91, + EmailField::ReceivedToSize => 91, EmailField::Archive => ARCHIVE_FIELD, } } @@ -143,11 +139,7 @@ impl From for u8 { impl From for u8 { fn from(value: EmailSubmissionField) -> Self { match value { - EmailSubmissionField::UndoStatus => 41, - EmailSubmissionField::EmailId => 83, - EmailSubmissionField::ThreadId => 33, - EmailSubmissionField::IdentityId => 95, - EmailSubmissionField::SendAt => 24, + EmailSubmissionField::Metadata => 49, EmailSubmissionField::Archive => ARCHIVE_FIELD, } }