diff --git a/crates/common/src/lib.rs b/crates/common/src/lib.rs index 6f587ad8..3be14500 100644 --- a/crates/common/src/lib.rs +++ b/crates/common/src/lib.rs @@ -102,6 +102,7 @@ pub const KV_LOCK_QUEUE_REPORT: u8 = 22; pub const KV_LOCK_EMAIL_TASK: u8 = 23; pub const KV_LOCK_HOUSEKEEPER: u8 = 24; pub const KV_LOCK_DAV: u8 = 25; +pub const KV_SIEVE_ID: u8 = 27; #[derive(Clone)] pub struct Server { diff --git a/crates/common/src/storage/index.rs b/crates/common/src/storage/index.rs index 6485f115..bea85fb3 100644 --- a/crates/common/src/storage/index.rs +++ b/crates/common/src/storage/index.rs @@ -4,13 +4,14 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ +use ahash::AHashSet; use jmap_proto::types::{property::Property, value::AclGrant}; -use std::{borrow::Cow, collections::HashSet, fmt::Debug}; +use std::{borrow::Cow, fmt::Debug}; use store::{ Serialize, SerializeInfallible, SerializedVersion, write::{ Archive, Archiver, BatchBuilder, BitmapClass, BlobOp, DirectoryClass, IntoOperations, - Operation, + MaybeDynamicId, Operation, TagValue, }, }; use utils::BlobHash; @@ -19,14 +20,35 @@ use crate::auth::AsTenantId; #[derive(Debug, Clone, PartialEq, Eq)] pub enum IndexValue<'x> { - Text { field: u8, value: Cow<'x, str> }, - U32 { field: u8, value: Option }, - U64 { field: u8, value: Option }, - U32List { field: u8, value: Cow<'x, [u32]> }, - Tag { field: u8, is_set: bool }, - Blob { value: BlobHash }, - Quota { used: u32 }, - Acl { value: Cow<'x, [AclGrant]> }, + Text { + field: u8, + value: Cow<'x, str>, + }, + U32 { + field: u8, + value: Option, + }, + U64 { + field: u8, + value: Option, + }, + U32List { + field: u8, + value: Cow<'x, [u32]>, + }, + Tag { + field: u8, + value: Vec>, + }, + Blob { + value: BlobHash, + }, + Quota { + used: u32, + }, + Acl { + value: Cow<'x, [AclGrant]>, + }, } pub trait IndexableObject: Sync + Send { @@ -178,13 +200,10 @@ fn build_index(batch: &mut BatchBuilder, item: IndexValue<'_>, tenant_id: Option }); } } - IndexValue::Tag { field, is_set } => { - if is_set { + IndexValue::Tag { field, value } => { + for item in value { batch.ops.push(Operation::Bitmap { - class: BitmapClass::Tag { - field, - value: ().into(), - }, + class: BitmapClass::Tag { field, value: item }, set, }); } @@ -311,8 +330,8 @@ fn merge_index( value: new_value, .. }, ) => { - let mut add_values = HashSet::new(); - let mut remove_values = HashSet::new(); + let mut add_values = AHashSet::new(); + let mut remove_values = AHashSet::new(); for current_value in old_value.as_ref() { remove_values.insert(current_value); @@ -336,27 +355,34 @@ fn merge_index( ( IndexValue::Tag { field, - is_set: was_set, + value: old_value, + }, + IndexValue::Tag { + value: new_value, .. }, - IndexValue::Tag { is_set, .. }, ) => { - if was_set { - batch.ops.push(Operation::Bitmap { - class: BitmapClass::Tag { - field, - value: ().into(), - }, - set: false, - }); + for old_tag in &old_value { + if !new_value.contains(old_tag) { + batch.ops.push(Operation::Bitmap { + class: BitmapClass::Tag { + field, + value: old_tag.clone(), + }, + set: false, + }); + } } - if is_set { - batch.ops.push(Operation::Bitmap { - class: BitmapClass::Tag { - field, - value: ().into(), - }, - set: true, - }); + + for new_tag in new_value { + if !old_value.contains(&new_tag) { + batch.ops.push(Operation::Bitmap { + class: BitmapClass::Tag { + field, + value: new_tag, + }, + set: true, + }); + } } } (IndexValue::Blob { value: old_hash }, IndexValue::Blob { value: new_hash }) => { diff --git a/crates/common/src/storage/mod.rs b/crates/common/src/storage/mod.rs index af726199..e7f4eec2 100644 --- a/crates/common/src/storage/mod.rs +++ b/crates/common/src/storage/mod.rs @@ -8,4 +8,3 @@ pub mod blob; pub mod folder; pub mod index; pub mod state; -pub mod tag; diff --git a/crates/dav/src/file/proppatch.rs b/crates/dav/src/file/proppatch.rs index 7c3e5992..166cfd9b 100644 --- a/crates/dav/src/file/proppatch.rs +++ b/crates/dav/src/file/proppatch.rs @@ -109,7 +109,7 @@ impl FilePropPatchRequestHandler for Server { .await?; // Deserialize - let node = node.into_deserialized().caused_by(trc::location!())?; + let node = node.to_deserialized().caused_by(trc::location!())?; let mut new_node = node.inner.clone(); // Remove properties diff --git a/crates/dav/src/file/update.rs b/crates/dav/src/file/update.rs index 6aed5591..9904e00d 100644 --- a/crates/dav/src/file/update.rs +++ b/crates/dav/src/file/update.rs @@ -135,9 +135,7 @@ impl FileUpdateRequestHandler for Server { // Build node let change_id = self.generate_snowflake_id().caused_by(trc::location!())?; - let node = node_archive - .into_deserialized() - .caused_by(trc::location!())?; + let node = node_archive.to_deserialized().caused_by(trc::location!())?; let mut new_node = node.inner.clone(); let new_file = new_node.file.as_mut().unwrap(); new_file.blob_hash = blob_hash; diff --git a/crates/email/src/mailbox/destroy.rs b/crates/email/src/mailbox/destroy.rs index 2820f476..bdfe8265 100644 --- a/crates/email/src/mailbox/destroy.rs +++ b/crates/email/src/mailbox/destroy.rs @@ -13,14 +13,16 @@ use jmap_proto::{ types::{acl::Acl, collection::Collection, id::Id, property::Property}, }; use store::{ - Serialize, SerializeInfallible, + SerializeInfallible, query::Filter, roaring::RoaringBitmap, - write::{AlignedBytes, Archive, Archiver, BatchBuilder, log::ChangeLogBuilder}, + write::{ + AlignedBytes, Archive, BatchBuilder, log::ChangeLogBuilder, serialize::rkyv_deserialize, + }, }; use trc::AddContext; -use crate::message::delete::EmailDeletion; +use crate::message::{delete::EmailDeletion, metadata::MessageData}; use super::*; @@ -100,80 +102,88 @@ impl MailboxDestroy for Server { // If the message is in multiple mailboxes, untag it from the current mailbox, // otherwise delete it. let mut destroy_ids = RoaringBitmap::new(); - for (message_id, mailbox_ids) in self + for (message_id, message_data_) in self .get_properties::, _>( account_id, Collection::Email, &message_ids, - Property::MailboxIds, + Property::Value, ) .await? { // Remove mailbox from list - let mut mailbox_ids = mailbox_ids - .into_deserialized::>() + let prev_message_data = message_data_ + .to_unarchived::() .caused_by(trc::location!())?; - let orig_len = mailbox_ids.inner.len(); - mailbox_ids.inner.retain(|id| id.mailbox_id != document_id); - if mailbox_ids.inner.len() == orig_len { + + if !prev_message_data + .inner + .mailboxes + .iter() + .any(|id| id.mailbox_id == document_id) + { continue; } - if !mailbox_ids.inner.is_empty() { - // Obtain threadId - if let Some(thread_id) = self - .get_property::( - account_id, - Collection::Email, - message_id, - Property::ThreadId, - ) - .await? - { - // Untag message from mailbox - let mut batch = BatchBuilder::new(); - batch - .with_account_id(account_id) - .with_collection(Collection::Email) - .update_document(message_id) - .assert_value(Property::MailboxIds, &mailbox_ids) - .set( - Property::MailboxIds, - Archiver::new(mailbox_ids.inner) - .serialize() - .caused_by(trc::location!())?, - ) - .untag(Property::MailboxIds, document_id); - match self.core.storage.data.write(batch.build()).await { - Ok(_) => changes.log_update( - Collection::Email, - Id::from_parts(thread_id, message_id), - ), - Err(err) if err.is_assertion_failure() => { - return Ok(Err(SetError::forbidden().with_description( - concat!( - "Another process modified a message in this mailbox ", - "while deleting it, please try again." - ), - ))); - } - Err(err) => { - return Err(err.caused_by(trc::location!())); - } - } - } else { - trc::event!( - Store(trc::StoreEvent::NotFound), - AccountId = account_id, - MessageId = message_id, - MailboxId = document_id, - Details = "Message does not have a threadId.", - CausedBy = trc::location!(), - ); - } - } else { + if prev_message_data.inner.mailboxes.len() == 1 { // Delete message destroy_ids.insert(message_id); + continue; + } + + let mut new_message_data = + rkyv_deserialize(prev_message_data.inner).caused_by(trc::location!())?; + + new_message_data + .mailboxes + .retain(|id| id.mailbox_id != document_id); + + // Obtain threadId + if let Some(thread_id) = self + .get_property::( + account_id, + Collection::Email, + message_id, + Property::ThreadId, + ) + .await? + { + // Untag message from mailbox + let mut batch = BatchBuilder::new(); + batch + .with_account_id(account_id) + .with_collection(Collection::Email) + .update_document(message_id) + .custom( + ObjectIndexBuilder::new() + .with_changes(new_message_data) + .with_current(prev_message_data), + ) + .caused_by(trc::location!())?; + match self.core.storage.data.write(batch.build()).await { + Ok(_) => changes.log_update( + Collection::Email, + Id::from_parts(thread_id, message_id), + ), + Err(err) if err.is_assertion_failure() => { + return Ok(Err(SetError::forbidden().with_description(concat!( + "Another process modified a message in this mailbox ", + "while deleting it, please try again." + )))); + } + Err(err) => { + return Err(err.caused_by(trc::location!())); + } + } + } else { + trc::event!( + Store(trc::StoreEvent::NotFound), + AccountId = account_id, + MessageId = message_id, + MailboxId = document_id, + Details = "Message does not have a threadId.", + CausedBy = trc::location!(), + ); } } diff --git a/crates/email/src/mailbox/index.rs b/crates/email/src/mailbox/index.rs index d022d627..42c062b0 100644 --- a/crates/email/src/mailbox/index.rs +++ b/crates/email/src/mailbox/index.rs @@ -12,9 +12,8 @@ use common::{ }, }; use jmap_proto::types::{property::Property, value::AclGrant}; -use store::write::{MaybeDynamicId, TagValue}; -use super::{ArchivedMailbox, ArchivedUidMailbox, Mailbox, UidMailbox}; +use super::{ArchivedMailbox, Mailbox}; impl IndexableObject for Mailbox { fn index_values(&self) -> impl Iterator> { @@ -29,7 +28,11 @@ impl IndexableObject for Mailbox { }, IndexValue::Tag { field: Property::Role.into(), - is_set: !matches!(self.role, SpecialUse::None), + value: if !matches!(self.role, SpecialUse::None) { + vec![().into()] + } else { + vec![] + }, }, IndexValue::U32 { field: Property::ParentId.into(), @@ -64,7 +67,11 @@ impl IndexableObject for &ArchivedMailbox { }, IndexValue::Tag { field: Property::Role.into(), - is_set: !matches!(self.role, ArchivedSpecialUse::None), + value: if !matches!(self.role, ArchivedSpecialUse::None) { + vec![().into()] + } else { + vec![] + }, }, IndexValue::U32 { field: Property::ParentId.into(), @@ -115,21 +122,3 @@ impl FolderHierarchy for ArchivedMailbox { 0 } } - -impl From<&UidMailbox> for TagValue { - fn from(value: &UidMailbox) -> Self { - TagValue::Id(MaybeDynamicId::Static(value.mailbox_id)) - } -} - -impl From for TagValue { - fn from(value: UidMailbox) -> Self { - TagValue::Id(MaybeDynamicId::Static(value.mailbox_id)) - } -} - -impl From<&ArchivedUidMailbox> for TagValue { - fn from(value: &ArchivedUidMailbox) -> Self { - TagValue::Id(MaybeDynamicId::Static(value.mailbox_id.into())) - } -} diff --git a/crates/email/src/mailbox/mod.rs b/crates/email/src/mailbox/mod.rs index 69b5e5ea..0930c0a9 100644 --- a/crates/email/src/mailbox/mod.rs +++ b/crates/email/src/mailbox/mod.rs @@ -45,18 +45,6 @@ impl SerializedVersion for Mailbox { } } -impl SerializedVersion for UidMailbox { - fn serialize_version() -> u8 { - 0 - } -} - -#[derive(Debug)] -pub struct ExpandPath<'x> { - pub path: Vec<&'x str>, - pub found_names: Vec<(String, u32, u32)>, -} - impl Mailbox { pub fn new(name: impl Into) -> Self { Mailbox { diff --git a/crates/email/src/message/copy.rs b/crates/email/src/message/copy.rs index 5ff46e64..a80f7db6 100644 --- a/crates/email/src/message/copy.rs +++ b/crates/email/src/message/copy.rs @@ -4,7 +4,7 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use common::{Server, auth::ResourceToken}; +use common::{Server, auth::ResourceToken, storage::index::ObjectIndexBuilder}; use jmap_proto::{ error::set::SetError, types::{ @@ -14,10 +14,9 @@ use jmap_proto::{ }; use mail_parser::parsers::fields::thread::thread_name; use store::{ - BlobClass, Serialize, SerializeInfallible, + BlobClass, write::{ - AlignedBytes, Archive, Archiver, BatchBuilder, MaybeDynamicId, TagValue, TaskQueueClass, - ValueClass, + AlignedBytes, Archive, BatchBuilder, MaybeDynamicId, TagValue, TaskQueueClass, ValueClass, log::{Changes, LogInsert}, }, }; @@ -28,7 +27,7 @@ use crate::mailbox::UidMailbox; use super::{ index::{MAX_ID_LENGTH, MAX_SORT_FIELD_LENGTH, TrimTextValue}, ingest::{EmailIngest, IngestedEmail, LogEmailInsert}, - metadata::{HeaderName, HeaderValue, MessageMetadata}, + metadata::{HeaderName, HeaderValue, MessageData, MessageMetadata}, }; pub trait EmailCopy: Sync + Send { @@ -182,21 +181,14 @@ impl EmailCopy for Server { .log(LogEmailInsert::new(thread_id)) .set(Property::ThreadId, maybe_thread_id) .tag(Property::ThreadId, TagValue::Id(maybe_thread_id)) - .tag_many(Property::MailboxIds, mailbox_ids.iter()) - .set( - Property::MailboxIds, - Archiver::new(mailbox_ids) - .serialize() - .caused_by(trc::location!())?, + .custom( + ObjectIndexBuilder::<(), _>::new().with_changes(MessageData { + mailboxes: mailbox_ids, + keywords, + change_id, + }), ) - .tag_many(Property::Keywords, keywords.iter()) - .set( - Property::Keywords, - Archiver::new(keywords) - .serialize() - .caused_by(trc::location!())?, - ) - .set(Property::Cid, change_id.serialize()) + .caused_by(trc::location!())? .set( ValueClass::TaskQueue(TaskQueueClass::IndexEmail { seq: self.generate_snowflake_id()?, diff --git a/crates/email/src/message/delete.rs b/crates/email/src/message/delete.rs index 9a5b8c9e..ea28a1ba 100644 --- a/crates/email/src/message/delete.rs +++ b/crates/email/src/message/delete.rs @@ -6,10 +6,9 @@ use std::time::Duration; -use common::{KV_LOCK_PURGE_ACCOUNT, Server}; +use common::{KV_LOCK_PURGE_ACCOUNT, Server, storage::index::ObjectIndexBuilder}; use jmap_proto::types::{ - collection::Collection, id::Id, keyword::Keyword, property::Property, state::StateChange, - type_state::DataType, + collection::Collection, id::Id, property::Property, state::StateChange, type_state::DataType, }; use store::{ BitmapKey, IterateParams, U32_LEN, ValueKey, @@ -28,6 +27,8 @@ use store::rand::prelude::SliceRandom; use crate::{mailbox::*, message::metadata::MessageMetadata}; +use super::metadata::MessageData; + pub trait EmailDeletion: Sync + Send { fn emails_tombstone( &self, @@ -63,24 +64,19 @@ impl EmailDeletion for Server { // Fetch mailboxes and threadIds let mut thread_ids: AHashMap = AHashMap::new(); - for (document_id, mailboxes) in self + for (document_id, data) in self .get_properties::, _>( account_id, Collection::Email, &document_ids, - Property::MailboxIds, + Property::Value, ) .await? { delete_properties.insert( document_id, DeleteProperties { - mailboxes: mailboxes - .unarchive::>() - .caused_by(trc::location!())? - .iter() - .map(|m| u32::from(m.mailbox_id)) - .collect(), + archive: Some(data), thread_id: None, }, ); @@ -151,17 +147,18 @@ impl EmailDeletion for Server { for (document_id, delete_properties) in delete_properties { batch.update_document(document_id); - if !delete_properties.mailboxes.is_empty() { - for mailbox_id in &delete_properties.mailboxes { - changes.log_child_update(Collection::Mailbox, *mailbox_id); + if let Some(data_) = delete_properties.archive { + let data = data_ + .to_unarchived::() + .caused_by(trc::location!())?; + + for mailbox in data.inner.mailboxes.iter() { + changes.log_child_update(Collection::Mailbox, u32::from(mailbox.mailbox_id)); } batch - .untag_many( - Property::MailboxIds, - delete_properties.mailboxes.iter().copied(), - ) - .clear(Property::MailboxIds); + .custom(ObjectIndexBuilder::<_, ()>::new().with_current(data)) + .caused_by(trc::location!())?; } else { trc::event!( Store(StoreEvent::NotFound), @@ -342,16 +339,16 @@ impl EmailDeletion for Server { // Find messages to destroy let mut destroy_ids = RoaringBitmap::new(); - for (document_id, cid) in self - .get_properties::( + for (document_id, data) in self + .get_properties::, _>( account_id, Collection::Email, &deletion_candidates, - Property::Cid, + Property::Value, ) .await? { - if cid < reference_cid { + if data.unarchive::()?.change_id < reference_cid { destroy_ids.insert(document_id); } } @@ -434,41 +431,12 @@ impl EmailDeletion for Server { .with_account_id(account_id) .with_collection(Collection::Email) .delete_document(document_id) - .clear(Property::Cid) + .clear(Property::Value) .untag( Property::MailboxIds, TagValue::Id(MaybeDynamicId::Static(TOMBSTONE_ID)), ); - // Remove keywords - if let Some(keywords_) = self - .core - .storage - .data - .get_value::>(ValueKey { - account_id, - collection: Collection::Email.into(), - document_id, - class: ValueClass::Property(Property::Keywords.into()), - }) - .await? - { - let keywords = keywords_ - .unarchive::>() - .caused_by(trc::location!())?; - batch - .untag_many(Property::Keywords, keywords.iter()) - .clear(Property::Keywords); - } else { - trc::event!( - Purge(trc::PurgeEvent::Error), - AccountId = account_id, - DocumentId = document_id, - Reason = "Failed to fetch keywords.", - CausedBy = trc::location!(), - ); - } - // Remove message metadata if let Some(metadata_) = self .core @@ -525,6 +493,6 @@ impl EmailDeletion for Server { #[derive(Default, Debug)] struct DeleteProperties { - mailboxes: Vec, + archive: Option>, thread_id: Option, } diff --git a/crates/email/src/message/index.rs b/crates/email/src/message/index.rs index 86a85d52..87bb45ee 100644 --- a/crates/email/src/message/index.rs +++ b/crates/email/src/message/index.rs @@ -4,7 +4,8 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use jmap_proto::types::{keyword::Keyword, property::Property}; +use common::storage::index::{IndexValue, IndexableObject, ObjectIndexBuilder}; +use jmap_proto::types::property::Property; use mail_parser::{ decoders::html::html_to_text, parsers::{fields::thread::thread_name, preview::preview_text}, @@ -15,18 +16,16 @@ use store::{ Serialize, SerializeInfallible, backend::MAX_TOKEN_LENGTH, fts::{Field, index::FtsDocument}, - write::{Archiver, BatchBuilder, BlobOp, DirectoryClass}, + write::{Archiver, BatchBuilder, BlobOp, DirectoryClass, MaybeDynamicId, TagValue}, }; use trc::AddContext; use utils::BlobHash; -use crate::mailbox::UidMailbox; - use super::metadata::{ Addr, Address, ArchivedAddress, ArchivedGetHeader, ArchivedHeaderName, ArchivedHeaderValue, - ArchivedMessageMetadata, ArchivedMessageMetadataContents, ArchivedMessageMetadataPart, - ArchivedMetadataPartType, DecodedPartContent, Group, HeaderName, HeaderValue, MessageMetadata, - MessageMetadataPart, + ArchivedMessageData, ArchivedMessageMetadata, ArchivedMessageMetadataContents, + ArchivedMessageMetadataPart, ArchivedMetadataPartType, DecodedPartContent, Group, HeaderName, + HeaderValue, MessageData, MessageMetadata, MessageMetadataPart, }; pub const MAX_MESSAGE_PARTS: usize = 1000; @@ -449,8 +448,7 @@ pub(super) trait IndexMessage { tenant_id: Option, message: mail_parser::Message<'_>, blob_hash: BlobHash, - keywords: Vec, - mailbox_ids: Vec, + data: MessageData, received_at: u64, ) -> trc::Result<&mut Self>; } @@ -462,26 +460,9 @@ impl IndexMessage for BatchBuilder { tenant_id: Option, message: mail_parser::Message<'_>, blob_hash: BlobHash, - keywords: Vec, - mailbox_ids: Vec, + data: MessageData, received_at: u64, ) -> trc::Result<&mut Self> { - // Index keywords - let keywords = Archiver::new(keywords); - self.set( - Property::Keywords, - keywords.serialize().caused_by(trc::location!())?, - ) - .tag_many(Property::Keywords, keywords.into_inner().into_iter()); - - // Index mailboxIds - self.tag_many(Property::MailboxIds, mailbox_ids.iter()).set( - Property::MailboxIds, - Archiver::new(mailbox_ids) - .serialize() - .caused_by(trc::location!())?, - ); - // Index size self.index( Property::Size, @@ -578,6 +559,10 @@ impl IndexMessage for BatchBuilder { Vec::new(), ); + // Store message data + self.custom(ObjectIndexBuilder::<(), _>::new().with_changes(data)) + .caused_by(trc::location!())?; + // Store message metadata self.set( Property::BodyStructure, @@ -590,6 +575,60 @@ impl IndexMessage for BatchBuilder { } } +impl IndexableObject for MessageData { + fn index_values(&self) -> impl Iterator> { + [ + IndexValue::Tag { + field: Property::MailboxIds.into(), + value: self + .mailboxes + .iter() + .map(|m| TagValue::Id(MaybeDynamicId::Static(m.mailbox_id))) + .collect(), + }, + IndexValue::Tag { + field: Property::Keywords.into(), + value: self + .keywords + .iter() + .map(|k| match k.id() { + Ok(id) => TagValue::Id(MaybeDynamicId::Static(id)), + Err(string) => TagValue::Text(string.into_bytes()), + }) + .collect(), + }, + ] + .into_iter() + } +} + +impl IndexableObject for &ArchivedMessageData { + fn index_values(&self) -> impl Iterator> { + [ + IndexValue::Tag { + field: Property::MailboxIds.into(), + value: self + .mailboxes + .iter() + .map(|m| TagValue::Id(MaybeDynamicId::Static(u32::from(m.mailbox_id)))) + .collect(), + }, + IndexValue::Tag { + field: Property::Keywords.into(), + value: self + .keywords + .iter() + .map(|k| match k.id() { + Ok(id) => TagValue::Id(MaybeDynamicId::Static(id)), + Err(string) => TagValue::Text(string.into_bytes()), + }) + .collect(), + }, + ] + .into_iter() + } +} + pub trait IndexMessageText<'x>: Sized { fn index_message(self, message: &'x ArchivedMessageMetadata, raw_message: &'x [u8]) -> Self; } @@ -781,24 +820,6 @@ impl Default for SortedAddressBuilder { } } -/*impl MessageMetadataPart { - fn language(&self) -> Option { - self.headers - .header_value(&HeaderName::ContentLanguage) - .and_then(|v| { - Language::from_iso_639(match v { - HeaderValue::Text(v) => v.as_ref(), - HeaderValue::TextList(v) => v.first()?, - _ => { - return None; - } - }) - .unwrap_or(Language::Unknown) - .into() - }) - } -}*/ - impl ArchivedMessageMetadataPart { fn language(&self) -> Option { self.headers diff --git a/crates/email/src/message/ingest.rs b/crates/email/src/message/ingest.rs index 7c888eef..894830f7 100644 --- a/crates/email/src/message/ingest.rs +++ b/crates/email/src/message/ingest.rs @@ -52,6 +52,7 @@ use crate::{ message::{ crypto::EncryptionParams, index::{IndexMessage, MAX_ID_LENGTH, VisitValues}, + metadata::MessageData, }, thread::cache::ThreadCache, }; @@ -499,12 +500,14 @@ impl EmailIngest for Server { tenant_id, message, blob_id.hash.clone(), - params.keywords, - mailbox_ids, + MessageData { + mailboxes: mailbox_ids, + keywords: params.keywords, + change_id, + }, params.received_at.unwrap_or_else(now), ) .caused_by(trc::location!())? - .set(Property::Cid, change_id.serialize()) .set(Property::ThreadId, maybe_thread_id) .tag(Property::ThreadId, TagValue::Id(maybe_thread_id)) .set( diff --git a/crates/email/src/message/metadata.rs b/crates/email/src/message/metadata.rs index ce2fa7ec..4d37a16f 100644 --- a/crates/email/src/message/metadata.rs +++ b/crates/email/src/message/metadata.rs @@ -6,6 +6,8 @@ use std::{borrow::Cow, collections::VecDeque, fmt::Display}; +use common::storage::index::IndexableAndSerializableObject; +use jmap_proto::types::keyword::{ArchivedKeyword, Keyword}; use mail_parser::{ PartType, decoders::{ @@ -21,6 +23,15 @@ use rkyv::{ use store::SerializedVersion; use utils::BlobHash; +use crate::mailbox::{ArchivedUidMailbox, UidMailbox}; + +#[derive(rkyv::Serialize, rkyv::Deserialize, rkyv::Archive, Debug)] +pub struct MessageData { + pub mailboxes: Vec, + pub keywords: Vec, + pub change_id: u64, +} + #[derive(rkyv::Serialize, rkyv::Deserialize, rkyv::Archive, Debug)] pub struct MessageMetadata { pub contents: Vec, @@ -32,6 +43,14 @@ pub struct MessageMetadata { pub raw_headers: Vec, } +impl IndexableAndSerializableObject for MessageData {} + +impl SerializedVersion for MessageData { + fn serialize_version() -> u8 { + 0 + } +} + impl SerializedVersion for MessageMetadata { fn serialize_version() -> u8 { 0 @@ -1227,3 +1246,112 @@ impl From<&CompactDateTime> for mail_parser::DateTime { } } } + +impl MessageData { + pub fn has_keyword(&self, keyword: &Keyword) -> bool { + self.keywords.iter().any(|k| k == keyword) + } + + pub fn set_keywords(&mut self, keywords: Vec) { + self.keywords = keywords; + } + + pub fn add_keyword(&mut self, keyword: Keyword) -> bool { + if !self.keywords.contains(&keyword) { + self.keywords.push(keyword); + true + } else { + false + } + } + + pub fn remove_keyword(&mut self, keyword: &Keyword) -> bool { + let prev_len = self.keywords.len(); + self.keywords.retain(|k| k != keyword); + self.keywords.len() != prev_len + } + + pub fn set_mailboxes(&mut self, mailboxes: Vec) { + self.mailboxes = mailboxes; + } + + pub fn add_mailbox(&mut self, mailbox: UidMailbox) { + if !self.mailboxes.contains(&mailbox) { + self.mailboxes.push(mailbox); + } + } + + pub fn remove_mailbox(&mut self, mailbox: u32) { + self.mailboxes.retain(|m| m.mailbox_id != mailbox); + } + + pub fn has_keyword_changes(&self, prev_data: &ArchivedMessageData) -> bool { + self.keywords.len() != prev_data.keywords.len() + || !self + .keywords + .iter() + .all(|k| prev_data.keywords.iter().any(|pk| pk == k)) + } + + pub fn has_mailbox_id(&self, mailbox_id: u32) -> bool { + self.mailboxes.iter().any(|m| m.mailbox_id == mailbox_id) + } + + pub fn added_keywords( + &self, + prev_data: &ArchivedMessageData, + ) -> impl Iterator { + self.keywords + .iter() + .filter(|k| prev_data.keywords.iter().all(|pk| pk != *k)) + } + + pub fn removed_keywords<'x>( + &'x self, + prev_data: &'x ArchivedMessageData, + ) -> impl Iterator { + prev_data + .keywords + .iter() + .filter(|k| self.keywords.iter().all(|pk| pk != *k)) + } + + pub fn added_mailboxes( + &self, + prev_data: &ArchivedMessageData, + ) -> impl Iterator { + self.mailboxes.iter().filter(|m| { + prev_data + .mailboxes + .iter() + .all(|pm| pm.mailbox_id != m.mailbox_id) + }) + } + + pub fn removed_mailboxes<'x>( + &'x self, + prev_data: &'x ArchivedMessageData, + ) -> impl Iterator { + prev_data.mailboxes.iter().filter(|m| { + self.mailboxes + .iter() + .all(|pm| pm.mailbox_id != m.mailbox_id) + }) + } + + pub fn has_mailbox_changes(&self, prev_data: &ArchivedMessageData) -> bool { + self.mailboxes.len() != prev_data.mailboxes.len() + || !self.mailboxes.iter().all(|m| { + prev_data + .mailboxes + .iter() + .any(|pm| pm.mailbox_id == m.mailbox_id) + }) + } +} + +impl ArchivedMessageData { + pub fn has_mailbox_id(&self, mailbox_id: u32) -> bool { + self.mailboxes.iter().any(|m| m.mailbox_id == mailbox_id) + } +} diff --git a/crates/email/src/sieve/ingest.rs b/crates/email/src/sieve/ingest.rs index f153c18b..39a72483 100644 --- a/crates/email/src/sieve/ingest.rs +++ b/crates/email/src/sieve/ingest.rs @@ -4,7 +4,7 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use std::{borrow::Cow, collections::HashSet, sync::Arc}; +use std::{borrow::Cow, sync::Arc}; use crate::{ mailbox::{INBOX_ID, TRASH_ID, manage::MailboxFnc}, @@ -22,16 +22,17 @@ use mail_parser::MessageParser; use sieve::{Envelope, Event, Input, Mailbox, Recipient, Sieve}; use store::{ Deserialize, Serialize, SerializeInfallible, - ahash::{AHashSet, RandomState}, + ahash::AHashMap, + dispatch::lookup::KeyValue, query::Filter, - write::{AlignedBytes, Archive, Archiver, BatchBuilder, BlobOp, LegacyBincode, now}, + write::{AlignedBytes, Archive, Archiver, BatchBuilder, BlobOp, LegacyBincode}, }; use trc::{AddContext, SieveEvent}; use utils::config::utils::ParseValue; use std::future::Future; -use super::{ActiveScript, SeenIdHash, SeenIds, SieveScript}; +use super::{ActiveScript, SeenIdHash, SieveScript}; struct SieveMessage<'x> { pub raw_message: Cow<'x, [u8]>, @@ -67,7 +68,7 @@ pub trait SieveScriptIngest: Sync + Send { &self, account_id: u32, document_id: u32, - ) -> impl Future> + Send; + ) -> impl Future> + Send; } impl SieveScriptIngest for Server { @@ -132,14 +133,12 @@ impl SieveScriptIngest for Server { let mut do_discard = false; let mut do_deliver = false; - let mut new_ids = AHashSet::new(); let mut reject_reason = None; let mut messages: Vec = vec![SieveMessage { raw_message: raw_message.into(), file_into: Vec::new(), flags: Vec::new(), }]; - let now = now(); let mut ingested_message = IngestedEmail { id: Id::default(), change_id: u64::MAX, @@ -147,7 +146,7 @@ impl SieveScriptIngest for Server { size: raw_message.len(), imap_uids: Vec::new(), }; - let mut seen_ids = active_script.seen_ids; + let mut checked_ids: AHashMap = AHashMap::new(); while let Some(event) = instance.run(input) { match event { @@ -251,13 +250,27 @@ impl SieveScriptIngest for Server { } } Event::DuplicateId { id, expiry, last } => { - let id_hash = SeenIdHash::new(&id, expiry + now); - let seen_id = seen_ids.ids.contains(&id_hash); - if !seen_id || last { - new_ids.insert(id_hash); - } + let id_hash = SeenIdHash::new(account_id, active_script.hash, &id); + if let Some(result) = checked_ids.get(&id_hash) { + input = (*result).into(); + } else { + let exists = self + .in_memory_store() + .key_get::<()>(id_hash.key()) + .await + .caused_by(trc::location!())? + .is_some(); - input = seen_id.into(); + if !exists || last { + self.in_memory_store() + .key_set(KeyValue::new(id_hash.key(), vec![]).expires(expiry)) + .await + .caused_by(trc::location!())?; + } + + checked_ids.insert(id_hash, exists); + input = exists.into(); + } } Event::Discard => { do_discard = true; @@ -524,25 +537,6 @@ impl SieveScriptIngest for Server { } } - // Save new ids script changes - if !new_ids.is_empty() || seen_ids.has_changes { - seen_ids.ids.extend(new_ids); - let mut batch = BatchBuilder::new(); - batch - .with_account_id(account_id) - .with_collection(Collection::SieveScript) - .update_document(active_script.document_id) - .set( - Property::EmailIds, - Archiver::new(seen_ids.ids) - .serialize() - .caused_by(trc::location!())?, - ); - if let Err(err) = self.store().write(batch).await.caused_by(trc::location!()) { - trc::error!(err.details("Failed to save Sieve seen ids changes.")); - } - } - if let Some(reject_reason) = reject_reason { Err( trc::EventType::MessageIngest(trc::MessageIngestEvent::Error) @@ -572,30 +566,13 @@ impl SieveScriptIngest for Server { .results .min() { - let (script, script_name) = self.sieve_script_compile(account_id, document_id).await?; - - let seen_ids = if let Some(seen_ids_archive_) = self - .get_property::>( - account_id, - Collection::SieveScript, - document_id, - Property::EmailIds, - ) - .await? - { - let seen_ids_archive = seen_ids_archive_ - .unarchive::>() - .caused_by(trc::location!())?; - SeenIds::from(seen_ids_archive) - } else { - SeenIds::default() - }; + let script = self.sieve_script_compile(account_id, document_id).await?; Ok(Some(ActiveScript { document_id, - script: Arc::new(script), - script_name, - seen_ids, + script: Arc::new(script.script), + script_name: script.name, + hash: script.hash, })) } else { Ok(None) @@ -622,7 +599,7 @@ impl SieveScriptIngest for Server { { self.sieve_script_compile(account_id, document_id) .await - .map(|(sieve, _)| Some(sieve)) + .map(|script| Some(script.script)) } else { Ok(None) } @@ -633,7 +610,7 @@ impl SieveScriptIngest for Server { &self, account_id: u32, document_id: u32, - ) -> trc::Result<(Sieve, String)> { + ) -> trc::Result { // Obtain script object let script_object = self .get_property::>( @@ -651,6 +628,7 @@ impl SieveScriptIngest for Server { })?; // Obtain the sieve script length + let hash = script_object.hash; let unarchived_script = script_object .unarchive::() .caused_by(trc::location!())?; @@ -676,7 +654,11 @@ impl SieveScriptIngest for Server { .get(script_offset..) .and_then(|bytes| LegacyBincode::::deserialize(bytes).ok()) { - Ok((sieve.inner, unarchived_script.name.to_string())) + Ok(CompiledScript { + script: sieve.inner, + name: unarchived_script.name.to_string(), + hash, + }) } else { // Deserialization failed, probably because the script compiler version changed match self.core.sieve.untrusted_compiler.compile( @@ -730,7 +712,11 @@ impl SieveScriptIngest for Server { .await .caused_by(trc::location!())?; - Ok((sieve.inner, new_archive.into_inner().name)) + Ok(CompiledScript { + script: sieve.inner, + name: new_archive.into_inner().name, + hash, + }) } Err(error) => Err(trc::StoreEvent::UnexpectedError .caused_by(trc::location!()) @@ -740,3 +726,9 @@ impl SieveScriptIngest for Server { } } } + +pub struct CompiledScript { + pub script: Sieve, + pub name: String, + pub hash: u32, +} diff --git a/crates/email/src/sieve/mod.rs b/crates/email/src/sieve/mod.rs index b4c6ed72..1455f0a2 100644 --- a/crates/email/src/sieve/mod.rs +++ b/crates/email/src/sieve/mod.rs @@ -4,43 +4,24 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use std::{collections::HashSet, sync::Arc}; +use std::sync::Arc; -use rkyv::collections::swiss_table::ArchivedHashSet; +use common::KV_SIEVE_ID; use sieve::Sieve; -use store::{SerializedVersion, ahash::RandomState, blake3, write::now}; +use store::{SerializedVersion, blake3}; use utils::BlobHash; pub mod activate; pub mod delete; pub mod index; pub mod ingest; -pub mod serialize; #[derive(Debug, Clone)] pub struct ActiveScript { pub document_id: u32, + pub hash: u32, pub script_name: String, pub script: Arc, - pub seen_ids: SeenIds, -} - -#[derive(rkyv::Archive, rkyv::Deserialize, rkyv::Serialize, Debug, Clone)] -pub struct SeenIdHash { - hash: [u8; 32], - expiry: u64, -} - -#[derive(Default, Debug, Clone, PartialEq, Eq)] -pub struct SeenIds { - pub ids: HashSet, - pub has_changes: bool, -} - -impl SerializedVersion for SeenIdHash { - fn serialize_version() -> u8 { - 0 - } } #[derive( @@ -114,77 +95,29 @@ impl SieveScript { } } -impl From<&ArchivedHashSet> for SeenIds { - fn from(archived: &ArchivedHashSet) -> Self { - let mut seen = SeenIds { - ids: HashSet::with_capacity_and_hasher(archived.len(), RandomState::new()), - has_changes: false, - }; - - let now = now(); - - for hash in archived.iter() { - if hash.expiry > now { - seen.ids.insert(SeenIdHash { - hash: hash.hash, - expiry: hash.expiry.into(), - }); - } else { - seen.has_changes = true; - } - } - - seen - } -} +#[derive(Debug, Default, Clone, PartialEq, Eq, Hash)] +#[repr(transparent)] +pub struct SeenIdHash(pub [u8; 32]); impl SeenIdHash { - pub fn new(id: &str, expiry: u64) -> Self { + pub fn new(account_id: u32, hash: u32, id: &str) -> Self { let mut hasher = blake3::Hasher::new(); + hasher.update(&account_id.to_be_bytes()); + hasher.update(&hash.to_be_bytes()); hasher.update(id.as_bytes()); - SeenIdHash { - hash: hasher.finalize().into(), - expiry, - } + SeenIdHash(hasher.finalize().into()) + } + + pub fn key(&self) -> Vec { + let mut result = Vec::with_capacity(self.0.len() + 1); + result.push(KV_SIEVE_ID); + result.extend_from_slice(&self.0); + result } } -impl PartialOrd for SeenIdHash { - fn partial_cmp(&self, other: &Self) -> Option { - Some(self.cmp(other)) +impl AsRef<[u8]> for SeenIdHash { + fn as_ref(&self) -> &[u8] { + &self.0 } } - -impl Ord for SeenIdHash { - fn cmp(&self, other: &Self) -> std::cmp::Ordering { - self.expiry.cmp(&other.expiry) - } -} - -impl std::hash::Hash for SeenIdHash { - fn hash(&self, state: &mut H) { - self.hash.hash(state); - } -} - -impl PartialEq for SeenIdHash { - fn eq(&self, other: &Self) -> bool { - self.hash == other.hash - } -} - -impl Eq for SeenIdHash {} - -impl std::hash::Hash for ArchivedSeenIdHash { - fn hash(&self, state: &mut H) { - self.hash.hash(state); - } -} - -impl PartialEq for ArchivedSeenIdHash { - fn eq(&self, other: &Self) -> bool { - self.hash == other.hash - } -} - -impl Eq for ArchivedSeenIdHash {} diff --git a/crates/email/src/sieve/serialize.rs b/crates/email/src/sieve/serialize.rs deleted file mode 100644 index d9f1a64e..00000000 --- a/crates/email/src/sieve/serialize.rs +++ /dev/null @@ -1,79 +0,0 @@ -/* - * SPDX-FileCopyrightText: 2020 Stalwart Labs Ltd - * - * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL - */ - -use std::collections::HashSet; - -use serde::ser::SerializeSeq; -use store::{ahash::RandomState, write::now}; - -use super::{SeenIdHash, SeenIds}; - -// SeenIds serializer -impl serde::Serialize for SeenIds { - fn serialize(&self, serializer: S) -> Result - where - S: serde::Serializer, - { - let mut seq = serializer.serialize_seq((self.ids.len() * 2).into())?; - for id in &self.ids { - seq.serialize_element(&id.expiry)?; - seq.serialize_element(&id.hash)?; - } - - seq.end() - } -} - -impl<'de> serde::Deserialize<'de> for SeenIds { - fn deserialize(deserializer: D) -> Result - where - D: serde::Deserializer<'de>, - { - deserializer.deserialize_seq(SeenIdsVisitor) - } -} - -struct SeenIdsVisitor; - -impl<'de> serde::de::Visitor<'de> for SeenIdsVisitor { - type Value = SeenIds; - - fn expecting(&self, formatter: &mut std::fmt::Formatter) -> std::fmt::Result { - formatter.write_str("invalid SeenIds") - } - - fn visit_seq(self, mut seq: A) -> Result - where - A: serde::de::SeqAccess<'de>, - { - let num_entries = seq.size_hint().unwrap_or(0) / 2; - let mut seen_ids = SeenIds { - ids: HashSet::with_capacity_and_hasher(num_entries, RandomState::new()), - has_changes: false, - }; - let now = now(); - - for _ in 0..num_entries { - let expiry = seq - .next_element::()? - .ok_or_else(|| serde::de::Error::custom("Expected expiry."))?; - if expiry > now { - seen_ids.ids.insert(SeenIdHash { - hash: seq - .next_element()? - .ok_or_else(|| serde::de::Error::custom("Expected hash."))?, - expiry, - }); - } else { - seq.next_element::<[u8; 32]>()? - .ok_or_else(|| serde::de::Error::custom("Expected hash."))?; - seen_ids.has_changes = true; - } - } - - Ok(seen_ids) - } -} diff --git a/crates/http/src/management/stores.rs b/crates/http/src/management/stores.rs index 4d7bc2fd..510c7539 100644 --- a/crates/http/src/management/stores.rs +++ b/crates/http/src/management/stores.rs @@ -16,7 +16,7 @@ use directory::{ Permission, backend::internal::manage::{self, ManageDirectory}, }; -use email::{mailbox::UidMailbox, message::ingest::EmailIngest}; +use email::message::{ingest::EmailIngest, metadata::MessageData}; use hyper::Method; use jmap_proto::types::{collection::Collection, property::Property}; use serde_json::json; @@ -372,23 +372,28 @@ pub async fn reset_imap_uids(server: &Server, account_id: u32) -> trc::Result<(u .caused_by(trc::location!())? .unwrap_or_default() { - let uids = server + let data = server .get_property::>( account_id, Collection::Email, message_id, - Property::MailboxIds, + Property::Value, ) .await .caused_by(trc::location!())?; - let mut uids = if let Some(uids) = uids { - uids.into_deserialized::>() - .caused_by(trc::location!())? + let data_ = if let Some(data) = data { + data } else { continue; }; + let data = data_ + .to_unarchived::() + .caused_by(trc::location!())?; + let mut new_data = data + .deserialize::() + .caused_by(trc::location!())?; - for uid_mailbox in &mut uids.inner { + for uid_mailbox in &mut new_data.mailboxes { uid_mailbox.uid = server .assign_imap_uid(account_id, uid_mailbox.mailbox_id) .await @@ -401,10 +406,10 @@ pub async fn reset_imap_uids(server: &Server, account_id: u32) -> trc::Result<(u .with_account_id(account_id) .with_collection(Collection::Email) .update_document(message_id) - .assert_value(ValueClass::Property(Property::MailboxIds.into()), &uids) + .assert_value(ValueClass::Property(Property::Value.into()), &data) .set( - Property::MailboxIds, - Archiver::new(uids.inner) + Property::Value, + Archiver::new(new_data) .serialize() .caused_by(trc::location!())?, ); diff --git a/crates/imap/src/core/message.rs b/crates/imap/src/core/message.rs index 4e70eb54..0a94ee48 100644 --- a/crates/imap/src/core/message.rs +++ b/crates/imap/src/core/message.rs @@ -8,7 +8,7 @@ use std::{collections::BTreeMap, sync::Arc}; use ahash::AHashMap; use common::{NextMailboxState, listener::SessionStream}; -use email::mailbox::UidMailbox; +use email::message::metadata::MessageData; use imap_proto::protocol::{Sequence, expunge, select::Exists}; use jmap_proto::types::{collection::Collection, property::Property}; use store::write::{AlignedBytes, Archive}; @@ -50,22 +50,23 @@ impl SessionData { // Obtain all message ids let mut uid_map = BTreeMap::new(); - for (message_id, uid_mailbox_) in self + for (message_id, message_data_) in self .server .get_properties::, _>( mailbox.account_id, Collection::Email, &message_ids, - Property::MailboxIds, + Property::Value, ) .await? .into_iter() { - let uid_mailbox = uid_mailbox_ - .unarchive::>() + let message_data = message_data_ + .unarchive::() .caused_by(trc::location!())?; // Make sure the message is still in this mailbox - if let Some(item) = uid_mailbox + if let Some(item) = message_data + .mailboxes .iter() .find(|item| item.mailbox_id == mailbox.mailbox_id) { diff --git a/crates/imap/src/op/copy_move.rs b/crates/imap/src/op/copy_move.rs index 87821412..c2e1da99 100644 --- a/crates/imap/src/op/copy_move.rs +++ b/crates/imap/src/op/copy_move.rs @@ -9,19 +9,20 @@ use std::{sync::Arc, time::Instant}; use directory::Permission; use email::{ mailbox::{JUNK_ID, UidMailbox}, - message::{bayes::EmailBayesTrain, copy::EmailCopy, ingest::EmailIngest}, + message::{ + bayes::EmailBayesTrain, copy::EmailCopy, ingest::EmailIngest, metadata::MessageData, + }, }; use imap_proto::{ Command, ResponseCode, ResponseType, StatusResponse, protocol::copy_move::Arguments, receiver::Request, }; -use trc::AddContext; use crate::{ core::{SelectedMailbox, Session, SessionData}, spawn_op, }; -use common::{MailboxId, listener::SessionStream, storage::tag::TagManager}; +use common::{MailboxId, listener::SessionStream, storage::index::ObjectIndexBuilder}; use jmap_proto::{ error::set::SetErrorType, types::{ @@ -30,7 +31,6 @@ use jmap_proto::{ }, }; use store::{ - SerializeInfallible, roaring::RoaringBitmap, write::{AlignedBytes, Archive, BatchBuilder, ValueClass, log::ChangeLogBuilder}, }; @@ -189,7 +189,7 @@ impl SessionData { for (id, imap_id) in ids { // Obtain mailbox tags - let (mut mailboxes, thread_id) = if let Some(result) = self + let (data_, thread_id) = if let Some(result) = self .get_mailbox_tags(account_id, id) .await .imap_ctx(&arguments.tag, trc::location!())? @@ -199,23 +199,46 @@ impl SessionData { continue; }; + // Deserialize + let data = data_ + .to_unarchived::() + .imap_ctx(&arguments.tag, trc::location!())?; + // Make sure the message still belongs to this mailbox - if !mailboxes - .current() - .contains(&UidMailbox::new_unassigned(src_mailbox.id.mailbox_id)) - || mailboxes.current().contains(&dest_mailbox_id) + if !data + .inner + .mailboxes + .iter() + .any(|mailbox| mailbox.mailbox_id == src_mailbox.id.mailbox_id) + || data + .inner + .mailboxes + .iter() + .any(|mailbox| mailbox.mailbox_id == dest_mailbox_id.mailbox_id) { continue; } + // Prepare changes + let mut new_data = data + .deserialize() + .imap_ctx(&arguments.tag, trc::location!())?; + if changelog.change_id == u64::MAX { + changelog.change_id = self + .server + .assign_change_id(account_id) + .imap_ctx(&arguments.tag, trc::location!())?; + } + new_data.change_id = changelog.change_id; + // Add destination folder - mailboxes.update(dest_mailbox_id, true); + new_data.add_mailbox(dest_mailbox_id); if is_move { - mailboxes.update(UidMailbox::new_unassigned(src_mailbox.id.mailbox_id), false); + new_data.remove_mailbox(src_mailbox.id.mailbox_id); } // Assign IMAP UIDs - for uid_mailbox in mailboxes.inner_tags_mut() { + for uid_mailbox in &mut new_data.mailboxes { if uid_mailbox.uid == 0 { let assigned_uid = self .server @@ -233,17 +256,13 @@ impl SessionData { batch .with_account_id(account_id) .with_collection(Collection::Email) - .update_document(id); - mailboxes - .update_batch(&mut batch, Property::MailboxIds) + .update_document(id) + .custom( + ObjectIndexBuilder::new() + .with_current(data) + .with_changes(new_data), + ) .imap_ctx(&arguments.tag, trc::location!())?; - if changelog.change_id == u64::MAX { - changelog.change_id = self - .server - .assign_change_id(account_id) - .imap_ctx(&arguments.tag, trc::location!())?; - } - batch.set(Property::Cid, changelog.change_id.serialize()); // Add bayes train task if can_spam_train { @@ -470,7 +489,7 @@ impl SessionData { &self, account_id: u32, id: u32, - ) -> trc::Result, u32)>> { + ) -> trc::Result, u32)>> { // Obtain mailbox tags if let (Some(mailboxes), Some(thread_id)) = ( self.server @@ -478,21 +497,14 @@ impl SessionData { account_id, Collection::Email, id, - Property::MailboxIds, + Property::Value, ) .await?, self.server .get_property::(account_id, Collection::Email, id, Property::ThreadId) .await?, ) { - Ok(Some(( - TagManager::new( - mailboxes - .into_deserialized::>() - .caused_by(trc::location!())?, - ), - thread_id, - ))) + Ok(Some((mailboxes, thread_id))) } else { trc::event!( Store(trc::StoreEvent::NotFound), diff --git a/crates/imap/src/op/expunge.rs b/crates/imap/src/op/expunge.rs index c6a565a1..5f9fe3c6 100644 --- a/crates/imap/src/op/expunge.rs +++ b/crates/imap/src/op/expunge.rs @@ -8,7 +8,7 @@ use std::{sync::Arc, time::Instant}; use ahash::AHashMap; use directory::Permission; -use email::{mailbox::UidMailbox, message::delete::EmailDeletion}; +use email::message::{delete::EmailDeletion, metadata::MessageData}; use imap_proto::{ Command, ResponseCode, ResponseType, StatusResponse, parser::parse_sequence_set, @@ -17,13 +17,12 @@ use imap_proto::{ use trc::AddContext; use crate::core::{SavedSearch, SelectedMailbox, Session, SessionData}; -use common::{ImapId, listener::SessionStream, storage::tag::TagManager}; +use common::{ImapId, listener::SessionStream, storage::index::ObjectIndexBuilder}; use jmap_proto::types::{ acl::Acl, collection::Collection, id::Id, keyword::Keyword, property::Property, state::StateChange, type_state::DataType, }; use store::{ - SerializeInfallible, roaring::RoaringBitmap, write::{AlignedBytes, Archive, BatchBuilder, log::ChangeLogBuilder}, }; @@ -189,100 +188,80 @@ impl SessionData { deleted_ids: &RoaringBitmap, changelog: &mut ChangeLogBuilder, ) -> trc::Result<()> { - let mailbox_id = UidMailbox::new_unassigned(mailbox_id); let mut destroy_ids = RoaringBitmap::new(); - for (id, mailbox_ids) in self + for (id, data_) in self .server .get_properties::, _>( account_id, Collection::Email, deleted_ids, - Property::MailboxIds, + Property::Value, ) .await .caused_by(trc::location!())? { - let mut mailboxes = TagManager::new( - mailbox_ids - .into_deserialized::>() - .caused_by(trc::location!())?, - ); + let data = data_ + .to_unarchived::() + .caused_by(trc::location!())?; - if mailboxes.current().contains(&mailbox_id) { - if mailboxes.current().len() > 1 { - // Remove deleted flag - let (mut keywords, thread_id) = if let (Some(keywords), Some(thread_id)) = ( - self.server - .get_property::>( - account_id, - Collection::Email, - id, - Property::Keywords, - ) - .await - .caused_by(trc::location!())?, - self.server - .get_property::( - account_id, - Collection::Email, - id, - Property::ThreadId, - ) - .await - .caused_by(trc::location!())?, - ) { - ( - TagManager::new( - keywords - .into_deserialized::>() - .caused_by(trc::location!())?, - ), - thread_id, - ) - } else { - continue; - }; + if !data.inner.has_mailbox_id(mailbox_id) { + continue; + } else if data.inner.mailboxes.len() == 1 { + destroy_ids.insert(id); + continue; + } - // Untag message from this mailbox and remove Deleted flag - mailboxes.update(mailbox_id, false); - keywords.update(Keyword::Deleted, false); + // Remove deleted flag + let thread_id = if let Some(thread_id) = self + .server + .get_property::(account_id, Collection::Email, id, Property::ThreadId) + .await + .caused_by(trc::location!())? + { + thread_id + } else { + continue; + }; - // Write changes - let mut batch = BatchBuilder::new(); - batch - .with_account_id(account_id) - .with_collection(Collection::Email) - .update_document(id); - mailboxes - .update_batch(&mut batch, Property::MailboxIds) - .caused_by(trc::location!())?; - keywords - .update_batch(&mut batch, Property::Keywords) - .caused_by(trc::location!())?; - if changelog.change_id == u64::MAX { - changelog.change_id = self.server.assign_change_id(account_id)? + // Prepare changes + let mut new_data = data.deserialize().caused_by(trc::location!())?; + if changelog.change_id == u64::MAX { + changelog.change_id = self.server.assign_change_id(account_id)? + } + new_data.change_id = changelog.change_id; + + // Untag message from this mailbox and remove Deleted flag + new_data.remove_mailbox(mailbox_id); + new_data.remove_keyword(&Keyword::Deleted); + + // Write changes + let mut batch = BatchBuilder::new(); + batch + .with_account_id(account_id) + .with_collection(Collection::Email) + .update_document(id) + .custom( + ObjectIndexBuilder::new() + .with_current(data) + .with_changes(new_data), + ) + .caused_by(trc::location!())?; + match self + .server + .store() + .write(batch) + .await + .caused_by(trc::location!()) + { + Ok(_) => { + changelog.log_update(Collection::Email, Id::from_parts(thread_id, id)); + changelog.log_child_update(Collection::Mailbox, mailbox_id); + } + Err(err) => { + if !err.is_assertion_failure() { + return Err(err.caused_by(trc::location!())); } - batch.set(Property::Cid, changelog.change_id.serialize()); - match self - .server - .store() - .write(batch) - .await - .caused_by(trc::location!()) - { - Ok(_) => { - changelog.log_update(Collection::Email, Id::from_parts(thread_id, id)); - changelog.log_child_update(Collection::Mailbox, mailbox_id.mailbox_id); - } - Err(err) => { - if !err.is_assertion_failure() { - return Err(err.caused_by(trc::location!())); - } - } - } - } else { - destroy_ids.insert(id); } } } diff --git a/crates/imap/src/op/fetch.rs b/crates/imap/src/op/fetch.rs index b9f3c536..2e4c0037 100644 --- a/crates/imap/src/op/fetch.rs +++ b/crates/imap/src/op/fetch.rs @@ -11,12 +11,12 @@ use crate::{ spawn_op, }; use ahash::AHashMap; -use common::listener::SessionStream; +use common::{listener::SessionStream, storage::index::ObjectIndexBuilder}; use directory::Permission; use email::message::metadata::{ ArchivedAddress, ArchivedGetHeader, ArchivedHeaderName, ArchivedHeaderValue, ArchivedMessageMetadata, ArchivedMessageMetadataContents, ArchivedMetadataPartType, - DecodedParts, MessageMetadata, + DecodedParts, MessageData, MessageMetadata, }; use imap_proto::{ Command, ResponseCode, ResponseType, StatusResponse, @@ -41,10 +41,9 @@ use jmap_proto::types::{ type_state::DataType, }; use store::{ - Serialize, SerializeInfallible, query::log::{Change, Query}, rkyv::rend::u16_le, - write::{AlignedBytes, Archive, Archiver, BatchBuilder, serialize::rkyv_deserialize}, + write::{AlignedBytes, Archive, BatchBuilder, log::ChangeLogBuilder}, }; use super::{FromModSeq, ImapContext}; @@ -293,7 +292,7 @@ impl SessionData { } } - let mut set_seen_ids = Vec::new(); + let mut update_batches = Vec::new(); // Process each message let mut ids = ids @@ -305,10 +304,14 @@ impl SessionData { .iter() .map(|id| trc::Value::from(id.2)) .collect::>(); + let change_id = self + .server + .generate_snowflake_id() + .imap_ctx(&arguments.tag, trc::location!())?; for (seqnum, uid, id) in ids { // Obtain attributes and keywords - let (metadata_, keywords_) = if let (Some(email), Some(keywords)) = ( + let (metadata_, data_) = if let (Some(email), Some(keywords)) = ( self.server .get_property::>( account_id, @@ -323,7 +326,7 @@ impl SessionData { account_id, Collection::Email, id, - &Property::Keywords, + &Property::Value, ) .await .imap_ctx(&arguments.tag, trc::location!())?, @@ -343,8 +346,8 @@ impl SessionData { let metadata = metadata_ .unarchive::() .imap_ctx(&arguments.tag, trc::location!())?; - let keywords = keywords_ - .unarchive::>() + let data = data_ + .to_unarchived::() .imap_ctx(&arguments.tag, trc::location!())?; // Fetch and parse blob @@ -380,8 +383,12 @@ impl SessionData { // Build response let mut items = Vec::with_capacity(arguments.attributes.len()); - let set_seen_flag = - set_seen_flags && !keywords.iter().any(|k| k == &ArchivedKeyword::Seen); + let set_seen_flag = set_seen_flags + && !data + .inner + .keywords + .iter() + .any(|k| k == &ArchivedKeyword::Seen); let thread_id = if needs_thread_id || set_seen_flag { if let Some(thread_id) = self .server @@ -404,7 +411,12 @@ impl SessionData { }); } Attribute::Flags => { - let mut flags = keywords.iter().map(Flag::from).collect::>(); + let mut flags = data + .inner + .keywords + .iter() + .map(Flag::from) + .collect::>(); if set_seen_flag { flags.push(Flag::Seen); } @@ -515,13 +527,9 @@ impl SessionData { } } Attribute::ModSeq => { - if let Ok(Some(modseq)) = self - .server - .get_property::(account_id, Collection::Email, id, Property::Cid) - .await - { - items.push(DataItem::ModSeq { modseq: modseq + 1 }); - } + items.push(DataItem::ModSeq { + modseq: u64::from(data.inner.change_id) + 1, + }); } Attribute::EmailId => { items.push(DataItem::EmailId { @@ -538,7 +546,12 @@ impl SessionData { // Add flags to the response if the message was unseen if set_seen_flag && !arguments.attributes.contains(&Attribute::Flags) { - let mut flags = keywords.iter().map(Flag::from).collect::>(); + let mut flags = data + .inner + .keywords + .iter() + .map(Flag::from) + .collect::>(); flags.push(Flag::Seen); items.push(DataItem::Flags { flags }); } @@ -550,40 +563,32 @@ impl SessionData { // Add to set flags if set_seen_flag { - set_seen_ids.push(( - Id::from_parts(thread_id, id), - Archive { - hash: keywords_.hash, - version: keywords_.version, - inner: rkyv_deserialize::<_, Vec>(keywords) - .imap_ctx(&arguments.tag, trc::location!())?, - }, - )); - } - } + let mut new_data = data + .deserialize() + .imap_ctx(&arguments.tag, trc::location!())?; + new_data.keywords.push(Keyword::Seen); + new_data.change_id = change_id; - // Set Seen ids - if !set_seen_ids.is_empty() { - let mut changelog = self - .server - .begin_changes(account_id) - .imap_ctx(&arguments.tag, trc::location!())?; - for (id, mut keywords) in set_seen_ids { - keywords.inner.push(Keyword::Seen); let mut batch = BatchBuilder::new(); batch .with_account_id(account_id) .with_collection(Collection::Email) - .update_document(id.document_id()) - .assert_value(Property::Keywords, &keywords) - .set( - Property::Keywords, - Archiver::new(keywords.inner) - .serialize() - .imap_ctx(&arguments.tag, trc::location!())?, + .update_document(id) + .custom( + ObjectIndexBuilder::new() + .with_current(data) + .with_changes(new_data), ) - .tag(Property::Keywords, Keyword::Seen) - .set(Property::Cid, changelog.change_id.serialize()); + .imap_ctx(&arguments.tag, trc::location!())?; + + update_batches.push((Id::from_parts(thread_id, id), batch)); + } + } + + // Set Seen ids + if !update_batches.is_empty() { + let mut changelog = ChangeLogBuilder::with_change_id(change_id); + for (id, batch) in update_batches { match self .server .store() diff --git a/crates/imap/src/op/store.rs b/crates/imap/src/op/store.rs index 0e474c9d..41cdc13d 100644 --- a/crates/imap/src/op/store.rs +++ b/crates/imap/src/op/store.rs @@ -11,12 +11,9 @@ use crate::{ spawn_op, }; use ahash::AHashSet; -use common::{listener::SessionStream, storage::tag::TagManager}; +use common::{listener::SessionStream, storage::index::ObjectIndexBuilder}; use directory::Permission; -use email::{ - mailbox::UidMailbox, - message::{bayes::EmailBayesTrain, ingest::EmailIngest}, -}; +use email::message::{bayes::EmailBayesTrain, ingest::EmailIngest, metadata::MessageData}; use imap_proto::{ Command, ResponseCode, ResponseType, StatusResponse, protocol::{ @@ -31,7 +28,6 @@ use jmap_proto::types::{ state::StateChange, type_state::DataType, }; use store::{ - SerializeInfallible, query::log::{Change, Query}, write::{AlignedBytes, Archive, BatchBuilder, ValueClass, log::ChangeLogBuilder}, }; @@ -205,13 +201,13 @@ impl SessionData { let mut try_count = 0; loop { // Obtain current keywords - let (mut keywords, thread_id) = if let (Some(keywords), Some(thread_id)) = ( + let (data_, thread_id) = if let (Some(data), Some(thread_id)) = ( self.server .get_property::>( account_id, Collection::Email, *id, - Property::Keywords, + Property::Value, ) .await .imap_ctx(response.tag.as_ref().unwrap(), trc::location!())?, @@ -220,40 +216,48 @@ impl SessionData { .await .imap_ctx(response.tag.as_ref().unwrap(), trc::location!())?, ) { - ( - TagManager::new( - keywords - .into_deserialized::>() - .imap_ctx(response.tag.as_ref().unwrap(), trc::location!())?, - ), - thread_id, - ) + (data, thread_id) } else { continue 'outer; }; + // Deserialize + let data = data_ + .to_unarchived::() + .imap_ctx(response.tag.as_ref().unwrap(), trc::location!())?; + let mut new_data = data + .deserialize() + .imap_ctx(response.tag.as_ref().unwrap(), trc::location!())?; + // Apply changes + let mut seen_changed = false; match arguments.operation { Operation::Set => { - keywords.set(set_keywords.clone()); + seen_changed = set_keywords.contains(&Keyword::Seen) + != new_data.has_keyword(&Keyword::Seen); + new_data.set_keywords(set_keywords.clone()); } Operation::Add => { for keyword in &set_keywords { - keywords.update(keyword.clone(), true); + if new_data.add_keyword(keyword.clone()) && keyword == &Keyword::Seen { + seen_changed = true; + } } } Operation::Clear => { for keyword in &set_keywords { - keywords.update(keyword.clone(), false); + if new_data.remove_keyword(keyword) && keyword == &Keyword::Seen { + seen_changed = true; + } } } } - if keywords.has_changes() { + if new_data.has_keyword_changes(data.inner) { // Train spam filter let mut train_spam = None; if can_spam_train { - for keyword in keywords.added() { + for keyword in new_data.added_keywords(data.inner) { if keyword == &Keyword::Junk { train_spam = Some(true); break; @@ -263,7 +267,7 @@ impl SessionData { } } if train_spam.is_none() { - for keyword in keywords.removed() { + for keyword in new_data.removed_keywords(data.inner) { if keyword == &Keyword::Junk { train_spam = Some(false); break; @@ -273,12 +277,9 @@ impl SessionData { }; // Convert keywords to flags - let seen_changed = keywords - .changed_tags() - .any(|keyword| keyword == &Keyword::Seen); let flags = if !arguments.is_silent { - keywords - .current() + new_data + .keywords .iter() .cloned() .map(Flag::from) @@ -287,22 +288,34 @@ impl SessionData { vec![] }; - // Write changes - let mut batch = BatchBuilder::new(); - batch - .with_account_id(account_id) - .with_collection(Collection::Email) - .update_document(*id); - keywords - .update_batch(&mut batch, Property::Keywords) - .imap_ctx(response.tag.as_ref().unwrap(), trc::location!())?; + // Add change id if changelog.change_id == u64::MAX { changelog.change_id = self .server .assign_change_id(account_id) .imap_ctx(response.tag.as_ref().unwrap(), trc::location!())? } - batch.set(Property::Cid, changelog.change_id.serialize()); + new_data.change_id = changelog.change_id; + + // Set all current mailboxes as changed if the Seen tag changed + if seen_changed { + for mailbox_id in new_data.mailboxes.iter() { + changed_mailboxes.insert(mailbox_id.mailbox_id); + } + } + + // Write changes + let mut batch = BatchBuilder::new(); + batch + .with_account_id(account_id) + .with_collection(Collection::Email) + .update_document(*id) + .custom( + ObjectIndexBuilder::new() + .with_current(data) + .with_changes(new_data), + ) + .imap_ctx(response.tag.as_ref().unwrap(), trc::location!())?; // Add spam train task if let Some(learn_spam) = train_spam { @@ -326,29 +339,6 @@ impl SessionData { .caused_by(trc::location!()) { Ok(_) => { - // Set all current mailboxes as changed if the Seen tag changed - if seen_changed { - if let Some(mailboxes) = self - .server - .get_property::>( - account_id, - Collection::Email, - *id, - Property::MailboxIds, - ) - .await - .imap_ctx(response.tag.as_ref().unwrap(), trc::location!())? - { - for mailbox_id in mailboxes - .unarchive::>() - .imap_ctx(response.tag.as_ref().unwrap(), trc::location!())? - .iter() - { - changed_mailboxes.insert(u32::from(mailbox_id.mailbox_id)); - } - } - } - // Update changelog changelog.log_update(Collection::Email, Id::from_parts(thread_id, *id)); diff --git a/crates/jmap-proto/src/types/keyword.rs b/crates/jmap-proto/src/types/keyword.rs index d9e2fd2f..79a9bfa3 100644 --- a/crates/jmap-proto/src/types/keyword.rs +++ b/crates/jmap-proto/src/types/keyword.rs @@ -39,7 +39,7 @@ pub const OTHER: usize = 12; serde::Serialize, )] #[serde(untagged)] -#[rkyv(derive(PartialEq))] +#[rkyv(derive(PartialEq), compare(PartialEq))] pub enum Keyword { #[serde(rename(serialize = "$seen"))] Seen, @@ -300,24 +300,6 @@ impl From<&Keyword> for TagValue { } } -impl From for TagValue { - fn from(value: Keyword) -> Self { - match value.into_id() { - Ok(id) => TagValue::Id(MaybeDynamicId::Static(id)), - Err(string) => TagValue::Text(string.into_bytes()), - } - } -} - -impl From<&Keyword> for TagValue { - fn from(value: &Keyword) -> Self { - match value.id() { - Ok(id) => TagValue::Id(MaybeDynamicId::Static(id)), - Err(string) => TagValue::Text(string.into_bytes()), - } - } -} - impl From<&ArchivedKeyword> for TagValue { fn from(value: &ArchivedKeyword) -> Self { match value.id() { diff --git a/crates/jmap/src/blob/get.rs b/crates/jmap/src/blob/get.rs index 6ae56b83..bcf14430 100644 --- a/crates/jmap/src/blob/get.rs +++ b/crates/jmap/src/blob/get.rs @@ -5,7 +5,7 @@ */ use common::{Server, auth::AccessToken}; -use email::mailbox::UidMailbox; +use email::message::metadata::MessageData; use jmap_proto::{ method::{ get::{GetRequest, GetResponse}, @@ -24,7 +24,10 @@ use jmap_proto::{ use mail_builder::encoders::base64::base64_encode; use sha1::{Digest, Sha1}; use sha2::{Sha256, Sha512}; -use store::{write::{AlignedBytes, Archive}, BlobClass}; +use store::{ + BlobClass, + write::{AlignedBytes, Archive}, +}; use trc::AddContext; use utils::map::vec_map::VecMap; @@ -243,15 +246,16 @@ impl BlobOperations for Server { req_account_id, Collection::Email, *document_id, - Property::MailboxIds, + Property::Value, ) .await? { matched_ids.append( DataType::Mailbox, mailboxes - .unarchive::>() + .unarchive::() .caused_by(trc::location!())? + .mailboxes .iter() .map(|m| { debug_assert!(m.uid != 0); diff --git a/crates/jmap/src/email/get.rs b/crates/jmap/src/email/get.rs index a298a587..e9a92d5d 100644 --- a/crates/jmap/src/email/get.rs +++ b/crates/jmap/src/email/get.rs @@ -7,9 +7,9 @@ use common::{Server, auth::AccessToken}; use email::{ - mailbox::UidMailbox, message::metadata::{ - ArchivedGetHeader, ArchivedHeaderName, ArchivedMetadataPartType, MessageMetadata, + ArchivedGetHeader, ArchivedHeaderName, ArchivedMetadataPartType, MessageData, + MessageMetadata, }, thread::cache::ThreadCache, }; @@ -22,7 +22,6 @@ use jmap_proto::{ collection::Collection, date::UTCDate, id::Id, - keyword::Keyword, property::{HeaderForm, Property}, value::{Object, Value}, }, @@ -155,7 +154,7 @@ impl EmailGet for Server { } } - 'outer: for id in ids { + for id in ids { // Obtain the email object if !message_ids.contains(id.document_id()) { response.not_found.push(id.into()); @@ -180,6 +179,26 @@ impl EmailGet for Server { .unarchive::() .caused_by(trc::location!())?; + // Obtain message data + let data_ = match self + .get_property::>( + account_id, + Collection::Email, + id.document_id(), + &Property::Value, + ) + .await? + { + Some(data) => data, + None => { + response.not_found.push(id.into()); + continue; + } + }; + let data = data_ + .unarchive::() + .caused_by(trc::location!())?; + // Retrieve raw message if needed let blob_hash = BlobHash::from(&metadata.blob_hash); let raw_message: Cow<[u8]> = if needs_body { @@ -228,73 +247,23 @@ impl EmailGet for Server { email.append(Property::BlobId, blob_id.clone()); } Property::MailboxIds => { - if let Some(mailboxes_) = self - .get_property::>( - account_id, - Collection::Email, - id.document_id(), - &Property::MailboxIds, - ) - .await? - { - let mailboxes = mailboxes_ - .unarchive::>() - .caused_by(trc::location!())?; - let mut obj = Object::with_capacity(mailboxes.len()); - for id in mailboxes.iter() { - debug_assert!(id.uid != 0); - obj.append( - Property::_T(Id::from(u32::from(id.mailbox_id)).to_string()), - true, - ); - } - - email.append(property.clone(), Value::Object(obj)); - } else { - trc::event!( - Store(StoreEvent::NotFound), - AccountId = account_id, - DocumentId = id.document_id(), - Collection = Collection::Email, - Details = "Mailbox property not found.", - CausedBy = trc::location!(), + let mut obj = Object::with_capacity(data.mailboxes.len()); + for id in data.mailboxes.iter() { + debug_assert!(id.uid != 0); + obj.append( + Property::_T(Id::from(u32::from(id.mailbox_id)).to_string()), + true, ); - - response.not_found.push(id.into()); - continue 'outer; } + + email.append(property.clone(), Value::Object(obj)); } Property::Keywords => { - if let Some(keywords_) = self - .get_property::>( - account_id, - Collection::Email, - id.document_id(), - &Property::Keywords, - ) - .await? - { - let keywords = keywords_ - .unarchive::>() - .caused_by(trc::location!())?; - let mut obj = Object::with_capacity(keywords.len()); - for keyword in keywords.iter() { - obj.append(Property::_T(keyword.to_string()), true); - } - email.append(property.clone(), Value::Object(obj)); - } else { - trc::event!( - Store(StoreEvent::NotFound), - AccountId = account_id, - DocumentId = id.document_id(), - Collection = Collection::Email, - Details = "Keywords property not found.", - CausedBy = trc::location!(), - ); - - response.not_found.push(id.into()); - continue 'outer; + let mut obj = Object::with_capacity(data.keywords.len()); + for keyword in data.keywords.iter() { + obj.append(Property::_T(keyword.to_string()), true); } + email.append(property.clone(), Value::Object(obj)); } Property::Size => { email.append(Property::Size, u32::from(metadata.size)); diff --git a/crates/jmap/src/email/set.rs b/crates/jmap/src/email/set.rs index 680e08ee..27ed9dc5 100644 --- a/crates/jmap/src/email/set.rs +++ b/crates/jmap/src/email/set.rs @@ -6,12 +6,13 @@ use std::{borrow::Cow, collections::HashMap}; -use common::{Server, auth::AccessToken, storage::tag::TagManager}; +use common::{Server, auth::AccessToken, storage::index::ObjectIndexBuilder}; use email::{ mailbox::{UidMailbox, manage::MailboxFnc}, message::{ delete::EmailDeletion, ingest::{EmailIngest, IngestEmail, IngestSource}, + metadata::MessageData, }, }; use http_proto::HttpSessionData; @@ -39,7 +40,9 @@ use mail_builder::{ }; use mail_parser::MessageParser; use store::{ - ahash::AHashSet, roaring::RoaringBitmap, write::{log::ChangeLogBuilder, AlignedBytes, Archive, BatchBuilder}, SerializeInfallible + ahash::AHashSet, + roaring::RoaringBitmap, + write::{AlignedBytes, Archive, BatchBuilder, log::ChangeLogBuilder}, }; use trc::AddContext; @@ -762,46 +765,29 @@ impl EmailSet for Server { continue 'update; } - // Obtain current keywords and mailboxes + // Obtain message data let document_id = id.document_id(); - let (mut mailboxes, mut keywords) = if let (Some(mailboxes), Some(keywords)) = ( - self.get_property::>( + let data_ = match self + .get_property::>( account_id, Collection::Email, document_id, - Property::MailboxIds, + &Property::Value, ) - .await?, - self.get_property::>( - account_id, - Collection::Email, - document_id, - Property::Keywords, - ) - .await?, - ) { - ( - TagManager::new( - mailboxes - .into_deserialized::>() - .caused_by(trc::location!())?, - ), - TagManager::new( - keywords - .into_deserialized::>() - .caused_by(trc::location!())?, - ), - ) - } else { - response.not_updated.append(id, SetError::not_found()); - continue 'update; + .await? + { + Some(data) => data, + None => { + response.not_updated.append(id, SetError::not_found()); + continue 'update; + } }; - - // Prepare write batch - let mut batch = BatchBuilder::new(); - batch - .with_account_id(account_id) - .with_collection(Collection::Email); + let data = data_ + .to_unarchived::() + .caused_by(trc::location!())?; + let mut new_data = data + .deserialize::() + .caused_by(trc::location!())?; for (property, value) in object.0 { let value = match response.eval_object_references(value) { @@ -813,7 +799,7 @@ impl EmailSet for Server { }; match (property, value) { (Property::MailboxIds, MaybePatchValue::Value(Value::List(ids))) => { - mailboxes.set( + new_data.set_mailboxes( ids.into_iter() .filter_map(|id| { UidMailbox::new_unassigned(id.try_unwrap_id()?.document_id()) @@ -825,14 +811,15 @@ impl EmailSet for Server { (Property::MailboxIds, MaybePatchValue::Patch(patch)) => { let mut patch = patch.into_iter(); if let Some(id) = patch.next().unwrap().try_unwrap_id() { - mailboxes.update( - UidMailbox::new_unassigned(id.document_id()), - patch.next().unwrap().try_unwrap_bool().unwrap_or_default(), - ); + if patch.next().unwrap().try_unwrap_bool().unwrap_or_default() { + new_data.add_mailbox(UidMailbox::new_unassigned(id.document_id())); + } else { + new_data.remove_mailbox(id.document_id()); + } } } (Property::Keywords, MaybePatchValue::Value(Value::List(keywords_))) => { - keywords.set( + new_data.set_keywords( keywords_ .into_iter() .filter_map(|keyword| keyword.try_unwrap_keyword()) @@ -842,10 +829,11 @@ impl EmailSet for Server { (Property::Keywords, MaybePatchValue::Patch(patch)) => { let mut patch = patch.into_iter(); if let Some(keyword) = patch.next().unwrap().try_unwrap_keyword() { - keywords.update( - keyword, - patch.next().unwrap().try_unwrap_bool().unwrap_or_default(), - ); + if patch.next().unwrap().try_unwrap_bool().unwrap_or_default() { + new_data.add_keyword(keyword); + } else { + new_data.remove_keyword(&keyword); + } } } (property, _) => { @@ -855,7 +843,9 @@ impl EmailSet for Server { } } - if !mailboxes.has_changes() && !keywords.has_changes() { + let has_keyword_changes = new_data.has_keyword_changes(data.inner); + let has_mailbox_changes = new_data.has_mailbox_changes(data.inner); + if !has_keyword_changes && !has_mailbox_changes { response.not_updated.append( id, SetError::invalid_properties() @@ -864,13 +854,21 @@ impl EmailSet for Server { continue 'update; } - // Log change - batch.update_document(document_id); + // Prepare write batch + let mut batch = BatchBuilder::new(); + batch + .with_account_id(account_id) + .with_collection(Collection::Email) + .update_document(document_id); + if changes.change_id == u64::MAX { + changes.change_id = self.assign_change_id(account_id)?; + } + new_data.change_id = changes.change_id; let mut changed_mailboxes = AHashSet::new(); changes.log_update(Collection::Email, id); // Process keywords - if keywords.has_changes() { + if has_keyword_changes { // Verify permissions on shared accounts if matches!(&can_modify_message_ids, Some(ids) if !ids.contains(document_id)) { response.not_updated.append( @@ -882,31 +880,23 @@ impl EmailSet for Server { } // Set all current mailboxes as changed if the Seen tag changed - if keywords - .changed_tags() + if new_data + .added_keywords(data.inner) .any(|keyword| keyword == &Keyword::Seen) + || new_data + .removed_keywords(data.inner) + .any(|keyword| keyword == &Keyword::Seen) { - for mailbox_id in mailboxes.current() { + for mailbox_id in new_data.mailboxes.iter() { changed_mailboxes.insert(mailbox_id.mailbox_id); } } - - // Update keywords property - keywords - .update_batch(&mut batch, Property::Keywords) - .caused_by(trc::location!())?; - - // Update last change id - if changes.change_id == u64::MAX { - changes.change_id = self.assign_change_id(account_id)?; - } - batch.set(Property::Cid, changes.change_id.serialize()); } // Process mailboxes - if mailboxes.has_changes() { + if has_mailbox_changes { // Make sure the message is at least in one mailbox - if !mailboxes.has_tags() { + if new_data.mailboxes.is_empty() { response.not_updated.append( id, SetError::invalid_properties() @@ -917,7 +907,7 @@ impl EmailSet for Server { } // Make sure all new mailboxIds are valid - for mailbox_id in mailboxes.added() { + for mailbox_id in new_data.added_mailboxes(data.inner) { if mailbox_ids.contains(mailbox_id.mailbox_id) { // Verify permissions on shared accounts if !matches!(&can_add_mailbox_ids, Some(ids) if !ids.contains(mailbox_id.mailbox_id)) @@ -948,11 +938,11 @@ impl EmailSet for Server { } // Add all removed mailboxes to change list - for mailbox_id in mailboxes.removed() { + for mailbox_id in new_data.removed_mailboxes(data.inner) { // Verify permissions on shared accounts - if !matches!(&can_delete_mailbox_ids, Some(ids) if !ids.contains(mailbox_id.mailbox_id)) + if !matches!(&can_delete_mailbox_ids, Some(ids) if !ids.contains(u32::from(mailbox_id.mailbox_id))) { - changed_mailboxes.insert(mailbox_id.mailbox_id); + changed_mailboxes.insert(u32::from(mailbox_id.mailbox_id)); } else { response.not_updated.append( id, @@ -966,7 +956,7 @@ impl EmailSet for Server { } // Obtain IMAP UIDs for added mailboxes - for uid_mailbox in mailboxes.inner_tags_mut() { + for uid_mailbox in &mut new_data.mailboxes { if uid_mailbox.uid == 0 { uid_mailbox.uid = self .assign_imap_uid(account_id, uid_mailbox.mailbox_id) @@ -974,11 +964,6 @@ impl EmailSet for Server { .caused_by(trc::location!())?; } } - - // Update mailboxIds property - mailboxes - .update_batch(&mut batch, Property::MailboxIds) - .caused_by(trc::location!())?; } // Log mailbox changes @@ -987,23 +972,29 @@ impl EmailSet for Server { } // Write changes - if !batch.is_empty() { - match self.core.storage.data.write(batch.build()).await { - Ok(_) => { - // Add to updated list - response.updated.append(id, None); - } - Err(err) if err.is_assertion_failure() => { - response.not_updated.append( - id, - SetError::forbidden().with_description( - "Another process modified this message, please try again.", - ), - ); - } - Err(err) => { - return Err(err.caused_by(trc::location!())); - } + batch + .custom( + ObjectIndexBuilder::new() + .with_current(data) + .with_changes(new_data), + ) + .caused_by(trc::location!())?; + + match self.core.storage.data.write(batch.build()).await { + Ok(_) => { + // Add to updated list + response.updated.append(id, None); + } + Err(err) if err.is_assertion_failure() => { + response.not_updated.append( + id, + SetError::forbidden().with_description( + "Another process modified this message, please try again.", + ), + ); + } + Err(err) => { + return Err(err.caused_by(trc::location!())); } } } diff --git a/crates/jmap/src/sieve/set.rs b/crates/jmap/src/sieve/set.rs index 889ab1ff..03ade972 100644 --- a/crates/jmap/src/sieve/set.rs +++ b/crates/jmap/src/sieve/set.rs @@ -9,7 +9,9 @@ use common::{ auth::{AccessToken, ResourceToken}, storage::index::ObjectIndexBuilder, }; -use email::sieve::{SieveScript, activate::SieveScriptActivate, delete::SieveScriptDelete}; +use email::sieve::{ + ArchivedSieveScript, SieveScript, activate::SieveScriptActivate, delete::SieveScriptDelete, +}; use http_proto::HttpSessionData; use jmap_proto::{ error::set::{SetError, SetErrorType}, @@ -28,7 +30,10 @@ use jmap_proto::{ use rand::distr::Alphanumeric; use sieve::compiler::ErrorType; use store::{ - query::Filter, rand::{rng, Rng}, write::{log::ChangeLogBuilder, AlignedBytes, Archive, BatchBuilder}, BlobClass + BlobClass, Serialize, + query::Filter, + rand::{Rng, rng}, + write::{AlignedBytes, Archive, BatchBuilder, LegacyBincode, log::ChangeLogBuilder}, }; use trc::AddContext; @@ -50,17 +55,17 @@ pub trait SieveScriptSet: Sync + Send { ) -> impl Future> + Send; #[allow(clippy::type_complexity)] - fn sieve_set_item( + fn sieve_set_item<'x>( &self, changes_: Object, - update: Option<(u32, Archive)>, + update: Option<(u32, Archive<&'x ArchivedSieveScript>)>, ctx: &SetContext, session_id: u64, ) -> impl Future< Output = trc::Result< Result< ( - ObjectIndexBuilder, + ObjectIndexBuilder<&'x ArchivedSieveScript, SieveScript>, Option>, ), SetError, @@ -173,7 +178,7 @@ impl SieveScriptSet for Server { // Obtain sieve script let document_id = id.document_id(); - if let Some(sieve) = self + if let Some(sieve_) = self .get_property::>( account_id, Collection::SieveScript, @@ -182,8 +187,8 @@ impl SieveScriptSet for Server { ) .await? { - let sieve = sieve - .into_deserialized::() + let sieve = sieve_ + .to_unarchived::() .caused_by(trc::location!())?; match self @@ -331,16 +336,16 @@ impl SieveScriptSet for Server { } #[allow(clippy::blocks_in_conditions)] - async fn sieve_set_item( + async fn sieve_set_item<'x>( &self, changes_: Object, - update: Option<(u32, Archive)>, + update: Option<(u32, Archive<&'x ArchivedSieveScript>)>, ctx: &SetContext<'_>, session_id: u64, ) -> trc::Result< Result< ( - ObjectIndexBuilder, + ObjectIndexBuilder<&'x ArchivedSieveScript, SieveScript>, Option>, ), SetError, @@ -360,7 +365,7 @@ impl SieveScriptSet for Server { // Parse properties let mut changes = update .as_ref() - .map(|(_, obj)| obj.inner.clone()) + .map(|(_, obj)| obj.deserialize().unwrap_or_default()) .unwrap_or_default(); let mut blob_id = None; for (property, value) in changes_.0 { @@ -464,7 +469,7 @@ impl SieveScriptSet for Server { match self.core.sieve.untrusted_compiler.compile(&bytes) { Ok(script) => { changes.size = bytes.len() as u32; - bytes.extend(bincode::serialize(&script).unwrap_or_default()); + bytes.extend(&LegacyBincode::new(script).serialize().unwrap_or_default()); bytes.into() } Err(err) => { diff --git a/crates/jmap/src/vacation/set.rs b/crates/jmap/src/vacation/set.rs index 08049a0d..58df21b4 100644 --- a/crates/jmap/src/vacation/set.rs +++ b/crates/jmap/src/vacation/set.rs @@ -25,9 +25,12 @@ use jmap_proto::{ use mail_builder::MessageBuilder; use mail_parser::decoders::html::html_to_text; use std::future::Future; -use store::write::{ - AlignedBytes, Archive, BatchBuilder, - log::{Changes, LogInsert}, +use store::{ + Serialize, + write::{ + AlignedBytes, Archive, BatchBuilder, LegacyBincode, + log::{Changes, LogInsert}, + }, }; use trc::AddContext; @@ -446,7 +449,11 @@ impl VacationResponseSet for Server { obj.size = script.len() as u32; // Serialize script - script.extend(bincode::serialize(&compiled_script).unwrap_or_default()); + script.extend( + LegacyBincode::new(compiled_script) + .serialize() + .unwrap_or_default(), + ); Ok(script) } diff --git a/crates/managesieve/src/op/putscript.rs b/crates/managesieve/src/op/putscript.rs index a9d25118..096b48e2 100644 --- a/crates/managesieve/src/op/putscript.rs +++ b/crates/managesieve/src/op/putscript.rs @@ -13,8 +13,9 @@ use imap_proto::receiver::Request; use jmap_proto::types::{collection::Collection, property::Property}; use sieve::compiler::ErrorType; use store::{ + Serialize, query::Filter, - write::{AlignedBytes, Archive, BatchBuilder, log::LogInsert}, + write::{AlignedBytes, Archive, BatchBuilder, LegacyBincode, log::LogInsert}, }; use trc::AddContext; @@ -79,7 +80,11 @@ impl Session { .compile(&script_bytes) { Ok(compiled_script) => { - script_bytes.extend(bincode::serialize(&compiled_script).unwrap_or_default()); + script_bytes.extend( + LegacyBincode::new(compiled_script) + .serialize() + .unwrap_or_default(), + ); } Err(err) => { return Err(if let ErrorType::ScriptTooLong = &err.error_type() { @@ -98,7 +103,7 @@ impl Session { // Validate name if let Some(document_id) = self.validate_name(account_id, &name).await? { // Obtain script values - let script = self + let script_ = self .server .get_property::>( account_id, @@ -113,8 +118,9 @@ impl Session { .into_err() .details("Script not found") .code(ResponseCode::NonExistent) - })? - .into_deserialized::() + })?; + let script = script_ + .to_unarchived::() .caused_by(trc::location!())?; // Write script blob @@ -135,8 +141,8 @@ impl Session { ObjectIndexBuilder::new() .with_changes( script - .inner - .clone() + .deserialize() + .caused_by(trc::location!())? .with_size(script_size as u32) .with_blob_hash(blob_hash.clone()), ) diff --git a/crates/pop3/src/mailbox.rs b/crates/pop3/src/mailbox.rs index 9919f0c5..2e126f11 100644 --- a/crates/pop3/src/mailbox.rs +++ b/crates/pop3/src/mailbox.rs @@ -7,7 +7,10 @@ use std::collections::BTreeMap; use common::listener::SessionStream; -use email::mailbox::{INBOX_ID, UidMailbox, manage::MailboxFnc}; +use email::{ + mailbox::{INBOX_ID, manage::MailboxFnc}, + message::metadata::MessageData, +}; use jmap_proto::types::{collection::Collection, property::Property}; use store::{ IndexKey, IterateParams, SerializeInfallible, U32_LEN, @@ -128,7 +131,7 @@ impl Session { account_id, Collection::Email, &message_ids, - Property::MailboxIds, + Property::Value, ) .await .caused_by(trc::location!())? @@ -136,8 +139,9 @@ impl Session { { // Make sure the message is still in Inbox if let Some(item) = uid_mailbox - .unarchive::>() + .unarchive::() .caused_by(trc::location!())? + .mailboxes .iter() .find(|item| item.mailbox_id == INBOX_ID) { diff --git a/crates/store/src/lib.rs b/crates/store/src/lib.rs index 8c044116..9efaf9c8 100644 --- a/crates/store/src/lib.rs +++ b/crates/store/src/lib.rs @@ -4,7 +4,7 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use std::{borrow::Cow, collections::HashSet, sync::Arc}; +use std::{borrow::Cow, sync::Arc}; pub mod backend; pub mod config; @@ -775,6 +775,12 @@ impl From> for trc::Value { } } +impl From> for () { + fn from(_: Value<'static>) -> Self { + unreachable!() + } +} + impl Stores { pub fn disable_enterprise_only(&mut self) { #[cfg(feature = "enterprise")] @@ -793,15 +799,3 @@ impl SerializedVersion for () { 0 } } - -impl SerializedVersion for Vec { - fn serialize_version() -> u8 { - T::serialize_version() - } -} - -impl SerializedVersion for HashSet { - fn serialize_version() -> u8 { - T::serialize_version() - } -} diff --git a/crates/store/src/write/serialize.rs b/crates/store/src/write/serialize.rs index ad7dc3de..fcb0f4ec 100644 --- a/crates/store/src/write/serialize.rs +++ b/crates/store/src/write/serialize.rs @@ -308,7 +308,7 @@ where + Sync + Send, { - pub fn into_deserialized(&self) -> trc::Result> + pub fn to_deserialized(&self) -> trc::Result> where T: rkyv::Deserialize>, { @@ -324,6 +324,17 @@ where inner, }) } + + pub fn deserialize(&self) -> trc::Result + where + T: rkyv::Deserialize>, + { + rkyv::deserialize::(self.inner).map_err(|err| { + trc::StoreEvent::DeserializeError + .caused_by(trc::location!()) + .reason(err) + }) + } } #[inline] diff --git a/tests/src/jmap/mod.rs b/tests/src/jmap/mod.rs index f113d9b2..b6a2c817 100644 --- a/tests/src/jmap/mod.rs +++ b/tests/src/jmap/mod.rs @@ -387,7 +387,7 @@ pub async fn jmap_tests() { mailbox::test(&mut params).await; delivery::test(&mut params).await; auth_acl::test(&mut params).await; - auth_limits::test(&mut params).await;*/ + auth_limits::test(&mut params).await; auth_oauth::test(&mut params).await; event_source::test(&mut params).await; push_subscription::test(&mut params).await; @@ -397,7 +397,7 @@ pub async fn jmap_tests() { websocket::test(&mut params).await; quota::test(&mut params).await; crypto::test(&mut params).await; - blob::test(&mut params).await; + blob::test(&mut params).await;*/ permissions::test(¶ms).await; purge::test(&mut params).await; enterprise::test(&mut params).await; diff --git a/tests/src/jmap/stress_test.rs b/tests/src/jmap/stress_test.rs index 08d6967e..e09fdd1d 100644 --- a/tests/src/jmap/stress_test.rs +++ b/tests/src/jmap/stress_test.rs @@ -9,7 +9,7 @@ use std::{sync::Arc, time::Duration}; use crate::jmap::{mailbox::destroy_all_mailboxes_no_wait, wait_for_index}; use common::Server; use directory::backend::internal::manage::ManageDirectory; -use email::mailbox::UidMailbox; +use email::message::metadata::MessageData; use futures::future::join_all; use jmap_client::{ client::Client, @@ -241,7 +241,7 @@ async fn email_tests(server: Server, client: Arc) { .await .unwrap() { - let mailbox_tags = mailbox_tags.deserialize::>().unwrap(); + let mailbox_tags = mailbox_tags.deserialize::().unwrap().mailboxes; if mailbox_tags.len() != 1 { panic!( "Email ORM has more than one mailbox {:?}! Id {} in mailbox {} with messages {:?}", diff --git a/tests/src/store/query.rs b/tests/src/store/query.rs index 58cea389..9802954d 100644 --- a/tests/src/store/query.rs +++ b/tests/src/store/query.rs @@ -18,7 +18,7 @@ use store::{ ahash::AHashMap, fts::{Field, FtsFilter, index::FtsDocument}, query::sort::Pagination, - write::ValueClass, + write::{TagValue, ValueClass}, }; use store::{ @@ -176,7 +176,10 @@ pub async fn test(db: Store, fts_store: FtsStore, do_insert: bool) { ValueClass::Property(field_id), field.to_lowercase().into_bytes(), ) - .tag(field_id, Keyword::Other(field.to_lowercase())) + .tag( + field_id, + TagValue::Text(field.to_lowercase().into_bytes()), + ) .index(field_id, field.to_lowercase()); } }