diff --git a/crates/core/src/blob/get.rs b/crates/core/src/blob/get.rs new file mode 100644 index 00000000..be05b74f --- /dev/null +++ b/crates/core/src/blob/get.rs @@ -0,0 +1,43 @@ +use mail_parser::{ + decoders::{base64::base64_decode, quoted_printable::quoted_printable_decode}, + Encoding, +}; +use protocol::types::blob::BlobId; + +use crate::JMAP; + +impl JMAP { + pub async fn blob_retrieve( + &self, + blob_id: &BlobId, + account_id: u32, + ) -> store::Result>> { + if !self + .store + .has_blob_access(&blob_id.hash, vec![account_id]) + .await? + { + // TODO: validate ACL + let acl = "true"; + return Ok(None); + } + + if let Some(section) = &blob_id.section { + Ok(self + .store + .get_blob( + &blob_id.hash, + (section.offset_start as u32) + ..(section.offset_start.saturating_add(section.size) as u32), + ) + .await? + .and_then(|bytes| match Encoding::from(section.encoding) { + Encoding::None => Some(bytes), + Encoding::Base64 => base64_decode(&bytes), + Encoding::QuotedPrintable => quoted_printable_decode(&bytes), + })) + } else { + self.store.get_blob(&blob_id.hash, 0..u32::MAX).await + } + } +} diff --git a/crates/core/src/blob/mod.rs b/crates/core/src/blob/mod.rs new file mode 100644 index 00000000..125ca70d --- /dev/null +++ b/crates/core/src/blob/mod.rs @@ -0,0 +1 @@ +pub mod get; diff --git a/crates/core/src/email/import.rs b/crates/core/src/email/import.rs index e1a51904..4076137e 100644 --- a/crates/core/src/email/import.rs +++ b/crates/core/src/email/import.rs @@ -1,17 +1,133 @@ use protocol::{ - error::method::MethodError, + error::{ + method::MethodError, + set::{SetError, SetErrorType}, + }, method::import::{ImportEmailRequest, ImportEmailResponse}, + types::{collection::Collection, property::Property, state::State}, }; +use store::BitmapKey; +use utils::map::vec_map::VecMap; -use crate::JMAP; +use crate::{MaybeError, JMAP}; impl JMAP { pub async fn email_import( &self, request: ImportEmailRequest, ) -> Result { - for (id, email) in request.emails {} + // Validate state + let account_id = request.account_id.document_id(); + let old_state: State = self + .store + .get_last_change_id(account_id, Collection::Email) + .await? + .into(); + if let Some(if_in_state) = request.if_in_state { + if old_state != if_in_state { + return Err(MethodError::StateMismatch); + } + } - todo!() + let cococ = "implement ACLS"; + let valid_mailbox_ids = self + .store + .get_bitmap(BitmapKey::document_ids(account_id, Collection::Mailbox)) + .await? + .unwrap_or_default(); + + let mut created = VecMap::with_capacity(request.emails.len()); + let mut not_created = VecMap::with_capacity(request.emails.len()); + + 'outer: for (id, email) in request.emails { + // Validate mailboxIds + let mailbox_ids = email + .mailbox_ids + .unwrap() + .into_iter() + .map(|m| m.unwrap().document_id()) + .collect::>(); + if mailbox_ids.is_empty() { + not_created.append( + id, + SetError::invalid_properties() + .with_property(Property::MailboxIds) + .with_description("Message must belong to at least one mailbox."), + ); + continue; + } + for mailbox_id in &mailbox_ids { + if !valid_mailbox_ids.contains(*mailbox_id) { + not_created.append( + id, + SetError::invalid_properties() + .with_property(Property::MailboxIds) + .with_description(format!("Mailbox {} does not exist.", mailbox_id)), + ); + continue 'outer; + } + } + + // Fetch raw message to import + let raw_message = + if let Some(raw_message) = self.blob_retrieve(&email.blob_id, account_id).await? { + raw_message + } else { + not_created.append( + id, + SetError::new(SetErrorType::BlobNotFound) + .with_description(format!("BlobId {} not found.", email.blob_id)), + ); + continue; + }; + + // Import message + match self + .email_ingest( + &raw_message, + account_id, + mailbox_ids, + email.keywords, + email.received_at.map(|r| r.into()), + ) + .await + { + Ok(email) => { + created.append(id, email.into()); + } + Err(MaybeError::Permanent(reason)) => { + not_created.append( + id, + SetError::new(SetErrorType::InvalidEmail).with_description(reason), + ); + } + Err(MaybeError::Temporary(_)) => { + return Err(MethodError::ServerPartialFail); + } + } + } + + Ok(ImportEmailResponse { + account_id: request.account_id, + new_state: if !created.is_empty() { + self.store + .get_last_change_id(account_id, Collection::Email) + .await? + .into() + } else { + old_state.clone() + }, + old_state: old_state.into(), + created: if !created.is_empty() { + created.into() + } else { + None + }, + not_created: if !not_created.is_empty() { + not_created.into() + } else { + None + }, + }) } } diff --git a/crates/core/src/email/ingest.rs b/crates/core/src/email/ingest.rs index 99a8fb5e..c2f20ba7 100644 --- a/crates/core/src/email/ingest.rs +++ b/crates/core/src/email/ingest.rs @@ -1,7 +1,13 @@ use mail_parser::{ parsers::fields::thread::thread_name, HeaderName, HeaderValue, Message, RfcHeader, }; -use protocol::types::{collection::Collection, id::Id, keyword::Keyword, property::Property}; +use protocol::{ + object::Object, + types::{ + blob::BlobId, collection::Collection, id::Id, keyword::Keyword, property::Property, + value::Value, + }, +}; use store::{ query::Filter, write::{log::ChangeLogBuilder, now, BatchBuilder, F_BITMAP, F_CLEAR, F_VALUE}, @@ -247,3 +253,12 @@ impl JMAP { } } } + +impl From for Object { + fn from(email: IngestedEmail) -> Self { + Object::with_capacity(3) + .with_property(Property::Id, email.id) + .with_property(Property::ThreadId, email.id.prefix_id()) + .with_property(Property::BlobId, BlobId::new(email.blob_hash)) + } +} diff --git a/crates/core/src/lib.rs b/crates/core/src/lib.rs index dcc27c53..47ee198a 100644 --- a/crates/core/src/lib.rs +++ b/crates/core/src/lib.rs @@ -1,6 +1,8 @@ +use protocol::error::method::MethodError; use store::{fts::Language, Store}; pub mod api; +pub mod blob; pub mod email; pub struct JMAP { @@ -33,3 +35,15 @@ impl From for MaybeError { } } } + +impl From for MethodError { + fn from(value: MaybeError) -> Self { + match value { + MaybeError::Temporary(msg) => { + let log = "true"; + MethodError::ServerPartialFail + } + MaybeError::Permanent(msg) => MethodError::InvalidArguments(msg), + } + } +} diff --git a/crates/protocol/src/method/import.rs b/crates/protocol/src/method/import.rs index b0cb93f1..3e7af966 100644 --- a/crates/protocol/src/method/import.rs +++ b/crates/protocol/src/method/import.rs @@ -28,7 +28,7 @@ pub struct ImportEmailRequest { #[derive(Debug, Clone)] pub struct ImportEmail { pub blob_id: BlobId, - pub mailbox_ids: Option>, ResultReference>>, + pub mailbox_ids: MaybeReference>, ResultReference>, pub keywords: Vec, pub received_at: Option, } @@ -102,7 +102,7 @@ impl JsonObjectParser for ImportEmail { { let mut request = ImportEmail { blob_id: BlobId::default(), - mailbox_ids: None, + mailbox_ids: MaybeReference::Value(vec![]), keywords: vec![], received_at: None, }; @@ -119,11 +119,11 @@ impl JsonObjectParser for ImportEmail { } 0x7364_4978_6f62_6c69_616d => { request.mailbox_ids = if !property.is_ref { - Some(MaybeReference::Value( + MaybeReference::Value( >>::parse(parser)?.values, - )) + ) } else { - Some(MaybeReference::Reference(ResultReference::parse(parser)?)) + MaybeReference::Reference(ResultReference::parse(parser)?) }; } 0x7364_726f_7779_656b if !property.is_ref => { diff --git a/crates/protocol/src/response/references.rs b/crates/protocol/src/response/references.rs index 220f5257..777bffff 100644 --- a/crates/protocol/src/response/references.rs +++ b/crates/protocol/src/response/references.rs @@ -130,23 +130,22 @@ impl Response { // Resolve email mailbox references for email in request.emails.values_mut() { match &mut email.mailbox_ids { - Some(MaybeReference::Reference(rr)) => { - email.mailbox_ids = Some(MaybeReference::Value( + MaybeReference::Reference(rr) => { + email.mailbox_ids = MaybeReference::Value( self.eval_result_references(rr) .unwrap_ids(rr)? .into_iter() .map(MaybeReference::Value) .collect(), - )); + ); } - Some(MaybeReference::Value(values)) => { + MaybeReference::Value(values) => { for value in values { if let MaybeReference::Reference(ir) = value { *value = MaybeReference::Value(self.eval_id_reference(ir)?); } } } - _ => (), } } } diff --git a/crates/protocol/src/types/date.rs b/crates/protocol/src/types/date.rs index f2a91949..7e4b9028 100644 --- a/crates/protocol/src/types/date.rs +++ b/crates/protocol/src/types/date.rs @@ -243,6 +243,12 @@ impl Serialize for UTCDate { } } +impl From for u64 { + fn from(value: UTCDate) -> Self { + value.timestamp() as u64 + } +} + #[cfg(test)] mod tests { use crate::{parser::json::Parser, types::date::UTCDate}; diff --git a/crates/store/src/backend/foundationdb/mod.rs b/crates/store/src/backend/foundationdb/mod.rs index adcd7c51..4767bee0 100644 --- a/crates/store/src/backend/foundationdb/mod.rs +++ b/crates/store/src/backend/foundationdb/mod.rs @@ -80,9 +80,9 @@ impl> Serialize for &BlobKey { KeySerializer::new(std::mem::size_of::>() + BLOB_HASH_LEN + 1) .write(SUBSPACE_BLOBS) .write(hash) - .write_leb128(self.account_id) + .write(self.account_id) .write(self.collection) - .write_leb128(self.document_id) + .write(self.document_id) .finalize() } } diff --git a/crates/store/src/backend/sqlite/mod.rs b/crates/store/src/backend/sqlite/mod.rs index 35c4f17d..360920e4 100644 --- a/crates/store/src/backend/sqlite/mod.rs +++ b/crates/store/src/backend/sqlite/mod.rs @@ -79,9 +79,9 @@ impl> Serialize for &BlobKey { let hash = self.hash.as_ref(); KeySerializer::new(std::mem::size_of::>() + BLOB_HASH_LEN + 1) .write(hash) - .write_leb128(self.account_id) + .write(self.account_id) .write(self.collection) - .write_leb128(self.document_id) + .write(self.document_id) .finalize() } } diff --git a/crates/store/src/blob/read.rs b/crates/store/src/blob/read.rs index f5c2ea54..673a29b5 100644 --- a/crates/store/src/blob/read.rs +++ b/crates/store/src/blob/read.rs @@ -1,11 +1,12 @@ use std::{io::SeekFrom, ops::Range}; +use roaring::RoaringBitmap; use tokio::{ fs::{self, File}, io::{AsyncReadExt, AsyncSeekExt}, }; -use crate::{BlobHash, Store}; +use crate::{write::key::DeserializeBigEndian, BlobHash, BlobKey, Store, BLOB_HASH_LEN}; use super::{get_path, BlobStore}; @@ -53,4 +54,70 @@ impl Store { BlobStore::Remote(_) => todo!(), } } + + pub async fn has_blob_access( + &self, + blob_hash: &BlobHash, + account_ids: Vec, + ) -> crate::Result { + // Check if the blob already exists + let from_key = BlobKey { + account_id: 0, + collection: 0, + document_id: 0, + hash: blob_hash.hash, + }; + let to_key = BlobKey { + account_id: u32::MAX, + collection: u8::MAX, + document_id: u32::MAX, + hash: blob_hash.hash, + }; + + self.iterate(false, from_key, to_key, true, false, move |acc, key, _| { + let account_id = key.deserialize_be_u32(BLOB_HASH_LEN)?; + if account_ids.contains(&account_id) { + *acc = true; + Ok(false) + } else { + Ok(true) + } + }) + .await + } + + pub async fn has_blob_access_doc( + &self, + blob_hash: &BlobHash, + account_id: u32, + collection: impl Into, + document_ids: RoaringBitmap, + ) -> crate::Result { + // Check if the blob already exists + let collection = collection.into(); + let from_key = BlobKey { + account_id, + collection, + document_id: 0, + hash: blob_hash.hash, + }; + let to_key = BlobKey { + account_id, + collection, + document_id: u32::MAX, + hash: blob_hash.hash, + }; + + self.iterate(false, from_key, to_key, true, false, move |acc, key, _| { + let document_id = + key.deserialize_be_u32(BLOB_HASH_LEN + std::mem::size_of::() + 1)?; + if document_ids.contains(document_id) { + *acc = true; + Ok(false) + } else { + Ok(true) + } + }) + .await + } }