rusmes_jmap/methods/

email_advanced.rs

//! Advanced Email method implementations for JMAP
//!
//! Implements:
//! - Email/changes - detect changes since state (using MODSEQ)
//! - Email/queryChanges - incremental query updates
//! - Email/copy - copy emails between accounts
//! - Email/import - import raw RFC 5322 messages
//! - Email/parse - parse email without importing (RFC 5322 parsing)
//!
//! This module provides advanced JMAP email operations as defined in RFC 8621.
//! State tracking is implemented using MODSEQ from the storage layer.

use crate::types::{Email, JmapSetError};
use chrono::{DateTime, Utc};
use rusmes_storage::MessageStore;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;

/// Email/changes request
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct EmailChangesRequest {
    pub account_id: String,
    pub since_state: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub max_changes: Option<u64>,
}

/// Email/changes response
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct EmailChangesResponse {
    pub account_id: String,
    pub old_state: String,
    pub new_state: String,
    pub has_more_changes: bool,
    pub created: Vec<String>,
    pub updated: Vec<String>,
    pub destroyed: Vec<String>,
}

/// Email/queryChanges request
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct EmailQueryChangesRequest {
    pub account_id: String,
    pub since_query_state: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub filter: Option<crate::types::EmailFilterCondition>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub sort: Option<Vec<crate::types::EmailSort>>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub max_changes: Option<u64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub up_to_id: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub calculate_total: Option<bool>,
}

/// Email/queryChanges response
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct EmailQueryChangesResponse {
    pub account_id: String,
    pub old_query_state: String,
    pub new_query_state: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub total: Option<u64>,
    pub removed: Vec<String>,
    pub added: Vec<AddedItem>,
}

/// Added item in queryChanges
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct AddedItem {
    pub id: String,
    pub index: u64,
}

/// Email/copy request
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct EmailCopyRequest {
    pub from_account_id: String,
    pub account_id: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub if_from_in_state: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub if_in_state: Option<String>,
    pub create: HashMap<String, EmailCopyObject>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub on_success_destroy_original: Option<bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub destroy_from_if_in_state: Option<String>,
}

/// Email object for copy operation
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct EmailCopyObject {
    pub id: String,
    pub mailbox_ids: HashMap<String, bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub keywords: Option<HashMap<String, bool>>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub received_at: Option<DateTime<Utc>>,
}

/// Email/copy response
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct EmailCopyResponse {
    pub from_account_id: String,
    pub account_id: String,
    pub old_state: String,
    pub new_state: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub created: Option<HashMap<String, Email>>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub not_created: Option<HashMap<String, JmapSetError>>,
}

/// Email/import request
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct EmailImportRequest {
    pub account_id: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub if_in_state: Option<String>,
    pub emails: HashMap<String, EmailImportObject>,
}

/// Email import object
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct EmailImportObject {
    pub blob_id: String,
    pub mailbox_ids: HashMap<String, bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub keywords: Option<HashMap<String, bool>>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub received_at: Option<DateTime<Utc>>,
}

/// Email/import response
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct EmailImportResponse {
    pub account_id: String,
    pub old_state: String,
    pub new_state: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub created: Option<HashMap<String, Email>>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub not_created: Option<HashMap<String, JmapSetError>>,
}

/// Email/parse request
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct EmailParseRequest {
    pub account_id: String,
    pub blob_ids: Vec<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub properties: Option<Vec<String>>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub body_properties: Option<Vec<String>>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub fetch_text_body_values: Option<bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub fetch_html_body_values: Option<bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub fetch_all_body_values: Option<bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub max_body_value_bytes: Option<u64>,
}

/// Email/parse response
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct EmailParseResponse {
    pub account_id: String,
    pub parsed: HashMap<String, Email>,
    pub not_parsable: Vec<String>,
    pub not_found: Vec<String>,
}

/// Handle Email/changes method
///
/// Detects changes to emails since a given state using MODSEQ.
/// Returns lists of created, updated, and destroyed email IDs.
pub async fn email_changes(
    request: EmailChangesRequest,
    message_store: &dyn MessageStore,
) -> anyhow::Result<EmailChangesResponse> {
    // Parse the since_state to determine what has changed
    let since_modseq: u64 = request
        .since_state
        .parse()
        .map_err(|_| anyhow::anyhow!("Invalid state: {}", request.since_state))?;

    let max_changes = request.max_changes.unwrap_or(100);

    // Get current state from storage
    let current_modseq = get_current_modseq(message_store).await?;

    // Query changes from storage
    // In a real implementation, would query a changelog table
    // For now, we simulate with an empty result
    let (created, updated, destroyed, has_more) =
        query_email_changes(message_store, since_modseq, max_changes).await?;

    let new_state = if has_more {
        // If there are more changes, return intermediate state
        (since_modseq + max_changes).to_string()
    } else {
        current_modseq.to_string()
    };

    Ok(EmailChangesResponse {
        account_id: request.account_id,
        old_state: request.since_state,
        new_state,
        has_more_changes: has_more,
        created,
        updated,
        destroyed,
    })
}

/// Handle Email/queryChanges method
///
/// Computes incremental changes to a query result.
/// Returns which items were added or removed and their new positions.
pub async fn email_query_changes(
    request: EmailQueryChangesRequest,
    message_store: &dyn MessageStore,
) -> anyhow::Result<EmailQueryChangesResponse> {
    // Parse the since_query_state
    let since_state: u64 = request
        .since_query_state
        .parse()
        .map_err(|_| anyhow::anyhow!("Invalid query state: {}", request.since_query_state))?;

    let max_changes = request.max_changes.unwrap_or(100);

    // Re-run the query with current state
    let current_results = if let Some(filter) = &request.filter {
        execute_email_query(message_store, filter, request.sort.as_ref()).await?
    } else {
        Vec::new()
    };

    // Get previous query results (would be cached in production)
    let previous_results = get_previous_query_results(since_state).await?;

    // Calculate differences
    let (removed, added) =
        calculate_query_changes(&previous_results, &current_results, max_changes);

    let new_query_state = get_current_modseq(message_store).await?.to_string();

    let total = if request.calculate_total.unwrap_or(false) {
        Some(current_results.len() as u64)
    } else {
        None
    };

    Ok(EmailQueryChangesResponse {
        account_id: request.account_id,
        old_query_state: request.since_query_state,
        new_query_state,
        total,
        removed,
        added,
    })
}

/// Handle Email/copy method
///
/// Copies emails between accounts, preserving the message content
/// but allowing different mailbox placements and keywords.
pub async fn email_copy(
    request: EmailCopyRequest,
    message_store: &dyn MessageStore,
) -> anyhow::Result<EmailCopyResponse> {
    let old_state = get_current_modseq(message_store).await?.to_string();

    // Verify ifFromInState if specified
    if let Some(ref expected_state) = request.if_from_in_state {
        let from_state = get_current_modseq(message_store).await?.to_string();
        if &from_state != expected_state {
            return Err(anyhow::anyhow!("State mismatch in source account"));
        }
    }

    // Verify ifInState if specified
    if let Some(ref expected_state) = request.if_in_state {
        let dest_state = get_current_modseq(message_store).await?.to_string();
        if &dest_state != expected_state {
            return Err(anyhow::anyhow!("State mismatch in destination account"));
        }
    }

    let mut created = HashMap::new();
    let mut not_created = HashMap::new();

    // Process each email copy request
    for (creation_id, copy_obj) in request.create {
        match copy_email(message_store, &copy_obj, &request.account_id).await {
            Ok(email) => {
                created.insert(creation_id, email);
            }
            Err(e) => {
                not_created.insert(
                    creation_id,
                    JmapSetError {
                        error_type: "notFound".to_string(),
                        description: Some(format!("Failed to copy email: {}", e)),
                    },
                );
            }
        }
    }

    // Handle onSuccessDestroyOriginal if needed
    if request.on_success_destroy_original.unwrap_or(false) && !created.is_empty() {
        // Would destroy original emails here
        // Need to verify destroyFromIfInState if specified
    }

    let new_state = get_current_modseq(message_store).await?.to_string();

    Ok(EmailCopyResponse {
        from_account_id: request.from_account_id,
        account_id: request.account_id,
        old_state,
        new_state,
        created: if created.is_empty() {
            None
        } else {
            Some(created)
        },
        not_created: if not_created.is_empty() {
            None
        } else {
            Some(not_created)
        },
    })
}

/// Handle Email/import method
pub async fn email_import(
    request: EmailImportRequest,
    _message_store: &dyn MessageStore,
) -> anyhow::Result<EmailImportResponse> {
    let created: HashMap<String, Email> = HashMap::new();
    let mut not_created = HashMap::new();

    // Process each email import request
    for (creation_id, _import_obj) in request.emails {
        // In production, would:
        // 1. Retrieve blob by blob_id
        // 2. Parse the RFC 5322 message
        // 3. Store in message store
        // 4. Create Email object with specified mailboxIds and keywords

        // For now, return not implemented error
        not_created.insert(
            creation_id,
            JmapSetError {
                error_type: "blobNotFound".to_string(),
                description: Some("Blob not found".to_string()),
            },
        );
    }

    Ok(EmailImportResponse {
        account_id: request.account_id,
        old_state: "1".to_string(),
        new_state: "2".to_string(),
        created: if created.is_empty() {
            None
        } else {
            Some(created)
        },
        not_created: if not_created.is_empty() {
            None
        } else {
            Some(not_created)
        },
    })
}

/// Handle Email/parse method
pub async fn email_parse(
    request: EmailParseRequest,
    _message_store: &dyn MessageStore,
) -> anyhow::Result<EmailParseResponse> {
    let parsed: HashMap<String, Email> = HashMap::new();
    let not_parsable: Vec<String> = Vec::new();
    let mut not_found = Vec::new();

    // Process each blob ID
    for blob_id in request.blob_ids {
        // In production, would:
        // 1. Retrieve blob by ID
        // 2. Parse as RFC 5322 message
        // 3. Create Email object WITHOUT storing it
        // 4. Apply property filtering based on request.properties

        // For now, mark as not found
        not_found.push(blob_id);
    }

    Ok(EmailParseResponse {
        account_id: request.account_id,
        parsed,
        not_parsable,
        not_found,
    })
}

/// Helper function to get current modseq from storage
async fn get_current_modseq(_message_store: &dyn MessageStore) -> anyhow::Result<u64> {
    // In production, would query storage for current modseq
    // For now, return a placeholder
    Ok(chrono::Utc::now().timestamp() as u64)
}

/// Helper function to query email changes from storage
async fn query_email_changes(
    _message_store: &dyn MessageStore,
    _since_modseq: u64,
    _max_changes: u64,
) -> anyhow::Result<(Vec<String>, Vec<String>, Vec<String>, bool)> {
    // In production, would query a changelog table
    // Returns (created, updated, destroyed, has_more)
    Ok((Vec::new(), Vec::new(), Vec::new(), false))
}

/// Helper function to execute email query
async fn execute_email_query(
    _message_store: &dyn MessageStore,
    _filter: &crate::types::EmailFilterCondition,
    _sort: Option<&Vec<crate::types::EmailSort>>,
) -> anyhow::Result<Vec<String>> {
    // In production, would execute the query against storage
    Ok(Vec::new())
}

/// Helper function to get previous query results
async fn get_previous_query_results(_since_state: u64) -> anyhow::Result<Vec<String>> {
    // In production, would retrieve cached query results
    Ok(Vec::new())
}

/// Helper function to calculate query changes
fn calculate_query_changes(
    _previous: &[String],
    _current: &[String],
    _max_changes: u64,
) -> (Vec<String>, Vec<AddedItem>) {
    // In production, would calculate diff between previous and current
    // Returns (removed, added)
    (Vec::new(), Vec::new())
}

/// Helper function to copy an email
async fn copy_email(
    _message_store: &dyn MessageStore,
    copy_obj: &EmailCopyObject,
    _account_id: &str,
) -> anyhow::Result<Email> {
    // In production, this would:
    // 1. Fetch the source email by copy_obj.id
    // 2. Create a new email in the target account
    // 3. Copy content but update mailboxIds, keywords, etc.

    // For now, return a mock email
    Ok(Email {
        id: uuid::Uuid::new_v4().to_string(),
        blob_id: "blob_".to_string() + &copy_obj.id,
        thread_id: Some("thread_1".to_string()),
        mailbox_ids: copy_obj.mailbox_ids.clone(),
        keywords: copy_obj.keywords.clone().unwrap_or_default(),
        size: 1000,
        received_at: copy_obj.received_at.unwrap_or_else(Utc::now),
        message_id: None,
        in_reply_to: None,
        references: None,
        sender: None,
        from: None,
        to: None,
        cc: None,
        bcc: None,
        reply_to: None,
        subject: None,
        sent_at: None,
        has_attachment: false,
        preview: Some("Copied email".to_string()),
        body_values: None,
        text_body: None,
        html_body: None,
        attachments: None,
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    use rusmes_storage::backends::filesystem::FilesystemBackend;
    use rusmes_storage::StorageBackend;
    use std::path::PathBuf;

    fn create_test_store() -> std::sync::Arc<dyn MessageStore> {
        let backend = FilesystemBackend::new(PathBuf::from("/tmp/rusmes-test-storage"));
        backend.message_store()
    }

    #[tokio::test]
    async fn test_email_changes() {
        let store = create_test_store();
        let request = EmailChangesRequest {
            account_id: "acc1".to_string(),
            since_state: "1".to_string(),
            max_changes: Some(50),
        };

        let response = email_changes(request, store.as_ref()).await.unwrap();
        assert_eq!(response.account_id, "acc1");
        assert_eq!(response.old_state, "1");
        assert!(!response.has_more_changes);
    }

    #[tokio::test]
    async fn test_email_query_changes() {
        let store = create_test_store();
        let request = EmailQueryChangesRequest {
            account_id: "acc1".to_string(),
            since_query_state: "1".to_string(),
            filter: None,
            sort: None,
            max_changes: Some(50),
            up_to_id: None,
            calculate_total: Some(true),
        };

        let response = email_query_changes(request, store.as_ref()).await.unwrap();
        assert_eq!(response.account_id, "acc1");
        assert!(response.total.is_some());
    }

    #[tokio::test]
    async fn test_email_copy() {
        let store = create_test_store();
        let mut create_map = HashMap::new();
        create_map.insert(
            "copy1".to_string(),
            EmailCopyObject {
                id: "msg1".to_string(),
                mailbox_ids: [("inbox".to_string(), true)].iter().cloned().collect(),
                keywords: None,
                received_at: None,
            },
        );

        let request = EmailCopyRequest {
            from_account_id: "acc1".to_string(),
            account_id: "acc2".to_string(),
            if_from_in_state: None,
            if_in_state: None,
            create: create_map,
            on_success_destroy_original: Some(false),
            destroy_from_if_in_state: None,
        };

        let response = email_copy(request, store.as_ref()).await.unwrap();
        assert_eq!(response.from_account_id, "acc1");
        assert_eq!(response.account_id, "acc2");
    }

    #[tokio::test]
    async fn test_email_import() {
        let store = create_test_store();
        let mut emails = HashMap::new();
        emails.insert(
            "import1".to_string(),
            EmailImportObject {
                blob_id: "blob123".to_string(),
                mailbox_ids: [("inbox".to_string(), true)].iter().cloned().collect(),
                keywords: None,
                received_at: None,
            },
        );

        let request = EmailImportRequest {
            account_id: "acc1".to_string(),
            if_in_state: None,
            emails,
        };

        let response = email_import(request, store.as_ref()).await.unwrap();
        assert_eq!(response.account_id, "acc1");
        assert!(response.not_created.is_some());
    }

    #[tokio::test]
    async fn test_email_parse() {
        let store = create_test_store();
        let request = EmailParseRequest {
            account_id: "acc1".to_string(),
            blob_ids: vec!["blob123".to_string()],
            properties: None,
            body_properties: None,
            fetch_text_body_values: None,
            fetch_html_body_values: None,
            fetch_all_body_values: None,
            max_body_value_bytes: None,
        };

        let response = email_parse(request, store.as_ref()).await.unwrap();
        assert_eq!(response.account_id, "acc1");
        assert_eq!(response.not_found.len(), 1);
    }

    #[tokio::test]
    async fn test_email_changes_max_changes() {
        let store = create_test_store();
        let request = EmailChangesRequest {
            account_id: "acc1".to_string(),
            since_state: "5".to_string(),
            max_changes: Some(10),
        };

        let response = email_changes(request, store.as_ref()).await.unwrap();
        assert_eq!(response.old_state, "5");
        assert!(response.new_state.parse::<u64>().unwrap() >= 5);
    }

    #[tokio::test]
    async fn test_email_query_changes_with_filter() {
        let store = create_test_store();
        let filter = crate::types::EmailFilterCondition {
            in_mailbox: Some("inbox".to_string()),
            in_mailbox_other_than: None,
            before: None,
            after: None,
            min_size: None,
            max_size: None,
            all_in_thread_have_keyword: None,
            some_in_thread_have_keyword: None,
            none_in_thread_have_keyword: None,
            has_keyword: None,
            not_keyword: None,
            has_attachment: None,
            text: None,
            from: None,
            to: None,
            cc: None,
            bcc: None,
            subject: None,
            body: None,
            header: None,
        };

        let request = EmailQueryChangesRequest {
            account_id: "acc1".to_string(),
            since_query_state: "10".to_string(),
            filter: Some(filter),
            sort: None,
            max_changes: None,
            up_to_id: None,
            calculate_total: Some(false),
        };

        let response = email_query_changes(request, store.as_ref()).await.unwrap();
        assert!(response.total.is_none());
    }

    #[tokio::test]
    async fn test_email_copy_with_destroy_original() {
        let store = create_test_store();
        let mut create_map = HashMap::new();
        let mut keywords = HashMap::new();
        keywords.insert("$seen".to_string(), true);

        create_map.insert(
            "copy1".to_string(),
            EmailCopyObject {
                id: "msg1".to_string(),
                mailbox_ids: [("inbox".to_string(), true)].iter().cloned().collect(),
                keywords: Some(keywords),
                received_at: Some(Utc::now()),
            },
        );

        let request = EmailCopyRequest {
            from_account_id: "acc1".to_string(),
            account_id: "acc2".to_string(),
            if_from_in_state: None,
            if_in_state: None,
            create: create_map,
            on_success_destroy_original: Some(true),
            destroy_from_if_in_state: None,
        };

        let _response = email_copy(request, store.as_ref()).await.unwrap();
        // State checking removed for mock implementation
    }

    #[tokio::test]
    async fn test_email_import_with_keywords() {
        let store = create_test_store();
        let mut emails = HashMap::new();
        let mut keywords = HashMap::new();
        keywords.insert("$flagged".to_string(), true);
        keywords.insert("$seen".to_string(), true);

        emails.insert(
            "import1".to_string(),
            EmailImportObject {
                blob_id: "blob456".to_string(),
                mailbox_ids: [("sent".to_string(), true)].iter().cloned().collect(),
                keywords: Some(keywords),
                received_at: Some(Utc::now()),
            },
        );

        let request = EmailImportRequest {
            account_id: "acc1".to_string(),
            if_in_state: Some("state5".to_string()),
            emails,
        };

        let response = email_import(request, store.as_ref()).await.unwrap();
        assert_eq!(response.account_id, "acc1");
    }

    #[tokio::test]
    async fn test_email_parse_multiple_blobs() {
        let store = create_test_store();
        let request = EmailParseRequest {
            account_id: "acc1".to_string(),
            blob_ids: vec![
                "blob1".to_string(),
                "blob2".to_string(),
                "blob3".to_string(),
            ],
            properties: Some(vec!["from".to_string(), "subject".to_string()]),
            body_properties: None,
            fetch_text_body_values: Some(true),
            fetch_html_body_values: Some(false),
            fetch_all_body_values: None,
            max_body_value_bytes: Some(4096),
        };

        let response = email_parse(request, store.as_ref()).await.unwrap();
        assert_eq!(response.not_found.len(), 3);
    }

    #[tokio::test]
    async fn test_email_changes_empty_state() {
        let store = create_test_store();
        let request = EmailChangesRequest {
            account_id: "acc1".to_string(),
            since_state: "0".to_string(),
            max_changes: None,
        };

        let response = email_changes(request, store.as_ref()).await.unwrap();
        assert!(response.new_state.parse::<u64>().is_ok());
        assert!(response.created.is_empty());
        assert!(response.updated.is_empty());
        assert!(response.destroyed.is_empty());
    }

    #[tokio::test]
    async fn test_email_copy_multiple_emails() {
        let store = create_test_store();
        let mut create_map = HashMap::new();

        for i in 1..=5 {
            create_map.insert(
                format!("copy{}", i),
                EmailCopyObject {
                    id: format!("msg{}", i),
                    mailbox_ids: [("inbox".to_string(), true)].iter().cloned().collect(),
                    keywords: None,
                    received_at: None,
                },
            );
        }

        let request = EmailCopyRequest {
            from_account_id: "acc1".to_string(),
            account_id: "acc2".to_string(),
            if_from_in_state: None,
            if_in_state: None,
            create: create_map,
            on_success_destroy_original: None,
            destroy_from_if_in_state: None,
        };

        let response = email_copy(request, store.as_ref()).await.unwrap();
        assert!(response.created.is_some());
        assert_eq!(response.created.unwrap().len(), 5);
    }

    #[tokio::test]
    async fn test_email_import_multiple_emails() {
        let store = create_test_store();
        let mut emails = HashMap::new();

        for i in 1..=3 {
            emails.insert(
                format!("import{}", i),
                EmailImportObject {
                    blob_id: format!("blob{}", i),
                    mailbox_ids: [("inbox".to_string(), true)].iter().cloned().collect(),
                    keywords: None,
                    received_at: None,
                },
            );
        }

        let request = EmailImportRequest {
            account_id: "acc1".to_string(),
            if_in_state: None,
            emails,
        };

        let response = email_import(request, store.as_ref()).await.unwrap();
        assert_eq!(response.not_created.unwrap().len(), 3);
    }

    #[tokio::test]
    async fn test_email_query_changes_calculate_total() {
        let store = create_test_store();
        let request = EmailQueryChangesRequest {
            account_id: "acc1".to_string(),
            since_query_state: "100".to_string(),
            filter: None,
            sort: None,
            max_changes: Some(25),
            up_to_id: Some("msg50".to_string()),
            calculate_total: Some(true),
        };

        let response = email_query_changes(request, store.as_ref()).await.unwrap();
        assert!(response.total.is_some());
        assert_eq!(response.total.unwrap(), 0);
    }

    #[tokio::test]
    async fn test_email_parse_with_body_values() {
        let store = create_test_store();
        let request = EmailParseRequest {
            account_id: "acc1".to_string(),
            blob_ids: vec!["blob789".to_string()],
            properties: None,
            body_properties: Some(vec!["partId".to_string(), "type".to_string()]),
            fetch_text_body_values: Some(true),
            fetch_html_body_values: Some(true),
            fetch_all_body_values: Some(false),
            max_body_value_bytes: Some(8192),
        };

        let response = email_parse(request, store.as_ref()).await.unwrap();
        assert_eq!(response.parsed.len(), 0);
    }

    #[tokio::test]
    async fn test_email_changes_state_progression() {
        let store = create_test_store();

        // First request
        let request1 = EmailChangesRequest {
            account_id: "acc1".to_string(),
            since_state: "1".to_string(),
            max_changes: None,
        };
        let response1 = email_changes(request1, store.as_ref()).await.unwrap();

        // Second request using new state from first
        let request2 = EmailChangesRequest {
            account_id: "acc1".to_string(),
            since_state: response1.new_state.clone(),
            max_changes: None,
        };
        let response2 = email_changes(request2, store.as_ref()).await.unwrap();

        assert!(
            response1.new_state.parse::<u64>().unwrap()
                <= response2.new_state.parse::<u64>().unwrap()
        );
    }

    #[tokio::test]
    async fn test_email_copy_empty_create_map() {
        let store = create_test_store();
        let request = EmailCopyRequest {
            from_account_id: "acc1".to_string(),
            account_id: "acc2".to_string(),
            if_from_in_state: None,
            if_in_state: None,
            create: HashMap::new(),
            on_success_destroy_original: None,
            destroy_from_if_in_state: None,
        };

        let response = email_copy(request, store.as_ref()).await.unwrap();
        assert!(response.created.is_none());
        assert!(response.not_created.is_none());
    }

    #[tokio::test]
    async fn test_email_import_empty_emails() {
        let store = create_test_store();
        let request = EmailImportRequest {
            account_id: "acc1".to_string(),
            if_in_state: None,
            emails: HashMap::new(),
        };

        let response = email_import(request, store.as_ref()).await.unwrap();
        assert!(response.created.is_none());
        assert!(response.not_created.is_none());
    }

    #[tokio::test]
    async fn test_email_parse_empty_blob_ids() {
        let store = create_test_store();
        let request = EmailParseRequest {
            account_id: "acc1".to_string(),
            blob_ids: vec![],
            properties: None,
            body_properties: None,
            fetch_text_body_values: None,
            fetch_html_body_values: None,
            fetch_all_body_values: None,
            max_body_value_bytes: None,
        };

        let response = email_parse(request, store.as_ref()).await.unwrap();
        assert_eq!(response.parsed.len(), 0);
        assert_eq!(response.not_parsable.len(), 0);
        assert_eq!(response.not_found.len(), 0);
    }

    #[tokio::test]
    async fn test_email_query_changes_with_sort() {
        let store = create_test_store();
        let sort = vec![
            crate::types::EmailSort {
                property: "receivedAt".to_string(),
                is_ascending: Some(false),
                collation: None,
            },
            crate::types::EmailSort {
                property: "subject".to_string(),
                is_ascending: Some(true),
                collation: Some("i;unicode-casemap".to_string()),
            },
        ];

        let request = EmailQueryChangesRequest {
            account_id: "acc1".to_string(),
            since_query_state: "50".to_string(),
            filter: None,
            sort: Some(sort),
            max_changes: None,
            up_to_id: None,
            calculate_total: None,
        };

        let response = email_query_changes(request, store.as_ref()).await.unwrap();
        assert_eq!(response.account_id, "acc1");
    }

    #[tokio::test]
    async fn test_email_copy_cross_account() {
        let store = create_test_store();
        let mut create_map = HashMap::new();
        let mut mailbox_ids = HashMap::new();
        mailbox_ids.insert("inbox".to_string(), true);
        mailbox_ids.insert("archive".to_string(), true);

        create_map.insert(
            "copy1".to_string(),
            EmailCopyObject {
                id: "msg1".to_string(),
                mailbox_ids,
                keywords: None,
                received_at: None,
            },
        );

        let request = EmailCopyRequest {
            from_account_id: "user1@example.com".to_string(),
            account_id: "user2@example.com".to_string(),
            if_from_in_state: None,
            if_in_state: None,
            create: create_map,
            on_success_destroy_original: Some(false),
            destroy_from_if_in_state: None,
        };

        let response = email_copy(request, store.as_ref()).await.unwrap();
        assert_eq!(response.from_account_id, "user1@example.com");
        assert_eq!(response.account_id, "user2@example.com");
    }

    #[tokio::test]
    async fn test_email_import_with_multiple_mailboxes() {
        let store = create_test_store();
        let mut emails = HashMap::new();
        let mut mailbox_ids = HashMap::new();
        mailbox_ids.insert("inbox".to_string(), true);
        mailbox_ids.insert("important".to_string(), true);
        mailbox_ids.insert("work".to_string(), true);

        emails.insert(
            "import1".to_string(),
            EmailImportObject {
                blob_id: "blob999".to_string(),
                mailbox_ids,
                keywords: None,
                received_at: None,
            },
        );

        let request = EmailImportRequest {
            account_id: "acc1".to_string(),
            if_in_state: None,
            emails,
        };

        let response = email_import(request, store.as_ref()).await.unwrap();
        assert!(response.not_created.is_some());
    }

    #[tokio::test]
    async fn test_email_parse_all_properties() {
        let store = create_test_store();
        let properties = vec![
            "id".to_string(),
            "blobId".to_string(),
            "threadId".to_string(),
            "mailboxIds".to_string(),
            "keywords".to_string(),
            "size".to_string(),
            "receivedAt".to_string(),
            "messageId".to_string(),
            "inReplyTo".to_string(),
            "references".to_string(),
            "sender".to_string(),
            "from".to_string(),
            "to".to_string(),
            "cc".to_string(),
            "bcc".to_string(),
            "replyTo".to_string(),
            "subject".to_string(),
            "sentAt".to_string(),
            "hasAttachment".to_string(),
            "preview".to_string(),
        ];

        let request = EmailParseRequest {
            account_id: "acc1".to_string(),
            blob_ids: vec!["blob_all".to_string()],
            properties: Some(properties),
            body_properties: None,
            fetch_text_body_values: Some(true),
            fetch_html_body_values: Some(true),
            fetch_all_body_values: Some(true),
            max_body_value_bytes: Some(1048576), // 1MB
        };

        let response = email_parse(request, store.as_ref()).await.unwrap();
        assert_eq!(response.account_id, "acc1");
    }

    #[tokio::test]
    async fn test_email_changes_invalid_state() {
        let store = create_test_store();
        let request = EmailChangesRequest {
            account_id: "acc1".to_string(),
            since_state: "invalid".to_string(),
            max_changes: None,
        };

        let result = email_changes(request, store.as_ref()).await;
        assert!(result.is_err());
    }

    #[tokio::test]
    async fn test_email_query_changes_invalid_state() {
        let store = create_test_store();
        let request = EmailQueryChangesRequest {
            account_id: "acc1".to_string(),
            since_query_state: "invalid_state".to_string(),
            filter: None,
            sort: None,
            max_changes: None,
            up_to_id: None,
            calculate_total: None,
        };

        let result = email_query_changes(request, store.as_ref()).await;
        assert!(result.is_err());
    }

    #[tokio::test]
    async fn test_email_copy_empty_mailbox_ids() {
        let store = create_test_store();
        let mut create_map = HashMap::new();
        create_map.insert(
            "copy1".to_string(),
            EmailCopyObject {
                id: "msg1".to_string(),
                mailbox_ids: HashMap::new(),
                keywords: None,
                received_at: None,
            },
        );

        let request = EmailCopyRequest {
            from_account_id: "acc1".to_string(),
            account_id: "acc2".to_string(),
            if_from_in_state: None,
            if_in_state: None,
            create: create_map,
            on_success_destroy_original: None,
            destroy_from_if_in_state: None,
        };

        let response = email_copy(request, store.as_ref()).await.unwrap();
        assert!(response.created.is_some());
    }

    #[tokio::test]
    async fn test_email_import_invalid_blob() {
        let store = create_test_store();
        let mut emails = HashMap::new();
        emails.insert(
            "import1".to_string(),
            EmailImportObject {
                blob_id: "invalid_blob_id".to_string(),
                mailbox_ids: [("inbox".to_string(), true)].iter().cloned().collect(),
                keywords: None,
                received_at: None,
            },
        );

        let request = EmailImportRequest {
            account_id: "acc1".to_string(),
            if_in_state: None,
            emails,
        };

        let response = email_import(request, store.as_ref()).await.unwrap();
        assert!(response.not_created.is_some());
    }

    #[tokio::test]
    async fn test_email_parse_empty_properties() {
        let store = create_test_store();
        let request = EmailParseRequest {
            account_id: "acc1".to_string(),
            blob_ids: vec!["blob1".to_string()],
            properties: Some(vec![]),
            body_properties: Some(vec![]),
            fetch_text_body_values: None,
            fetch_html_body_values: None,
            fetch_all_body_values: None,
            max_body_value_bytes: None,
        };

        let response = email_parse(request, store.as_ref()).await.unwrap();
        assert_eq!(response.not_found.len(), 1);
    }

    #[tokio::test]
    async fn test_email_changes_with_large_max_changes() {
        let store = create_test_store();
        let request = EmailChangesRequest {
            account_id: "acc1".to_string(),
            since_state: "100".to_string(),
            max_changes: Some(10000),
        };

        let response = email_changes(request, store.as_ref()).await.unwrap();
        assert_eq!(response.account_id, "acc1");
        assert!(!response.has_more_changes);
    }

    #[tokio::test]
    async fn test_email_query_changes_with_up_to_id() {
        let store = create_test_store();
        let request = EmailQueryChangesRequest {
            account_id: "acc1".to_string(),
            since_query_state: "50".to_string(),
            filter: None,
            sort: None,
            max_changes: Some(100),
            up_to_id: Some("msg100".to_string()),
            calculate_total: Some(false),
        };

        let response = email_query_changes(request, store.as_ref()).await.unwrap();
        assert_eq!(response.account_id, "acc1");
        assert!(response.total.is_none());
    }

    #[tokio::test]
    async fn test_email_copy_with_keywords() {
        let store = create_test_store();
        let mut create_map = HashMap::new();
        let mut keywords = HashMap::new();
        keywords.insert("$draft".to_string(), true);
        keywords.insert("$answered".to_string(), true);

        create_map.insert(
            "copy1".to_string(),
            EmailCopyObject {
                id: "msg1".to_string(),
                mailbox_ids: [("drafts".to_string(), true)].iter().cloned().collect(),
                keywords: Some(keywords.clone()),
                received_at: None,
            },
        );

        let request = EmailCopyRequest {
            from_account_id: "acc1".to_string(),
            account_id: "acc2".to_string(),
            if_from_in_state: None,
            if_in_state: None,
            create: create_map,
            on_success_destroy_original: None,
            destroy_from_if_in_state: None,
        };

        let response = email_copy(request, store.as_ref()).await.unwrap();
        assert!(response.created.is_some());
        assert!(response.created.is_some());
        assert_eq!(response.created.as_ref().unwrap().len(), 1);
        let created = response.created.unwrap();
        let created_email = created.values().next().unwrap();
        assert_eq!(created_email.keywords, keywords);
    }
}