blobs

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<Option<Vec<u8>>> {
+        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
+        }
+    }
+}

crates/core/src/blob/mod.rs

@@ -0,0 +1 @@
+pub mod get;

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<ImportEmailResponse, MethodError> {
-        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::<Vec<_>>();
+            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
+            },
+        })
     }
 }

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<IngestedEmail> for Object<Value> {
+    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))
+    }
+}

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<store::Error> for MaybeError {
         }
     }
 }
+
+impl From<MaybeError> for MethodError {
+    fn from(value: MaybeError) -> Self {
+        match value {
+            MaybeError::Temporary(msg) => {
+                let log = "true";
+                MethodError::ServerPartialFail
+            }
+            MaybeError::Permanent(msg) => MethodError::InvalidArguments(msg),
+        }
+    }
+}

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<MaybeReference<Vec<MaybeReference<Id, String>>, ResultReference>>,
+    pub mailbox_ids: MaybeReference<Vec<MaybeReference<Id, String>>, ResultReference>,
     pub keywords: Vec<Keyword>,
     pub received_at: Option<UTCDate>,
 }
@@ -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(
                             <SetValueMap<MaybeReference<Id, String>>>::parse(parser)?.values,
-                        ))
+                        )
                     } else {
-                        Some(MaybeReference::Reference(ResultReference::parse(parser)?))
+                        MaybeReference::Reference(ResultReference::parse(parser)?)
                     };
                 }
                 0x7364_726f_7779_656b if !property.is_ref => {

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)) => {
+                        MaybeReference::Reference(rr) => {
-                            email.mailbox_ids = Some(MaybeReference::Value(
+                            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)?);
                                 }
                             }
                         }
-                        _ => (),
                     }
                 }
             }

crates/protocol/src/types/date.rs

@@ -243,6 +243,12 @@ impl Serialize for UTCDate {
     }
 }
 
+impl From<UTCDate> for u64 {
+    fn from(value: UTCDate) -> Self {
+        value.timestamp() as u64
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use crate::{parser::json::Parser, types::date::UTCDate};

crates/store/src/backend/foundationdb/mod.rs

@@ -80,9 +80,9 @@ impl<T: AsRef<[u8]>> Serialize for &BlobKey<T> {
         KeySerializer::new(std::mem::size_of::<BlobKey<T>>() + 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()
     }
 }

crates/store/src/backend/sqlite/mod.rs

@@ -79,9 +79,9 @@ impl<T: AsRef<[u8]>> Serialize for &BlobKey<T> {
         let hash = self.hash.as_ref();
         KeySerializer::new(std::mem::size_of::<BlobKey<T>>() + 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()
     }
 }

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<u32>,
+    ) -> crate::Result<bool> {
+        // 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<u8>,
+        document_ids: RoaringBitmap,
+    ) -> crate::Result<bool> {
+        // 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::<u32>() + 1)?;
+            if document_ids.contains(document_id) {
+                *acc = true;
+                Ok(false)
+            } else {
+                Ok(true)
+            }
+        })
+        .await
+    }
 }