diff --git a/src/modules/account/migration.rs b/src/modules/account/migration.rs index ec48633..af0abef 100644 --- a/src/modules/account/migration.rs +++ b/src/modules/account/migration.rs @@ -35,7 +35,10 @@ use crate::{ cache::imap::mailbox::MailBox, database::{list_all_impl, secondary_find_impl, with_transaction}, error::BichonResult, - indexer::manager::{EML_INDEX_MANAGER, ENVELOPE_INDEX_MANAGER}, + indexer::{ + attachment::ATTACHMENT_INDEX_MANAGER, eml::EML_INDEX_MANAGER, + manager::ENVELOPE_INDEX_MANAGER, + }, users::{role::DEFAULT_ACCOUNT_MANAGER_ROLE_ID, UserModel, DEFAULT_ADMIN_USER_ID}, }, utc_now, @@ -355,17 +358,15 @@ impl AccountV4 { if matches!(account.account_type, AccountType::IMAP) { SYNC_TASKS.stop(account.id).await?; AccountRunningState::delete(account.id).await?; - //BICHON_CONTEXT.clean_account(account.id).await?; } OAuth2AccessToken::try_delete(account.id).await?; UserModel::cleanup_account(account.id).await?; MailBox::clean(account.id).await?; - ENVELOPE_INDEX_MANAGER - .delete_account_envelopes(account.id) - .await?; - EML_INDEX_MANAGER + let content_hashes = ENVELOPE_INDEX_MANAGER .delete_account_envelopes(account.id) .await?; + EML_INDEX_MANAGER.delete(&content_hashes).await?; + ATTACHMENT_INDEX_MANAGER.delete(&content_hashes).await?; Self::delete_account(account.id).await?; info!("Sequential cleanup completed for account: {}", account.id); Ok(()) diff --git a/src/modules/cache/imap/sync/rebuild.rs b/src/modules/cache/imap/sync/rebuild.rs index 98885d7..d309adc 100644 --- a/src/modules/cache/imap/sync/rebuild.rs +++ b/src/modules/cache/imap/sync/rebuild.rs @@ -27,7 +27,10 @@ use crate::{ SEMAPHORE, }, error::{code::ErrorCode, BichonError, BichonResult}, - indexer::manager::{EML_INDEX_MANAGER, ENVELOPE_INDEX_MANAGER}, + indexer::{ + attachment::ATTACHMENT_INDEX_MANAGER, eml::EML_INDEX_MANAGER, + manager::ENVELOPE_INDEX_MANAGER, + }, }, raise_error, }; @@ -194,12 +197,15 @@ pub async fn rebuild_mailbox_cache( local_mailbox: &MailBox, remote_mailbox: &MailBox, ) -> BichonResult<()> { - ENVELOPE_INDEX_MANAGER - .delete_mailbox_envelopes(account.id, vec![local_mailbox.id]) - .await?; - EML_INDEX_MANAGER + let content_hashes = ENVELOPE_INDEX_MANAGER .delete_mailbox_envelopes(account.id, vec![local_mailbox.id]) .await?; + + if !content_hashes.is_empty() { + EML_INDEX_MANAGER.delete(&content_hashes).await?; + ATTACHMENT_INDEX_MANAGER.delete(&content_hashes).await?; + } + if remote_mailbox.exists == 0 { info!( "Account {}: Mailbox '{}' has no emails on the remote server. The mailbox is empty, no envelopes to fetch.", @@ -225,12 +231,15 @@ pub async fn rebuild_mailbox_cache_by_date( remote: &MailBox, direction: FetchDirection, ) -> BichonResult<()> { - ENVELOPE_INDEX_MANAGER - .delete_mailbox_envelopes(account.id, vec![local_mailbox_id]) - .await?; - EML_INDEX_MANAGER + let content_hashes = ENVELOPE_INDEX_MANAGER .delete_mailbox_envelopes(account.id, vec![local_mailbox_id]) .await?; + + if !content_hashes.is_empty() { + EML_INDEX_MANAGER.delete(&content_hashes).await?; + ATTACHMENT_INDEX_MANAGER.delete(&content_hashes).await?; + } + if remote.exists == 0 { info!( "Account {}: Mailbox '{}' has no emails on the remote server. The mailbox is empty, no envelopes to fetch.", diff --git a/src/modules/duckdb/build.rs b/src/modules/duckdb/build.rs index 3b2a4f4..3e6869b 100644 --- a/src/modules/duckdb/build.rs +++ b/src/modules/duckdb/build.rs @@ -59,6 +59,7 @@ pub fn build_record_batch(items: &[Envelope]) -> RecordBatch { Field::new("message_id", DataType::Utf8, true), Field::new("has_attachment", DataType::Boolean, false), Field::new("attachment_count", DataType::Int32, false), + Field::new("regular_attachment_count", DataType::Int32, false), Field::new( "tags", DataType::List(Arc::new(Field::new("item", DataType::Utf8, true))), @@ -88,6 +89,7 @@ pub fn build_record_batch(items: &[Envelope]) -> RecordBatch { let mut msg_id_b = StringBuilder::with_capacity(capacity, capacity * 30); let mut has_att_b = BooleanArray::builder(capacity); let mut att_count_b = Int32Array::builder(capacity); + let mut regular_att_count_b = Int32Array::builder(capacity); let mut tags_b = ListBuilder::new(StringBuilder::new()); let mut shard_id_b = UInt64Array::builder(capacity); @@ -100,30 +102,26 @@ pub fn build_record_batch(items: &[Envelope]) -> RecordBatch { subject_b.append_value(&e.subject); body_b.append_value(&e.text); from_b.append_value(&e.from); - for addr in &e.to { to_b.values().append_value(addr); } to_b.append(true); - for addr in &e.cc { cc_b.values().append_value(addr); } cc_b.append(true); - for addr in &e.bcc { bcc_b.values().append_value(addr); } bcc_b.append(true); - date_b.append_value(e.date); internal_date_b.append_value(e.internal_date); size_b.append_value(e.size as u64); thread_id_b.append_value(&e.thread_id); msg_id_b.append_value(&e.message_id); - - has_att_b.append_value(e.attachment_count > 0); + has_att_b.append_value(e.regular_attachment_count > 0); att_count_b.append_value(e.attachment_count as i32); + regular_att_count_b.append_value(e.regular_attachment_count as i32); tags_b.append(true); shard_id_b.append_value(DEFAULT_SHARD_ID); } @@ -149,116 +147,10 @@ pub fn build_record_batch(items: &[Envelope]) -> RecordBatch { Arc::new(msg_id_b.finish()), Arc::new(has_att_b.finish()), Arc::new(att_count_b.finish()), + Arc::new(regular_att_count_b.finish()), Arc::new(tags_b.finish()), Arc::new(shard_id_b.finish()), ], ) .expect("Failed to build RecordBatch") } - -#[cfg(test)] -mod integration_tests { - use super::*; - use duckdb::{Connection, Result}; - - #[test] - fn test_envelope_ingestion_and_query() -> Result<()> { - let conn = Connection::open_in_memory()?; - conn.execute_batch( - r#" - CREATE TABLE IF NOT EXISTS envelopes ( - id UUID PRIMARY KEY, - account_id UBIGINT NOT NULL, - mailbox_id UBIGINT NOT NULL, - uid UBIGINT NOT NULL, - - content_hash VARCHAR(64), - - subject TEXT, - body TEXT, - - sender TEXT, - recipients VARCHAR[], - cc VARCHAR[], - bcc VARCHAR[], - - sent_at BIGINT, - received_at BIGINT, - - size_bytes UBIGINT, - thread_id VARCHAR, - message_id TEXT, - - has_attachment BOOLEAN NOT NULL, - attachment_count INTEGER NOT NULL CHECK (attachment_count >= 0), - tags VARCHAR[], - shard_id UBIGINT NOT NULL - ); - "#, - )?; - - let test_uuid = uuid::Uuid::new_v4().to_string(); - let test_hash = - "a1b2c3d4e5f6g7h8i9j0k1l2m3n4o5p6q7r8s9t0u1v2w3x4y5z6a1b2c3d4e5f6".to_string(); - - let items = vec![Envelope { - id: test_uuid.clone(), - account_id: 1, - mailbox_id: 1, - uid: 50, - content_hash: test_hash.clone(), - subject: "Testing Arrow".to_string(), - text: "Content".to_string(), - from: "sender@test.com".to_string(), - to: vec!["user1@test.com".to_string(), "user2@test.com".to_string()], - cc: vec!["manager@test.com".to_string()], - bcc: vec![], - date: 1000, - internal_date: 1001, - size: 2048, - thread_id: "1".to_string(), - message_id: "id123".to_string(), - attachment_count: 1, - tags: None, - account_email: None, - mailbox_name: None, - }]; - - { - let batch = build_record_batch(&items); - let mut appender = conn.appender("envelopes")?; - appender.append_record_batch(batch)?; - appender.flush()?; - } - - let mut stmt = - conn.prepare("SELECT subject, size_bytes, content_hash FROM envelopes WHERE id = ?")?; - let mut rows = stmt.query([&test_uuid])?; - - if let Some(row) = rows.next()? { - let subject: String = row.get(0)?; - let size: u64 = row.get(1)?; - let hash_in_db: String = row.get(2)?; - - assert_eq!(subject, "Testing Arrow"); - assert_eq!(size, 2048); - assert_eq!(hash_in_db, test_hash); - } - - let mut stmt = conn.prepare( - "SELECT count(*) FROM envelopes WHERE list_contains(\"recipients\", 'user2@test.com')", - )?; - let count: i64 = stmt.query_row([], |r| r.get(0))?; - assert_eq!(count, 1); - - let has_att: bool = conn.query_row( - "SELECT has_attachment FROM envelopes WHERE id = ?", - [&test_uuid], - |r| r.get(0), - )?; - assert!(has_att); - - println!("Integration test with content_hash passed!"); - Ok(()) - } -} diff --git a/src/modules/duckdb/init.rs b/src/modules/duckdb/init.rs index 809d143..1cbd7bc 100644 --- a/src/modules/duckdb/init.rs +++ b/src/modules/duckdb/init.rs @@ -33,12 +33,12 @@ use crate::{ duckdb::{build::build_record_batch, refinery::DuckDBConnection}, error::{code::ErrorCode, BichonResult}, indexer::{ - envelope::Envelope, - manager::{EML_INDEX_MANAGER, ENVELOPE_INDEX_MANAGER}, + attachment::ATTACHMENT_INDEX_MANAGER, eml::EML_INDEX_MANAGER, envelope::Envelope, + manager::ENVELOPE_INDEX_MANAGER, }, message::{ attachment::AttachmentMetadata, - content::AttachmentInfo, + content::{AttachmentDetail, AttachmentInfo}, search::{SearchFilter, SortBy}, tags::TagCount, }, @@ -133,90 +133,154 @@ impl DuckDBManager { Ok(()) } - pub fn delete_envelopes_by_account(&self, account_id: u64) -> BichonResult { + pub fn delete_account_envelopes_with_orphans( + &self, + account_id: u64, + ) -> BichonResult> { let mut conn = self.conn()?; let tx = conn .transaction() - .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; + .map_err(|e| raise_error!(e.to_string(), ErrorCode::InternalError))?; + + let orphan_sql = r#" + WITH target_hashes AS ( + SELECT content_hash FROM envelopes WHERE account_id = ? + UNION + SELECT content_hash FROM envelope_attachments WHERE account_id = ? + ), + active_hashes AS ( + SELECT content_hash FROM envelopes WHERE account_id != ? + UNION + SELECT content_hash FROM envelope_attachments WHERE account_id != ? + ) + SELECT content_hash FROM target_hashes + EXCEPT + SELECT content_hash FROM active_hashes + "#; + + let mut stmt = tx + .prepare(orphan_sql) + .map_err(|e| raise_error!(e.to_string(), ErrorCode::InternalError))?; + + let orphan_hashes: Vec = stmt + .query_map([account_id, account_id, account_id, account_id], |row| { + row.get::<_, String>(0) + }) + .map_err(|e| raise_error!(e.to_string(), ErrorCode::InternalError))? + .collect::, _>>() + .map_err(|e| raise_error!(e.to_string(), ErrorCode::InternalError))?; tx.execute( - "DELETE FROM envelope_attachments WHERE account_id = ?;", - params![account_id], + "DELETE FROM envelope_attachments WHERE account_id = ?", + [account_id], ) - .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; + .map_err(|e| raise_error!(e.to_string(), ErrorCode::InternalError))?; - let rows_deleted = tx - .execute( - "DELETE FROM envelopes WHERE account_id = ?;", - params![account_id], - ) - .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; + let count = tx + .execute("DELETE FROM envelopes WHERE account_id = ?", [account_id]) + .map_err(|e| raise_error!(e.to_string(), ErrorCode::InternalError))?; tx.commit() - .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; + .map_err(|e| raise_error!(e.to_string(), ErrorCode::InternalError))?; - if rows_deleted > 0 { + if count > 0 { tracing::info!( - "Account cleanup: removed {} emails and their attachments for account {}", - rows_deleted, - account_id + "Account {} data cleared. Deleted {} envelopes, {} orphan hashes identified.", + account_id, + count, + orphan_hashes.len() ); } - Ok(rows_deleted) + Ok(orphan_hashes) } - pub fn delete_mailbox_envelopes( + pub fn delete_mailbox_envelopes_with_orphans( &self, account_id: u64, mailbox_ids: Vec, - ) -> BichonResult<()> { + ) -> BichonResult> { if mailbox_ids.is_empty() { - return Ok(()); + return Ok(vec![]); } + let mut conn = self.conn()?; let tx = conn .transaction() .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; - { - let placeholders = mailbox_ids - .iter() - .map(|_| "?") - .collect::>() - .join(", "); - let mut sql_params: Vec = vec![account_id.into()]; - sql_params.extend(mailbox_ids.iter().map(|&id| duckdb::types::Value::from(id))); - let param_iter = duckdb::params_from_iter(sql_params); + let placeholders = mailbox_ids + .iter() + .map(|_| "?") + .collect::>() + .join(", "); + let mut sql_params: Vec = vec![account_id.into()]; + sql_params.extend(mailbox_ids.iter().map(|&id| duckdb::types::Value::from(id))); + let param_iter = duckdb::params_from_iter(sql_params); - let delete_att_sql = format!( - "DELETE FROM envelope_attachments WHERE account_id = ? AND mailbox_id IN ({})", - placeholders - ); - tx.execute(&delete_att_sql, param_iter.clone()) - .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; + let orphan_sql = format!( + r#" + WITH target_hashes AS ( + SELECT content_hash FROM envelopes WHERE account_id = ? AND mailbox_id IN ({0}) + UNION + SELECT content_hash FROM envelope_attachments WHERE account_id = ? AND mailbox_id IN ({0}) + ), + active_hashes AS ( + SELECT content_hash FROM envelopes WHERE NOT (account_id = ? AND mailbox_id IN ({0})) + UNION + SELECT content_hash FROM envelope_attachments WHERE NOT (account_id = ? AND mailbox_id IN ({0})) + ) + SELECT content_hash FROM target_hashes + EXCEPT + SELECT content_hash FROM active_hashes + "#, + placeholders + ); - let delete_env_sql = format!( - "DELETE FROM envelopes WHERE account_id = ? AND mailbox_id IN ({})", - placeholders - ); - let count = tx - .execute(&delete_env_sql, param_iter) - .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; - - if count > 0 { - tracing::info!( - "Deleted {} emails and their associated attachments for account: {}, mailboxes: {:?}", - count, - account_id, - mailbox_ids - ); - } + let mut query_params: Vec = Vec::new(); + for _ in 0..4 { + query_params.push(account_id.into()); + query_params.extend(mailbox_ids.iter().map(|&id| duckdb::types::Value::from(id))); } + let mut stmt = tx + .prepare(&orphan_sql) + .map_err(|e| raise_error!(e.to_string(), ErrorCode::InternalError))?; + let orphan_hashes: Vec = stmt + .query_map(duckdb::params_from_iter(query_params), |row| { + row.get::<_, String>(0) + }) + .map_err(|e| raise_error!(e.to_string(), ErrorCode::InternalError))? + .collect::, _>>() + .map_err(|e| raise_error!(e.to_string(), ErrorCode::InternalError))?; + + let delete_att_sql = format!( + "DELETE FROM envelope_attachments WHERE account_id = ? AND mailbox_id IN ({})", + placeholders + ); + tx.execute(&delete_att_sql, param_iter.clone()) + .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; + + let delete_env_sql = format!( + "DELETE FROM envelopes WHERE account_id = ? AND mailbox_id IN ({})", + placeholders + ); + let count = tx + .execute(&delete_env_sql, param_iter) + .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; + tx.commit() .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; - Ok(()) + + if count > 0 { + tracing::info!( + "Deleted {} emails. Found {} orphan hashes to clean up.", + count, + orphan_hashes.len() + ); + } + + Ok(orphan_hashes) } pub fn append_envelopes_with_attachments( @@ -245,12 +309,15 @@ impl DuckDBManager { env.account_id, env.mailbox_id, att.filename, + att.is_message, + att.inline, + att.content_id, att.get_extension(), att.get_category(), att.file_type.to_ascii_lowercase(), att.size as u64, - env.content_hash.clone(), // It's the hash of the attachment content itself, not the hash of the full email. - 0, + att.content_hash, + 0 ]) .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; } @@ -329,6 +396,46 @@ impl DuckDBManager { Ok(max_uid) } + pub fn get_attachments_by_envelope_id( + &self, + account_id: u64, + envelope_id: String, + ) -> BichonResult> { + let conn = self.conn()?; + + let mut stmt = conn + .prepare( + "SELECT * FROM envelope_attachments + WHERE account_id = ? AND envelope_id = ?;", + ) + .map_err(|e| { + raise_error!( + format!("Prepare failed: {:#?}", e), + ErrorCode::InternalError + ) + })?; + + let attachment_iter = stmt + .query_map(params![account_id, envelope_id], |row| { + AttachmentDetail::from_row(row) + }) + .map_err(|e| { + raise_error!(format!("Query failed: {:#?}", e), ErrorCode::InternalError) + })?; + + let mut attachments = Vec::new(); + for att_res in attachment_iter { + attachments.push(att_res.map_err(|e| { + raise_error!( + format!("Row mapping failed: {:#?}", e), + ErrorCode::InternalError + ) + })?); + } + + Ok(attachments) + } + pub fn get_envelope_by_id( &self, account_id: u64, @@ -558,9 +665,9 @@ impl DuckDBManager { let conn = self.conn()?; let mut sql = r#" SELECT - CAST(array_agg(DISTINCT extension) AS JSON) AS extensions, - CAST(array_agg(DISTINCT ext_category) AS JSON) AS categories, - CAST(array_agg(DISTINCT content_type) AS JSON) AS content_types + COALESCE(CAST(array_agg(DISTINCT extension) FILTER (WHERE extension IS NOT NULL) AS JSON), '[]') AS extensions, + COALESCE(CAST(array_agg(DISTINCT ext_category) AS JSON), '[]') AS categories, + COALESCE(CAST(array_agg(DISTINCT content_type) AS JSON), '[]') AS content_types FROM envelope_attachments "# .to_string(); @@ -583,9 +690,10 @@ impl DuckDBManager { let result = stmt .query_row(duckdb::params_from_iter(params_vec), |row| { - let exts_raw: String = row.get(0)?; - let cats_raw: String = row.get(1)?; - let ctypes_raw: String = row.get(2)?; + let exts_raw: String = row.get(0).unwrap_or_else(|_| "[]".to_string()); + let cats_raw: String = row.get(1).unwrap_or_else(|_| "[]".to_string()); + let ctypes_raw: String = row.get(2).unwrap_or_else(|_| "[]".to_string()); + let exts: Vec = serde_json::from_str(&exts_raw).unwrap_or_default(); let cats: Vec = serde_json::from_str(&cats_raw).unwrap_or_default(); let ctypes: Vec = serde_json::from_str(&ctypes_raw).unwrap_or_default(); @@ -628,8 +736,8 @@ impl DuckDBManager { let del_attachments_query = format!( "DELETE FROM envelope_attachments - WHERE account_id = ? - AND envelope_id IN ({})", + WHERE account_id = ? + AND envelope_id IN ({})", placeholders ); @@ -643,8 +751,8 @@ impl DuckDBManager { let query = format!( "DELETE FROM envelopes - WHERE account_id = ? - AND id IN ({})", + WHERE account_id = ? + AND id IN ({})", placeholders ); @@ -659,6 +767,115 @@ impl DuckDBManager { Ok(()) } + pub fn get_orphan_hashes_in_memory( + &self, + deletes: HashMap>, + ) -> BichonResult> { + let conn = self.conn()?; + let all_delete_ids: Vec = deletes.values().flatten().cloned().collect(); + if all_delete_ids.is_empty() { + return Ok(vec![]); + } + + if all_delete_ids.len() > 100 { + return Err(raise_error!( + "Too many IDs for batch delete, please shrink the batch".into(), + ErrorCode::InvalidParameter + )); + } + + let mut target_hashes = HashSet::new(); + let placeholders = vec!["?"; all_delete_ids.len()].join(", "); + let params = duckdb::params_from_iter(&all_delete_ids); + + let mut stmt = conn + .prepare(&format!( + "SELECT content_hash FROM envelopes WHERE id IN ({})", + placeholders + )) + .map_err(|e| raise_error!(e.to_string(), ErrorCode::InternalError))?; + + let rows = stmt + .query_map(params.clone(), |r| r.get::<_, String>(0)) + .map_err(|e| raise_error!(e.to_string(), ErrorCode::InternalError))?; + + for h in rows { + target_hashes + .insert(h.map_err(|e| raise_error!(e.to_string(), ErrorCode::InternalError))?); + } + + let mut stmt = conn + .prepare(&format!( + "SELECT content_hash FROM envelope_attachments WHERE envelope_id IN ({})", + placeholders + )) + .map_err(|e| raise_error!(e.to_string(), ErrorCode::InternalError))?; + + let rows = stmt + .query_map(params, |r| r.get::<_, String>(0)) + .map_err(|e| raise_error!(e.to_string(), ErrorCode::InternalError))?; + for h in rows { + target_hashes + .insert(h.map_err(|e| raise_error!(e.to_string(), ErrorCode::InternalError))?); + } + + if target_hashes.is_empty() { + return Ok(vec![]); + } + + let hash_placeholders = vec!["?"; target_hashes.len()].join(", "); + let mut still_used_hashes = HashSet::new(); + + let mut check_params: Vec> = Vec::new(); + for id in &all_delete_ids { + check_params.push(Box::new(id.clone())); + } + for hash in &target_hashes { + check_params.push(Box::new(hash.clone())); + } + let check_params_refs: Vec<&dyn duckdb::ToSql> = + check_params.iter().map(|p| p.as_ref()).collect(); + + let mut stmt = conn + .prepare(&format!( + "SELECT DISTINCT content_hash FROM envelopes WHERE id NOT IN ({}) AND content_hash IN ({})", + placeholders, hash_placeholders + )) + .map_err(|e| raise_error!(e.to_string(), ErrorCode::InternalError))?; + + let rows = stmt + .query_map(duckdb::params_from_iter(&check_params_refs), |r| { + r.get::<_, String>(0) + }) + .map_err(|e| raise_error!(e.to_string(), ErrorCode::InternalError))?; + for h in rows { + still_used_hashes + .insert(h.map_err(|e| raise_error!(e.to_string(), ErrorCode::InternalError))?); + } + + let mut stmt = conn.prepare(&format!( + "SELECT DISTINCT content_hash FROM envelope_attachments WHERE envelope_id NOT IN ({}) AND content_hash IN ({})", + placeholders, hash_placeholders + )).map_err(|e| raise_error!(e.to_string(), ErrorCode::InternalError))?; + + let rows = stmt + .query_map(duckdb::params_from_iter(&check_params_refs), |r| { + r.get::<_, String>(0) + }) + .map_err(|e| raise_error!(e.to_string(), ErrorCode::InternalError))?; + for h in rows { + still_used_hashes + .insert(h.map_err(|e| raise_error!(e.to_string(), ErrorCode::InternalError))?); + } + + let orphans: Vec = target_hashes + .into_iter() + .filter(|h| !still_used_hashes.contains(h)) + .collect(); + + Ok(orphans) + } + pub fn top_10_largest_emails( &self, accounts: Option>, @@ -943,24 +1160,36 @@ impl DuckDBManager { } else { tracing::warn!(account_id, "account not found in top accounts query"); tokio::spawn(async move { - if let Err(e) = ENVELOPE_INDEX_MANAGER + let content_hashes = match ENVELOPE_INDEX_MANAGER .delete_account_envelopes(account_id) .await { - tracing::error!( - account_id = account_id, - error = %e, - "failed to cleanup envelope index" - ); - } - if let Err(e) = EML_INDEX_MANAGER.delete_account_envelopes(account_id).await - { + Ok(content_hashes) => content_hashes, + Err(e) => { + tracing::error!( + account_id = account_id, + error = %e, + "failed to cleanup envelope index" + ); + return; + } + }; + + if let Err(e) = EML_INDEX_MANAGER.delete(&content_hashes).await { tracing::error!( account_id = account_id, error = %e, "failed to cleanup eml index" ); } + + if let Err(e) = ATTACHMENT_INDEX_MANAGER.delete(&content_hashes).await { + tracing::error!( + account_id = account_id, + error = %e, + "failed to cleanup attachment index" + ); + } }); } } @@ -1342,19 +1571,19 @@ impl DuckDBManager { base_sql.push_str(" AND a.filename ILIKE ? "); args.push(format!("%{}%", name).into()); } - + // Normalized to lowercase at write-time. LIKE is sufficient here instead of ILIKE. if let Some(ext) = filter.attachment_extension { - base_sql.push_str(" AND a.extension ILIKE ? "); + base_sql.push_str(" AND a.extension LIKE ? "); args.push(format!("%{}%", ext).into()); } - + // Normalized to lowercase at write-time. LIKE is sufficient here instead of ILIKE. if let Some(cat) = filter.attachment_category { - base_sql.push_str(" AND a.ext_category ILIKE ? "); + base_sql.push_str(" AND a.ext_category LIKE ? "); args.push(format!("%{}%", cat).into()); } - + // Normalized to lowercase at write-time. LIKE is sufficient here instead of ILIKE. if let Some(ctype) = filter.attachment_content_type { - base_sql.push_str(" AND a.content_type ILIKE ? "); + base_sql.push_str(" AND a.content_type LIKE ? "); args.push(format!("%{}%", ctype).into()); } diff --git a/src/modules/duckdb/migrations/V1__initial.sql b/src/modules/duckdb/migrations/V1__initial.sql index e14e3df..9d98706 100644 --- a/src/modules/duckdb/migrations/V1__initial.sql +++ b/src/modules/duckdb/migrations/V1__initial.sql @@ -36,11 +36,11 @@ CREATE TABLE IF NOT EXISTS envelopes ( -- attachment summary has_attachment BOOLEAN NOT NULL, attachment_count INTEGER NOT NULL CHECK (attachment_count >= 0), + regular_attachment_count INTEGER NOT NULL CHECK (regular_attachment_count >= 0), tags VARCHAR[], shard_id UBIGINT NOT NULL ); -CREATE INDEX IF NOT EXISTS idx_env_mailbox_sent ON envelopes(account_id, mailbox_id, sent_at); -- ========================= -- envelope_attachments -- @@ -52,10 +52,12 @@ CREATE TABLE IF NOT EXISTS envelope_attachments ( account_id UBIGINT NOT NULL, mailbox_id UBIGINT NOT NULL, -- Original attachment filename (for display) - filename TEXT NOT NULL, - + filename TEXT, + is_message BOOLEAN NOT NULL DEFAULT FALSE, + is_inline BOOLEAN NOT NULL DEFAULT FALSE, + cid TEXT, -- Normalized file extension (lowercase, without dot) - extension TEXT NOT NULL, + extension TEXT, -- Extension category (document / image / archive / ...) ext_category TEXT NOT NULL, @@ -64,7 +66,5 @@ CREATE TABLE IF NOT EXISTS envelope_attachments ( -- 0 if unknown size_bytes UBIGINT NOT NULL, content_hash VARCHAR(64) NOT NULL, - shard_id UBIGINT NOT NULL -); - -CREATE INDEX IF NOT EXISTS idx_attachments_env_id ON envelope_attachments (envelope_id); \ No newline at end of file + shard_id UINTEGER NOT NULL +); \ No newline at end of file diff --git a/src/modules/envelope/extractor.rs b/src/modules/envelope/extractor.rs index 8804b49..5b8c789 100644 --- a/src/modules/envelope/extractor.rs +++ b/src/modules/envelope/extractor.rs @@ -20,20 +20,26 @@ use crate::modules::common::AddrVec; use crate::modules::envelope::utils::normalize_subject; use crate::modules::error::code::ErrorCode; use crate::modules::error::BichonResult; +use crate::modules::indexer::attachment::ATTACHMENT_INDEX_MANAGER; +use crate::modules::indexer::eml::EML_INDEX_MANAGER; +use crate::modules::indexer::manager::ENVELOPE_INDEX_MANAGER; +use crate::modules::indexer::schema::SchemaTools; use crate::modules::message::content::AttachmentInfo; use crate::modules::utils::html::extract_text; -use crate::modules::utils::{content_hash, hex_hash}; +use crate::modules::utils::{compute_content_hash, hex_hash}; use crate::{id, modules::indexer::envelope::Envelope}; use crate::{raise_error, utc_now}; use async_imap::types::Fetch; use mail_parser::{Address, HeaderName, Message, MessageParser, MimeHeaders}; +use tantivy::doc; +use tracing::error; use uuid::Uuid; -pub fn extract_envelope( +pub async fn extract_envelope_and_store_it( fetch: &Fetch, account_id: u64, mailbox_id: u64, -) -> BichonResult<(Envelope, Vec)> { +) -> BichonResult<()> { let internal_date = fetch .internal_date() .map(|d| d.timestamp_millis()) @@ -43,30 +49,22 @@ pub fn extract_envelope( .body() .ok_or_else(|| raise_error!("No body available".into(), ErrorCode::InternalError))?; let size = fetch.size.unwrap_or(body.len() as u32); - - extract_envelope_core(body, uid, size, internal_date, account_id, mailbox_id) + extract_envelope_core(body, uid, size, internal_date, account_id, mailbox_id).await } -pub fn extract_envelope_from_eml( +pub async fn extract_envelope_from_eml( body: &[u8], account_id: u64, mailbox_id: u64, -) -> BichonResult<(Envelope, Vec)> { - extract_envelope_core(body, 0, body.len() as u32, 0, account_id, mailbox_id).map( - |(mut env, att)| { - if env.internal_date == 0 { - env.internal_date = env.date; - } - (env, att) - }, - ) +) -> BichonResult<()> { + extract_envelope_core(body, 0, body.len() as u32, 0, account_id, mailbox_id).await } -pub fn extract_envelope_from_smtp( +pub async fn extract_envelope_from_smtp( body: &[u8], account_id: u64, mailbox_id: u64, -) -> BichonResult<(Envelope, Vec)> { +) -> BichonResult<()> { extract_envelope_core( body, 0, @@ -75,18 +73,19 @@ pub fn extract_envelope_from_smtp( account_id, mailbox_id, ) + .await } -fn extract_envelope_core( +async fn extract_envelope_core( body: &[u8], uid: u32, size: u32, internal_date: i64, account_id: u64, mailbox_id: u64, -) -> BichonResult<(Envelope, Vec)> { - let content_hash = content_hash(body); - let message = MessageParser::new().parse(body).ok_or_else(|| { +) -> BichonResult<()> { + let email_content_hash = compute_content_hash(body); + let message: Message<'_> = MessageParser::new().parse(body).ok_or_else(|| { raise_error!( "Email header parse result is not available".into(), ErrorCode::InternalError @@ -116,7 +115,11 @@ fn extract_envelope_core( } let date = message.date().map(|d| d.to_timestamp() * 1000).unwrap_or(0); - + let internal_date = if internal_date == 0 { + date + } else { + internal_date + }; let parse_addrs = |addrs: Option<&Address<'_>>| { addrs .map(|addr| { @@ -138,42 +141,13 @@ fn extract_envelope_core( .and_then(|addr| AddrVec::from(addr).0.into_iter().next()) .and_then(|add| add.address) .unwrap_or_else(|| "unknown".to_string()); - let attachments: Vec = message - .attachments() - .filter_map(|attachment| { - let content_id = attachment.content_id().map(Into::into); - let inline = attachment - .content_disposition() - .map(|d| d.is_inline()) - .unwrap_or(false); - if inline && content_id.is_some() { - return None; - } - - let file_type = attachment - .content_type() - .map(|ct| { - format!( - "{}/{}", - ct.c_type.as_ref(), - ct.c_subtype.as_deref().unwrap_or("") - ) - }) - .unwrap_or_else(|| "application/octet-stream".to_string()); - //注意:有些附件是没有名字的，这样extension也就不存在，那么在获取附件的时候，就不能通过name定位 - Some(AttachmentInfo { - filename: attachment - .attachment_name() - .map(|name| name.to_string()) - .unwrap_or_default(), - size: attachment.contents().len(), - inline, - file_type, - content_id, - }) - }) - .collect(); + let attachment_count = message.attachment_count(); + let attachments = detach_and_store_attachments(body, &message, &email_content_hash).await; + let inline_with_id_count = attachments + .iter() + .filter(|a| a.inline && a.content_id.is_some()) + .count(); let envelope = Envelope { id: Uuid::new_v4().to_string(), message_id, @@ -190,17 +164,20 @@ fn extract_envelope_core( internal_date, size, thread_id, - attachment_count: attachments.len(), + attachment_count, + regular_attachment_count: attachment_count - inline_with_id_count, tags: None, account_email: None, mailbox_name: None, - content_hash, + content_hash: email_content_hash, }; - - Ok((envelope, attachments)) + ENVELOPE_INDEX_MANAGER + .add_document((envelope, attachments)) + .await; + Ok(()) } -pub fn extract_envelope_from_message( +pub fn extract_envelope_from_nested_message( message: Message<'_>, account_id: u64, ) -> BichonResult { @@ -267,6 +244,7 @@ pub fn extract_envelope_from_message( size: Default::default(), thread_id, attachment_count: Default::default(), + regular_attachment_count: Default::default(), tags: Default::default(), account_email: Default::default(), mailbox_name: Default::default(), @@ -303,6 +281,175 @@ fn extract_references(message: &Message<'_>) -> Option> { } } +pub async fn detach_and_store_attachments( + original_body: &[u8], + message: &Message<'_>, + eml_content_hash: &str, +) -> Vec { + let mut stripped_eml = original_body.to_vec(); + let mut attachment_infos = Vec::new(); + // Step 1: Collect and sort attachment ranges in reverse to maintain offset integrity + let mut ranges: Vec<_> = message + .attachments() + .map(|att| { + ( + att.raw_body_offset() as usize, + att.raw_end_offset() as usize, + att, + ) + }) + .collect(); + + ranges.sort_by(|a, b| b.0.cmp(&a.0)); + + let fields = SchemaTools::fields(); + for (raw_start, raw_end, att) in ranges { + // Step 2: Extract raw bytes and store them as standalone documents + let raw_bytes = &original_body[raw_start..raw_end]; + let content_hash = compute_content_hash(raw_bytes); + + ATTACHMENT_INDEX_MANAGER + .add_document( + content_hash.clone(), + doc!( + fields.f_id => content_hash.clone(), + fields.f_blob => raw_bytes + ), + ) + .await; + // Step 3: Replace raw attachment content with a hash-based placeholder + let placeholder = format!("<>", &content_hash); + let p_bytes = placeholder.as_bytes(); + stripped_eml.splice(raw_start..raw_end, p_bytes.iter().cloned()); + + let info = AttachmentInfo { + filename: att.attachment_name().map(|n| n.to_string()), + size: att.contents().len(), + inline: att + .content_disposition() + .map(|d| d.is_inline()) + .unwrap_or(false), + file_type: att + .content_type() + .map(|ct| { + format!( + "{}/{}", + ct.c_type.as_ref(), + ct.c_subtype.as_deref().unwrap_or("") + ) + }) + .unwrap_or_else(|| "application/octet-stream".to_string()), + content_id: att.content_id().map(|id| id.to_string()), + content_hash: content_hash.clone(), + is_message: att.is_message(), + }; + + attachment_infos.push(info); + } + // Step 4: Store the final stripped EML content + EML_INDEX_MANAGER + .add_document( + eml_content_hash.to_string(), + doc!( + fields.f_id => eml_content_hash.to_string(), + fields.f_blob => stripped_eml + ), + ) + .await; + + attachment_infos +} + +pub async fn reattach_eml_content( + account_id: u64, + envelope_id: String, +) -> BichonResult<(Envelope, Vec)> { + let envelope = ENVELOPE_INDEX_MANAGER + .get_envelope_by_id(account_id, envelope_id.clone()) + .await? + .ok_or_else(|| { + raise_error!( + format!( + "Envelope not found: account_id={} envelope_id={}", + account_id, &envelope_id + ), + ErrorCode::ResourceNotFound + ) + })?; + + let mut restored_eml = EML_INDEX_MANAGER + .get(&envelope.content_hash) + .await? + .ok_or_else(|| { + raise_error!( + format!( + "Original email content not found: account_id={} envelope_id={} content_hash={}", + account_id, &envelope_id, &envelope.content_hash + ), + ErrorCode::ResourceNotFound + ) + })?; + + if !envelope.has_attachments() { + return Ok((envelope, restored_eml)); + } + + let account_detail = ENVELOPE_INDEX_MANAGER + .get_attachments_by_envelope_id(account_id, envelope_id) + .await?; + + if envelope.attachment_count != account_detail.len() { + return Err(raise_error!( + "Consistency check failed: attachment_count does not match account_detail length" + .into(), + ErrorCode::InternalError + )); + } + + let mut tasks = Vec::new(); + for detail in account_detail { + let placeholder_str = format!("<>", &detail.info.content_hash); + let pattern = placeholder_str.as_bytes(); + let pattern_len = pattern.len(); + + let mut search_cursor = 0; + while let Some(pos) = restored_eml[search_cursor..] + .windows(pattern_len) + .position(|window| window == pattern) + { + let absolute_start = search_cursor + pos; + let absolute_end = absolute_start + pattern_len; + + tasks.push(( + absolute_start, + absolute_end, + detail.info.content_hash.clone(), + )); + search_cursor = absolute_end; + } + } + + tasks.sort_by(|a, b| b.0.cmp(&a.0)); + + for (start, end, hash) in tasks { + if let Some(original_data) = ATTACHMENT_INDEX_MANAGER.get(&hash).await? { + let actual_hash = compute_content_hash(&original_data); + if actual_hash != hash { + error!( + "[ERROR] Content Hash Mismatch! Expected: {}, Actual: {}", + hash, actual_hash + ); + continue; + } + restored_eml.splice(start..end, original_data.iter().cloned()); + } else { + error!("[ERROR] Missing attachment blob for hash: {}", hash); + } + } + + Ok((envelope, restored_eml)) +} + #[cfg(test)] mod test { use html2text::config; diff --git a/src/modules/imap/executor.rs b/src/modules/imap/executor.rs index 817cf93..8ada3e6 100644 --- a/src/modules/imap/executor.rs +++ b/src/modules/imap/executor.rs @@ -20,18 +20,15 @@ use crate::modules::account::migration::AccountModel; use crate::modules::account::state::AccountRunningState; use crate::modules::cache::imap::mailbox::MailBox; use crate::modules::cache::imap::sync::flow::{generate_uid_sequence_hashset, DEFAULT_BATCH_SIZE}; -use crate::modules::envelope::extractor::extract_envelope; +use crate::modules::envelope::extractor::extract_envelope_and_store_it; use crate::modules::error::code::ErrorCode; use crate::modules::imap::session::SessionStream; -use crate::modules::indexer::manager::{EML_INDEX_MANAGER, ENVELOPE_INDEX_MANAGER}; -use crate::modules::indexer::schema::SchemaTools; use crate::modules::{error::BichonResult, imap::manager::ImapConnectionManager}; use crate::raise_error; use async_imap::types::Name; use async_imap::Session; use futures::TryStreamExt; use std::collections::HashSet; -use tantivy::doc; use tracing::info; const BODY_FETCH_COMMAND: &str = "(UID INTERNALDATE RFC822.SIZE BODY.PEEK[])"; @@ -200,19 +197,12 @@ impl ImapExecutor { .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::ImapCommandFailed))?; let mut count = 0; - let fields = SchemaTools::fields(); while let Some(fetch) = stream .try_next() .await .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::ImapCommandFailed))? { - let envelope = extract_envelope(&fetch, account_id, mailbox_id)?; - let content_hash = envelope.0.content_hash.clone(); - ENVELOPE_INDEX_MANAGER.add_document(envelope).await; - let body = fetch.body().ok_or_else(|| { - raise_error!("missing a body".into(), ErrorCode::ImapUnexpectedResult) - })?; - EML_INDEX_MANAGER.add_document( content_hash.clone(), doc!(fields.f_id => content_hash, fields.f_account_id => account_id, fields.f_mailbox_id => mailbox_id, fields.f_blob => body)).await; + extract_envelope_and_store_it(&fetch, account_id, mailbox_id).await?; count += 1; } Ok(count) @@ -234,19 +224,12 @@ impl ImapExecutor { .uid_fetch(uid_set, BODY_FETCH_COMMAND) .await .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::ImapCommandFailed))?; - let fields = SchemaTools::fields(); while let Some(fetch) = stream .try_next() .await .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::ImapCommandFailed))? { - let envelope = extract_envelope(&fetch, account_id, mailbox_id)?; - let content_hash = envelope.0.content_hash.clone(); - ENVELOPE_INDEX_MANAGER.add_document(envelope).await; - let body = fetch.body().ok_or_else(|| { - raise_error!("missing a body".into(), ErrorCode::ImapUnexpectedResult) - })?; - EML_INDEX_MANAGER.add_document( content_hash.clone(), doc!(fields.f_id => content_hash, fields.f_account_id => account_id, fields.f_mailbox_id => mailbox_id, fields.f_blob => body)).await; + extract_envelope_and_store_it(&fetch, account_id, mailbox_id).await?; } Ok(()) } diff --git a/src/modules/imap/tests.rs b/src/modules/imap/tests.rs index b761ea3..f9fe96b 100644 --- a/src/modules/imap/tests.rs +++ b/src/modules/imap/tests.rs @@ -179,3 +179,25 @@ async fn test_bulk_attachment_stripping_blake3() { println!("✅ All attachments replaced successfully from back to front."); } + +#[tokio::test] +async fn test_667() { + let path = r"C:\Users\polly\Downloads\test777.eml"; + let input = std::fs::read(path).expect("Failed to read EML file"); + + let message = MessageParser::default() + .parse(&input) + .expect("Failed to parse EML"); + + for att in message.attachments() { + println!("name: {:#?}", att.attachment_name()); + println!("content_type: {:#?}", att.content_type()); + println!("is_message: {:#?}", att.is_message()); + println!("content_disposition: {:#?}", att.content_disposition()); + println!( + "content_transfer_encoding: {:#?}", + att.content_transfer_encoding() + ); + println!("content_id: {:#?}", att.content_id()); + } +} diff --git a/src/modules/import/mod.rs b/src/modules/import/mod.rs index 37f35e0..19b0d9c 100644 --- a/src/modules/import/mod.rs +++ b/src/modules/import/mod.rs @@ -27,11 +27,7 @@ use crate::{ account::migration::{AccountModel, AccountType}, cache::imap::mailbox::{Attribute, AttributeEnum, MailBox}, envelope::extractor::extract_envelope_from_eml, - error::{code::ErrorCode, BichonResult}, - indexer::{ - manager::{EML_INDEX_MANAGER, ENVELOPE_INDEX_MANAGER}, - schema::SchemaTools, - }, + error::{BichonResult, code::ErrorCode}, utils::create_hash, }, raise_error, @@ -112,7 +108,6 @@ impl ImportEmls { }, }; - let fields = SchemaTools::fields(); let account_id = account.id; let mut success_count = 0; let mut failed_details: Vec = Vec::new(); // Store failure details @@ -133,8 +128,10 @@ impl ImportEmls { } }; - let envelope = match extract_envelope_from_eml(&decoded, account_id, mailbox_id) { - Ok(env) => env, + match extract_envelope_from_eml(&decoded, account_id, mailbox_id).await { + Ok(_) => { + success_count += 1; + }, Err(e) => { let error_msg = format!( "Failed to extract envelope from EML at index {}: {:?}", @@ -148,24 +145,6 @@ impl ImportEmls { continue; } }; - let content_hash = envelope.0.content_hash.clone(); - ENVELOPE_INDEX_MANAGER - .add_document(envelope) - .await; - - EML_INDEX_MANAGER - .add_document( - content_hash.clone(), - doc!( - fields.f_id => content_hash, - fields.f_account_id => account_id, - fields.f_mailbox_id => mailbox_id, - fields.f_blob => decoded - ), - ) - .await; - - success_count += 1; } let failed_count = failed_details.len(); diff --git a/src/modules/indexer/attachment.rs b/src/modules/indexer/attachment.rs new file mode 100644 index 0000000..79bd36d --- /dev/null +++ b/src/modules/indexer/attachment.rs @@ -0,0 +1,299 @@ +// +// Copyright (c) 2025-2026 rustmailer.com (https://rustmailer.com) +// +// This file is part of the Bichon Email Archiving Project +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +use std::{ + collections::HashMap, + path::PathBuf, + sync::{Arc, LazyLock}, + time::Duration, +}; + +use crate::modules::{indexer::DocumentOp, settings::cli::SETTINGS}; +use crate::{ + modules::{ + common::signal::SIGNAL_MANAGER, + error::{code::ErrorCode, BichonResult}, + indexer::schema::SchemaTools, + settings::dir::DATA_DIR_MANAGER, + }, + raise_error, +}; + +use tantivy::indexer::{NoMergePolicy, UserOperation}; +use tantivy::{ + collector::TopDocs, + query::TermQuery, + schema::{IndexRecordOption, Value}, + store::Compressor, + Index, IndexBuilder, IndexReader, IndexSettings, IndexWriter, TantivyDocument, Term, +}; +use tokio::{ + sync::{mpsc, Mutex}, + task, +}; +use tracing::info; + +pub const ATTACHMENT_BATCH_SIZE: usize = 10; +const MAX_BUFFER_DURATION: Duration = Duration::from_secs(10); + +pub static ATTACHMENT_INDEX_MANAGER: LazyLock = + LazyLock::new(AttachmentManager::new); + +pub struct AttachmentManager { + index_writer: Arc>, + sender: mpsc::Sender, + reader: IndexReader, +} + +impl AttachmentManager { + pub fn new() -> Self { + let index = Self::open_or_create_index(&DATA_DIR_MANAGER.attachment_dir); + + let writer: IndexWriter = index + .writer_with_num_threads( + SETTINGS.bichon_tantivy_threads as usize, + SETTINGS.bichon_tantivy_buffer_size, + ) + .unwrap_or_else(|e| { + panic!( + "Failed to create IndexWriter (threads: {}, buffer: {}B) for {:?}: {}", + SETTINGS.bichon_tantivy_threads, + SETTINGS.bichon_tantivy_buffer_size, + DATA_DIR_MANAGER.attachment_dir, + e + ) + }); + + writer.set_merge_policy(Box::new(NoMergePolicy)); + let index_writer = Arc::new(Mutex::new(writer)); + + let reader = index.reader().unwrap_or_else(|e| { + panic!( + "Failed to create IndexReader for {:?}: {}", + DATA_DIR_MANAGER.eml_dir, e + ) + }); + let (sender, mut receiver) = mpsc::channel::(100); + task::spawn(async move { + let mut buffer: HashMap = + HashMap::with_capacity(ATTACHMENT_BATCH_SIZE); + let mut interval = tokio::time::interval(MAX_BUFFER_DURATION); + let mut shutdown = SIGNAL_MANAGER.subscribe(); + loop { + tokio::select! { + maybe_msg = receiver.recv() => { + match maybe_msg { + Some(DocumentOp::Document((eid, doc))) => { + buffer.insert(eid, doc); + if buffer.len() >= ATTACHMENT_BATCH_SIZE { + ATTACHMENT_INDEX_MANAGER.drain_and_commit(&mut buffer).await; + } + } + Some(DocumentOp::Shutdown) => { + ATTACHMENT_INDEX_MANAGER.drain_and_commit(&mut buffer).await; + break; + } + None => break, + } + } + _ = interval.tick() => { + if !buffer.is_empty() { + ATTACHMENT_INDEX_MANAGER.drain_and_commit(&mut buffer).await; + } + } + _ = shutdown.recv() => { + let _ = ATTACHMENT_INDEX_MANAGER.sender.send(DocumentOp::Shutdown).await; + } + } + } + }); + Self { + index_writer, + sender, + reader, + } + } + + pub async fn add_document(&self, content_hash: String, doc: TantivyDocument) { + let _ = self + .sender + .send(DocumentOp::Document((content_hash, doc))) + .await; + } + + fn open_or_create_index(index_dir: &PathBuf) -> Index { + let need_create = !index_dir.exists() + || index_dir + .read_dir() + .map(|mut d| d.next().is_none()) + .unwrap_or(true); + + if need_create { + info!( + "Attachment storage not found or empty, creating new attachment storage at {}", + index_dir.display() + ); + std::fs::create_dir_all(&index_dir).unwrap_or_else(|e| { + panic!("Failed to create index directory {:?}: {}", index_dir, e) + }); + IndexBuilder::new() + .schema(SchemaTools::schema()) + .settings(IndexSettings { + docstore_compression: Compressor::None, + docstore_compress_dedicated_thread: Default::default(), + docstore_blocksize: Default::default(), + }) + .create_in_dir(&index_dir) + .unwrap_or_else(|e| panic!("Failed to create index in {:?}: {}", index_dir, e)) + } else { + info!( + "Opening existing attachment data storage at {}", + index_dir.display() + ); + open(&index_dir) + } + } + + fn term(&self, content_hash: &str) -> Term { + Term::from_field_text(SchemaTools::fields().f_id, content_hash) + } + + pub async fn get(&self, content_hash: &str) -> BichonResult>> { + let searcher = self.reader.searcher(); + let term = Term::from_field_text(SchemaTools::fields().f_id, content_hash); + let query = TermQuery::new(term, IndexRecordOption::Basic); + let docs = searcher + .search(&query, &TopDocs::with_limit(1)) + .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; + + if docs.is_empty() { + return Ok(None); + } + + let (_, doc_address) = docs.first().unwrap(); + let doc: TantivyDocument = searcher + .doc_async(*doc_address) + .await + .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; + let fields = SchemaTools::fields(); + let value = doc.get_first(fields.f_blob).ok_or_else(|| { + raise_error!( + format!("miss '{}' field in tantivy document", stringify!(field)), + ErrorCode::InternalError + ) + })?; + let bytes = value.as_bytes().ok_or_else(|| { + raise_error!( + format!("'{}' field is not a bytes", stringify!(field)), + ErrorCode::InternalError + ) + })?; + + Ok(Some(bytes.to_vec())) + } + + pub async fn delete( + &self, + content_hashes: &Vec, // HashMap + ) -> BichonResult<()> { + if content_hashes.is_empty() { + tracing::warn!("deletes is empty, nothing to delete"); + return Ok(()); + } + + let mut writer = self.index_writer.lock().await; + for hash in content_hashes { + let term = self.term(hash); + writer.delete_term(term); + } + + writer + .commit() + .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; + + Ok(()) + } + + async fn drain_and_commit(&self, buffer: &mut HashMap) { + if buffer.is_empty() { + return; + } + let mut writer = self.index_writer.lock().await; + let mut operations = Vec::new(); + + for (content_hash, doc) in buffer.drain() { + let delete_term = Term::from_field_text(SchemaTools::fields().f_id, &content_hash); + operations.push(UserOperation::Delete(delete_term)); + operations.push(UserOperation::Add(doc)); + } + if let Err(e) = writer.run(operations) { + eprintln!("[FATAL] Tantivy run failed: {e:?}"); + std::process::exit(1); + } + + fatal_commit(&mut writer); + } +} + +fn fatal_commit(writer: &mut IndexWriter) { + const MAX_RETRIES: usize = 3; + const RETRY_DELAY_MS: u64 = 1000; + + for attempt in 0..=MAX_RETRIES { + match writer.commit() { + Ok(_) => { + if attempt > 0 { + eprintln!("[INFO] Commit succeeded on attempt {}", attempt + 1); + } + return; + } + Err(e) => match &e { + tantivy::TantivyError::IoError(io_error) => { + if attempt < MAX_RETRIES { + eprintln!( + "[WARN] Commit failed (attempt {}/{}): {:?}. Retrying in {}ms...", + attempt + 1, + MAX_RETRIES + 1, + io_error, + RETRY_DELAY_MS * (attempt as u64 + 1) + ); + std::thread::sleep(std::time::Duration::from_millis( + RETRY_DELAY_MS * (attempt as u64 + 1), + )); + } else { + eprintln!( + "[FATAL] Tantivy commit failed after {} attempts: {:?}", + MAX_RETRIES + 1, + io_error + ); + std::process::exit(1); + } + } + _ => { + eprintln!("[FATAL] Tantivy commit failed with non-IO error: {e:?}"); + std::process::exit(1); + } + }, + } + } +} + +fn open(index_dir: &PathBuf) -> Index { + Index::open_in_dir(index_dir) + .unwrap_or_else(|e| panic!("Failed to open index in {:?}: {}", index_dir, e)) +} diff --git a/src/modules/indexer/eml.rs b/src/modules/indexer/eml.rs new file mode 100644 index 0000000..cf0ce22 --- /dev/null +++ b/src/modules/indexer/eml.rs @@ -0,0 +1,331 @@ +// +// Copyright (c) 2025-2026 rustmailer.com (https://rustmailer.com) +// +// This file is part of the Bichon Email Archiving Project +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +use std::{ + collections::HashMap, + path::PathBuf, + sync::{Arc, LazyLock}, + time::Duration, +}; + +use crate::modules::{ + envelope::extractor::reattach_eml_content, indexer::DocumentOp, settings::cli::SETTINGS, +}; +use crate::{ + modules::{ + common::signal::SIGNAL_MANAGER, + error::{code::ErrorCode, BichonResult}, + indexer::schema::SchemaTools, + settings::dir::DATA_DIR_MANAGER, + }, + raise_error, +}; +use tantivy::indexer::{NoMergePolicy, UserOperation}; +use tantivy::{ + collector::TopDocs, + query::TermQuery, + schema::{IndexRecordOption, Value}, + store::{Compressor, ZstdCompressor}, + Index, IndexBuilder, IndexReader, IndexSettings, IndexWriter, TantivyDocument, Term, +}; +use tokio::{ + fs::File, + io::AsyncWriteExt, + sync::{mpsc, Mutex}, + task, +}; +use tracing::info; + +pub const EML_BATCH_SIZE: usize = 100; +const MAX_BUFFER_DURATION: Duration = Duration::from_secs(10); + +pub static EML_INDEX_MANAGER: LazyLock = LazyLock::new(EmlIndexManager::new); + +pub struct EmlIndexManager { + index_writer: Arc>, + sender: mpsc::Sender, + reader: IndexReader, +} + +impl EmlIndexManager { + pub fn new() -> Self { + let index = Self::open_or_create_index(&DATA_DIR_MANAGER.eml_dir); + + let writer: IndexWriter = index + .writer_with_num_threads( + SETTINGS.bichon_tantivy_threads as usize, + SETTINGS.bichon_tantivy_buffer_size, + ) + .unwrap_or_else(|e| { + panic!( + "Failed to create IndexWriter (threads: {}, buffer: {}B) for {:?}: {}", + SETTINGS.bichon_tantivy_threads, + SETTINGS.bichon_tantivy_buffer_size, + DATA_DIR_MANAGER.eml_dir, + e + ) + }); + + writer.set_merge_policy(Box::new(NoMergePolicy)); + let index_writer = Arc::new(Mutex::new(writer)); + + let reader = index.reader().unwrap_or_else(|e| { + panic!( + "Failed to create IndexReader for {:?}: {}", + DATA_DIR_MANAGER.eml_dir, e + ) + }); + let (sender, mut receiver) = mpsc::channel::(100); + task::spawn(async move { + let mut buffer: HashMap = + HashMap::with_capacity(EML_BATCH_SIZE); + let mut interval = tokio::time::interval(MAX_BUFFER_DURATION); + let mut shutdown = SIGNAL_MANAGER.subscribe(); + loop { + tokio::select! { + maybe_msg = receiver.recv() => { + match maybe_msg { + Some(DocumentOp::Document((eid, doc))) => { + buffer.insert(eid, doc); + if buffer.len() >= EML_BATCH_SIZE { + EML_INDEX_MANAGER.drain_and_commit(&mut buffer).await; + } + } + Some(DocumentOp::Shutdown) => { + EML_INDEX_MANAGER.drain_and_commit(&mut buffer).await; + break; + } + None => break, + } + } + _ = interval.tick() => { + if !buffer.is_empty() { + EML_INDEX_MANAGER.drain_and_commit(&mut buffer).await; + } + } + _ = shutdown.recv() => { + let _ = EML_INDEX_MANAGER.sender.send(DocumentOp::Shutdown).await; + } + } + } + }); + Self { + index_writer, + sender, + reader, + } + } + + /// Adds a document to the indexer. + /// + /// # Parameters + /// - `eid`: A hash derived from **Account ID + Message ID**. + /// This acts as a unique identifier for the EML content itself. + /// + /// - `doc`: The `TantivyDocument` representing the mail body/content. + /// + /// # Logical Design + /// Unlike the `envelope_id` (which is a hash of Account + Folder + Message ID), + /// this `eid` ignores the folder context. This ensures that while metadata + /// (envelopes) can be duplicated across different folders, the physical + /// EML/document storage remains de-duplicated and unique. + pub async fn add_document(&self, content_hash: String, doc: TantivyDocument) { + let _ = self + .sender + .send(DocumentOp::Document((content_hash, doc))) + .await; + } + + fn open_or_create_index(index_dir: &PathBuf) -> Index { + let need_create = !index_dir.exists() + || index_dir + .read_dir() + .map(|mut d| d.next().is_none()) + .unwrap_or(true); + + if need_create { + info!( + "Email storage not found or empty, creating new mail storage at {}", + index_dir.display() + ); + std::fs::create_dir_all(&index_dir).unwrap_or_else(|e| { + panic!("Failed to create index directory {:?}: {}", index_dir, e) + }); + IndexBuilder::new() + .schema(SchemaTools::schema()) + .settings(IndexSettings { + docstore_compression: Compressor::Zstd(ZstdCompressor { + compression_level: Some(SETTINGS.bichon_eml_compression_level as i32), + }), + docstore_compress_dedicated_thread: true, + docstore_blocksize: SETTINGS.bichon_eml_blocksize, + }) + .create_in_dir(&index_dir) + .unwrap_or_else(|e| panic!("Failed to create index in {:?}: {}", index_dir, e)) + } else { + info!("Opening existing email storage at {}", index_dir.display()); + open(&index_dir) + } + } + + fn term(&self, content_hash: &str) -> Term { + Term::from_field_text(SchemaTools::fields().f_id, content_hash) + } + + pub async fn get(&self, content_hash: &str) -> BichonResult>> { + let searcher = self.reader.searcher(); + let term = Term::from_field_text(SchemaTools::fields().f_id, content_hash); + let query = TermQuery::new(term, IndexRecordOption::Basic); + let docs = searcher + .search(&query, &TopDocs::with_limit(1)) + .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; + + if docs.is_empty() { + return Ok(None); + } + + let (_, doc_address) = docs.first().unwrap(); + let doc: TantivyDocument = searcher + .doc_async(*doc_address) + .await + .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; + let fields = SchemaTools::fields(); + let value = doc.get_first(fields.f_blob).ok_or_else(|| { + raise_error!( + format!("miss '{}' field in tantivy document", stringify!(field)), + ErrorCode::InternalError + ) + })?; + let bytes = value.as_bytes().ok_or_else(|| { + raise_error!( + format!("'{}' field is not a bytes", stringify!(field)), + ErrorCode::InternalError + ) + })?; + + Ok(Some(bytes.to_vec())) + } + + pub async fn get_reader(&self, account_id: u64, eid: String) -> BichonResult { + let (envelope, data) = reattach_eml_content(account_id, eid).await?; + let mut path = DATA_DIR_MANAGER.temp_dir.clone(); + path.push(format!("{}.eml", envelope.content_hash)); + { + let mut file = File::create(&path) + .await + .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; + file.write_all(&data) + .await + .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; + } + let file = File::open(&path) + .await + .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; + Ok(file) + } + + pub async fn delete( + &self, + content_hashes: &Vec, // HashMap + ) -> BichonResult<()> { + if content_hashes.is_empty() { + tracing::warn!("delete_email_multi_account: deletes is empty, nothing to delete"); + return Ok(()); + } + + let mut writer = self.index_writer.lock().await; + for hash in content_hashes { + let term = self.term(hash); + writer.delete_term(term); + } + writer + .commit() + .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; + + Ok(()) + } + + // Deduplicate directly by content_hash, regardless of the account. + async fn drain_and_commit(&self, buffer: &mut HashMap) { + if buffer.is_empty() { + return; + } + let mut writer = self.index_writer.lock().await; + let mut operations = Vec::new(); + + for (content_hash, doc) in buffer.drain() { + let delete_term = Term::from_field_text(SchemaTools::fields().f_id, &content_hash); + operations.push(UserOperation::Delete(delete_term)); + operations.push(UserOperation::Add(doc)); + } + if let Err(e) = writer.run(operations) { + eprintln!("[FATAL] Tantivy run failed: {e:?}"); + std::process::exit(1); + } + + fatal_commit(&mut writer); + } +} + +fn fatal_commit(writer: &mut IndexWriter) { + const MAX_RETRIES: usize = 3; + const RETRY_DELAY_MS: u64 = 1000; + + for attempt in 0..=MAX_RETRIES { + match writer.commit() { + Ok(_) => { + if attempt > 0 { + eprintln!("[INFO] Commit succeeded on attempt {}", attempt + 1); + } + return; + } + Err(e) => match &e { + tantivy::TantivyError::IoError(io_error) => { + if attempt < MAX_RETRIES { + eprintln!( + "[WARN] Commit failed (attempt {}/{}): {:?}. Retrying in {}ms...", + attempt + 1, + MAX_RETRIES + 1, + io_error, + RETRY_DELAY_MS * (attempt as u64 + 1) + ); + std::thread::sleep(std::time::Duration::from_millis( + RETRY_DELAY_MS * (attempt as u64 + 1), + )); + } else { + eprintln!( + "[FATAL] Tantivy commit failed after {} attempts: {:?}", + MAX_RETRIES + 1, + io_error + ); + std::process::exit(1); + } + } + _ => { + eprintln!("[FATAL] Tantivy commit failed with non-IO error: {e:?}"); + std::process::exit(1); + } + }, + } + } +} + +fn open(index_dir: &PathBuf) -> Index { + Index::open_in_dir(index_dir) + .unwrap_or_else(|e| panic!("Failed to open index in {:?}: {}", index_dir, e)) +} diff --git a/src/modules/indexer/envelope.rs b/src/modules/indexer/envelope.rs index e0d0a7e..d5b8d24 100644 --- a/src/modules/indexer/envelope.rs +++ b/src/modules/indexer/envelope.rs @@ -42,11 +42,17 @@ pub struct Envelope { pub size: u32, pub thread_id: String, pub attachment_count: usize, + pub regular_attachment_count: usize, pub tags: Option>, + /// Hash of the content. pub content_hash: String, } impl Envelope { + pub fn has_attachments(&self) -> bool { + self.attachment_count > 0 + } + pub fn from_row(row: &duckdb::Row) -> duckdb::Result { let get_list = |col_name: &str| -> Vec { row.get::<_, Value>(col_name) @@ -100,6 +106,7 @@ impl Envelope { size: row.get::<_, u64>("size_bytes")? as u32, thread_id: row.get("thread_id")?, attachment_count: row.get::<_, i32>("attachment_count")? as usize, + regular_attachment_count: row.get::<_, i32>("regular_attachment_count")? as usize, tags: { let t = get_list("tags"); if t.is_empty() { diff --git a/src/modules/indexer/fields.rs b/src/modules/indexer/fields.rs index 6b5400d..9a102b0 100644 --- a/src/modules/indexer/fields.rs +++ b/src/modules/indexer/fields.rs @@ -18,15 +18,10 @@ use tantivy::schema::Field; -pub const F_ACCOUNT_ID: &str = "account_id"; -pub const F_MAILBOX_ID: &str = "mailbox_id"; - pub const F_ID: &str = "id"; pub const F_BLOB: &str = "blob"; pub struct BlobFields { pub f_id: Field, - pub f_account_id: Field, - pub f_mailbox_id: Field, pub f_blob: Field, } diff --git a/src/modules/indexer/manager.rs b/src/modules/indexer/manager.rs index f69f10a..f158cc6 100644 --- a/src/modules/indexer/manager.rs +++ b/src/modules/indexer/manager.rs @@ -18,52 +18,35 @@ use std::{ collections::{HashMap, HashSet}, - path::PathBuf, - sync::{Arc, LazyLock}, + sync::LazyLock, time::Duration, }; use crate::modules::{ duckdb::init::duckdb, message::{ - attachment::AttachmentMetadata, content::AttachmentInfo, search::SortBy, tags::TagCount, + attachment::AttachmentMetadata, + content::{AttachmentDetail, AttachmentInfo}, + search::SortBy, + tags::TagCount, }, - settings::cli::SETTINGS, }; use crate::{ modules::{ common::signal::SIGNAL_MANAGER, dashboard::{DashboardStats, LargestEmail}, error::{code::ErrorCode, BichonResult}, - indexer::{envelope::Envelope, schema::SchemaTools}, + indexer::envelope::Envelope, message::search::SearchFilter, rest::response::DataPage, - settings::dir::DATA_DIR_MANAGER, }, raise_error, }; -use mail_parser::{MessageParser, MimeHeaders}; - -use tantivy::indexer::{LogMergePolicy, UserOperation}; -use tantivy::{ - collector::TopDocs, - query::{BooleanQuery, Occur, Query, TermQuery}, - schema::{IndexRecordOption, Value}, - store::{Compressor, ZstdCompressor}, - Index, IndexBuilder, IndexReader, IndexSettings, IndexWriter, TantivyDocument, Term, -}; -use tokio::{ - fs::File, - io::AsyncWriteExt, - sync::{mpsc, Mutex}, - task, -}; -use tracing::info; +use tokio::{sync::mpsc, task}; pub static ENVELOPE_INDEX_MANAGER: LazyLock = LazyLock::new(EnvelopeIndexManager::new); -pub static EML_INDEX_MANAGER: LazyLock = LazyLock::new(EmlIndexManager::new); pub const ENVELOPE_BATCH_SIZE: usize = 500; pub const EML_BATCH_SIZE: usize = 100; @@ -75,11 +58,6 @@ pub enum MetadataOp { Shutdown, } -pub enum DocumentOp { - Document((String, TantivyDocument)), - Shutdown, -} - pub struct EnvelopeIndexManager { sender: mpsc::Sender, } @@ -150,30 +128,31 @@ impl EnvelopeIndexManager { .map_err(|e| raise_error!(format!("{:?}", e), ErrorCode::InternalError))? } - pub async fn delete_account_envelopes(&self, account_id: u64) -> BichonResult<()> { - let _ = - tokio::task::spawn_blocking(move || duckdb()?.delete_envelopes_by_account(account_id)) - .await - .map_err(|e| raise_error!(format!("{:?}", e), ErrorCode::InternalError))?; - Ok(()) + pub async fn delete_account_envelopes(&self, account_id: u64) -> BichonResult> { + let content_hashes = tokio::task::spawn_blocking(move || { + duckdb()?.delete_account_envelopes_with_orphans(account_id) + }) + .await + .map_err(|e| raise_error!(format!("{:?}", e), ErrorCode::InternalError))??; + Ok(content_hashes) } pub async fn delete_mailbox_envelopes( &self, account_id: u64, mailbox_ids: Vec, - ) -> BichonResult<()> { + ) -> BichonResult> { if mailbox_ids.is_empty() { tracing::warn!("delete_mailbox_envelopes: mailbox_ids is empty, nothing to delete"); - return Ok(()); + return Ok(vec![]); } - let _ = tokio::task::spawn_blocking(move || { - duckdb()?.delete_mailbox_envelopes(account_id, mailbox_ids) + let content_hashes = tokio::task::spawn_blocking(move || { + duckdb()?.delete_mailbox_envelopes_with_orphans(account_id, mailbox_ids) }) .await - .map_err(|e| raise_error!(format!("{:?}", e), ErrorCode::InternalError))?; - Ok(()) + .map_err(|e| raise_error!(format!("{:?}", e), ErrorCode::InternalError))??; + Ok(content_hashes) } pub async fn get_all_tags( @@ -203,6 +182,15 @@ impl EnvelopeIndexManager { .map_err(|e| raise_error!(format!("{:?}", e), ErrorCode::InternalError))? } + pub async fn get_orphan_hashes_in_memory( + &self, + deletes: HashMap>, + ) -> BichonResult> { + tokio::task::spawn_blocking(move || duckdb()?.get_orphan_hashes_in_memory(deletes)) + .await + .map_err(|e| raise_error!(format!("{:?}", e), ErrorCode::InternalError))? + } + pub async fn delete_envelopes_multi_account( &self, deletes: HashMap>, // HashMap @@ -211,7 +199,6 @@ impl EnvelopeIndexManager { tracing::warn!("delete_envelopes_multi_account: deletes is empty, nothing to delete"); return Ok(()); } - tokio::task::spawn_blocking(move || duckdb()?.delete_envelopes_multi_account(deletes)) .await .map_err(|e| raise_error!(format!("{:?}", e), ErrorCode::InternalError))? @@ -293,6 +280,18 @@ impl EnvelopeIndexManager { .map_err(|e| raise_error!(format!("{:?}", e), ErrorCode::InternalError))? } + pub async fn get_attachments_by_envelope_id( + &self, + account_id: u64, + envelope_id: String, + ) -> BichonResult> { + tokio::task::spawn_blocking(move || { + duckdb()?.get_attachments_by_envelope_id(account_id, envelope_id) + }) + .await + .map_err(|e| raise_error!(format!("{:?}", e), ErrorCode::InternalError))? + } + pub async fn top_10_largest_emails( &self, accounts: Option>, @@ -339,537 +338,3 @@ impl EnvelopeIndexManager { .map_err(|e| raise_error!(format!("{:?}", e), ErrorCode::InternalError))? } } - -pub struct EmlIndexManager { - index_writer: Arc>, - sender: mpsc::Sender, - reader: IndexReader, -} - -impl EmlIndexManager { - pub fn new() -> Self { - let index = Self::open_or_create_index(&DATA_DIR_MANAGER.eml_dir); - - let writer: IndexWriter = index - .writer_with_num_threads( - SETTINGS.bichon_tantivy_threads as usize, - SETTINGS.bichon_tantivy_buffer_size, - ) - .unwrap_or_else(|e| { - panic!( - "Failed to create IndexWriter (threads: {}, buffer: {}B) for {:?}: {}", - SETTINGS.bichon_tantivy_threads, - SETTINGS.bichon_tantivy_buffer_size, - DATA_DIR_MANAGER.eml_dir, - e - ) - }); - - let mut merge_policy = LogMergePolicy::default(); - merge_policy.set_min_num_segments(20); - merge_policy.set_max_docs_before_merge(10_000); - merge_policy.set_min_layer_size(1000); - writer.set_merge_policy(Box::new(merge_policy)); - - let index_writer = Arc::new(Mutex::new(writer)); - - let reader = index.reader().unwrap_or_else(|e| { - panic!( - "Failed to create IndexReader for {:?}: {}", - DATA_DIR_MANAGER.eml_dir, e - ) - }); - let (sender, mut receiver) = mpsc::channel::(100); - task::spawn(async move { - let mut buffer: HashMap = - HashMap::with_capacity(EML_BATCH_SIZE); - let mut interval = tokio::time::interval(MAX_BUFFER_DURATION); - let mut shutdown = SIGNAL_MANAGER.subscribe(); - loop { - tokio::select! { - maybe_msg = receiver.recv() => { - match maybe_msg { - Some(DocumentOp::Document((eid, doc))) => { - buffer.insert(eid, doc); - if buffer.len() >= EML_BATCH_SIZE { - EML_INDEX_MANAGER.drain_and_commit(&mut buffer).await; - } - } - Some(DocumentOp::Shutdown) => { - EML_INDEX_MANAGER.drain_and_commit(&mut buffer).await; - break; - } - None => break, - } - } - _ = interval.tick() => { - if !buffer.is_empty() { - EML_INDEX_MANAGER.drain_and_commit(&mut buffer).await; - } - } - _ = shutdown.recv() => { - let _ = EML_INDEX_MANAGER.sender.send(DocumentOp::Shutdown).await; - } - } - } - }); - Self { - index_writer, - sender, - reader, - } - } - /// Adds a document to the indexer. - /// - /// # Parameters - /// - `eid`: A hash derived from **Account ID + Message ID**. - /// This acts as a unique identifier for the EML content itself. - /// - /// - `doc`: The `TantivyDocument` representing the mail body/content. - /// - /// # Logical Design - /// Unlike the `envelope_id` (which is a hash of Account + Folder + Message ID), - /// this `eid` ignores the folder context. This ensures that while metadata - /// (envelopes) can be duplicated across different folders, the physical - /// EML/document storage remains de-duplicated and unique. - pub async fn add_document(&self, content_hash: String, doc: TantivyDocument) { - let _ = self - .sender - .send(DocumentOp::Document((content_hash, doc))) - .await; - } - - fn open_or_create_index(index_dir: &PathBuf) -> Index { - let need_create = !index_dir.exists() - || index_dir - .read_dir() - .map(|mut d| d.next().is_none()) - .unwrap_or(true); - - if need_create { - info!( - "Email data storage not found or empty, creating new mail storage at {}", - index_dir.display() - ); - std::fs::create_dir_all(&index_dir).unwrap_or_else(|e| { - panic!("Failed to create index directory {:?}: {}", index_dir, e) - }); - IndexBuilder::new() - .schema(SchemaTools::schema()) - .settings(IndexSettings { - docstore_compression: Compressor::Zstd(ZstdCompressor { - compression_level: Some(SETTINGS.bichon_eml_compression_level as i32), - }), - docstore_compress_dedicated_thread: true, - docstore_blocksize: SETTINGS.bichon_eml_blocksize, - }) - .create_in_dir(&index_dir) - .unwrap_or_else(|e| panic!("Failed to create index in {:?}: {}", index_dir, e)) - } else { - info!( - "Opening existing email data storage at {}", - index_dir.display() - ); - open(&index_dir) - } - } - - fn envelope_query(&self, account_id: u64, eid: &str) -> Box { - let account_id_query = TermQuery::new( - Term::from_field_u64(SchemaTools::fields().f_account_id, account_id), - IndexRecordOption::Basic, - ); - let envelope_id_query = TermQuery::new( - Term::from_field_text(SchemaTools::fields().f_id, eid), - IndexRecordOption::Basic, - ); - let boolean_query = BooleanQuery::new(vec![ - (Occur::Must, Box::new(account_id_query)), - (Occur::Must, Box::new(envelope_id_query)), - ]); - Box::new(boolean_query) - } - - pub async fn get(&self, account_id: u64, eml_id: &str) -> BichonResult>> { - let searcher = self.reader.searcher(); - let query = self.envelope_query(account_id, eml_id); - let docs = searcher - .search(query.as_ref(), &TopDocs::with_limit(1)) - .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; - - if docs.is_empty() { - return Ok(None); - } - - let (_, doc_address) = docs.first().unwrap(); - let doc: TantivyDocument = searcher - .doc_async(*doc_address) - .await - .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; - let fields = SchemaTools::fields(); - let value = doc.get_first(fields.f_blob).ok_or_else(|| { - raise_error!( - format!("miss '{}' field in tantivy document", stringify!(field)), - ErrorCode::InternalError - ) - })?; - let bytes = value.as_bytes().ok_or_else(|| { - raise_error!( - format!("'{}' field is not a bytes", stringify!(field)), - ErrorCode::InternalError - ) - })?; - - Ok(Some(bytes.to_vec())) - } - - pub async fn get_reader(&self, account_id: u64, eid: String) -> BichonResult { - let envelope = duckdb()? - .get_envelope_by_id(account_id, eid.clone())? - .ok_or_else(|| { - raise_error!( - format!( - "Email envelope not found: account_id={} id={}", - account_id, &eid - ), - ErrorCode::ResourceNotFound - ) - })?; - let data = self - .get(account_id, &envelope.content_hash) - .await? - .ok_or_else(|| { - raise_error!( - format!("Eml not found: account_id={}, eid={}", account_id, &eid), - ErrorCode::ResourceNotFound - ) - })?; - let mut path = DATA_DIR_MANAGER.temp_dir.clone(); - - path.push(format!("{eid}.eml")); - { - let mut file = File::create(&path) - .await - .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; - file.write_all(&data) - .await - .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; - } - let file = File::open(&path) - .await - .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; - Ok(file) - } - - pub async fn get_attachment_content( - &self, - account_id: u64, - eid: String, - file_name: &str, - ) -> BichonResult> { - let envelope = duckdb()? - .get_envelope_by_id(account_id, eid.clone())? - .ok_or_else(|| { - raise_error!( - format!( - "Email envelope not found: account_id={} id={}", - account_id, &eid - ), - ErrorCode::ResourceNotFound - ) - })?; - let data = self - .get(account_id, envelope.content_hash.as_str()) - .await? - .ok_or_else(|| { - raise_error!( - format!("Email not found: account_id={}, eid={}", account_id, &eid), - ErrorCode::ResourceNotFound - ) - })?; - let message = MessageParser::default().parse(&data).ok_or_else(|| { - raise_error!( - format!( - "Failed to parse email: account_id={}, eid={}", - account_id, &eid - ), - ErrorCode::InternalError - ) - })?; - - let content = message - .attachments() - .find(|att| { - att.attachment_name() - .map(|name| name == file_name) - .unwrap_or(false) - }) - .map(|att| att.contents().to_vec()) - .ok_or_else(|| { - raise_error!( - format!("Attachment '{}' not found in email {}", file_name, eid), - ErrorCode::ResourceNotFound - ) - })?; - - Ok(content) - } - - pub async fn get_attachment( - &self, - account_id: u64, - eid: String, - file_name: &str, - ) -> BichonResult { - let content = self - .get_attachment_content(account_id, eid.clone(), file_name) - .await?; - let mut path = DATA_DIR_MANAGER.temp_dir.clone(); - path.push(format!("{eid}.{file_name}.attachment")); - { - let mut file = File::create(&path) - .await - .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; - file.write_all(&content) - .await - .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; - } - let file = File::open(&path) - .await - .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; - Ok(file) - } - - pub async fn get_nested_attachment( - &self, - account_id: u64, - eid: String, - file_name: &str, - nested_file_name: &str, - ) -> BichonResult { - let content = self - .get_attachment_content(account_id, eid.clone(), file_name) - .await?; - - let message = MessageParser::default().parse(&content).ok_or_else(|| { - raise_error!( - format!( - "Failed to parse email: account_id={}, eid={}", - account_id, &eid - ), - ErrorCode::InternalError - ) - })?; - - let content = message - .attachments() - .find(|att| { - att.attachment_name() - .map(|name| name == nested_file_name) - .unwrap_or(false) - }) - .map(|att| att.contents().to_vec()) - .ok_or_else(|| { - raise_error!( - format!( - "Nested attachment '{}' not found in email {}", - nested_file_name, &eid - ), - ErrorCode::ResourceNotFound - ) - })?; - - let mut path = DATA_DIR_MANAGER.temp_dir.clone(); - path.push(format!("{eid}.{file_name}.{nested_file_name}.attachment")); - { - let mut file = File::create(&path) - .await - .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; - file.write_all(&content) - .await - .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; - } - let file = File::open(&path) - .await - .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; - Ok(file) - } - - fn account_query(&self, account_id: u64) -> Box { - let account_term = Term::from_field_u64(SchemaTools::fields().f_account_id, account_id); - Box::new(TermQuery::new(account_term, IndexRecordOption::Basic)) - } - - pub async fn delete_account_envelopes(&self, account_id: u64) -> BichonResult<()> { - let query = self.account_query(account_id); - let mut writer = self.index_writer.lock().await; - writer - .delete_query(query) - .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; - writer - .commit() - .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; - Ok(()) - } - - fn mailbox_query(&self, account_id: u64, mailbox_id: u64) -> Box { - let account_query = TermQuery::new( - Term::from_field_u64(SchemaTools::fields().f_account_id, account_id), - IndexRecordOption::Basic, - ); - let mailbox_query = TermQuery::new( - Term::from_field_u64(SchemaTools::fields().f_mailbox_id, mailbox_id), - IndexRecordOption::Basic, - ); - let boolean_query = BooleanQuery::new(vec![ - (Occur::Must, Box::new(account_query)), - (Occur::Must, Box::new(mailbox_query)), - ]); - Box::new(boolean_query) - } - - pub async fn delete_mailbox_envelopes( - &self, - account_id: u64, - mailbox_ids: Vec, - ) -> BichonResult<()> { - if mailbox_ids.is_empty() { - tracing::warn!("delete_mailbox_envelopes: mailbox_ids is empty, nothing to delete"); - return Ok(()); - } - let mut queries: Vec> = Vec::with_capacity(mailbox_ids.len()); - for mailbox_id in mailbox_ids { - queries.push(self.mailbox_query(account_id, mailbox_id)); - } - let mut writer = self.index_writer.lock().await; - for query in queries { - writer - .delete_query(query) - .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; - } - writer - .commit() - .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; - Ok(()) - } - - pub async fn delete_email_multi_account( - &self, - deletes: &HashMap>, // HashMap - ) -> BichonResult<()> { - if deletes.is_empty() { - tracing::warn!("delete_email_multi_account: deletes is empty, nothing to delete"); - return Ok(()); - } - - let mut writer = self.index_writer.lock().await; - - for (account_id, envelope_ids) in deletes { - let unique_ids: Vec<&str> = envelope_ids - .iter() - .map(|s| s.as_str()) - .collect::>() - .into_iter() - .collect(); - if unique_ids.is_empty() { - continue; - } - - for chunk in unique_ids.chunks(100) { - let envelopes = duckdb()?.get_envelopes_by_ids(*account_id, chunk)?; - let found_ids_set: HashSet<&str> = - envelopes.iter().map(|e| e.id.as_str()).collect(); - for &original_id in chunk { - if !found_ids_set.contains(&original_id) { - tracing::warn!( - "delete_email_multi_account: envelope not found in DB, skipping tantivy delete. account_id: {}, envelope_id: {}", - account_id, original_id - ); - } - } - - for envelope in envelopes { - let hashed_id = &envelope.content_hash; - let query = self.envelope_query(*account_id, hashed_id); - - writer - .delete_query(query) - .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; - } - } - } - writer - .commit() - .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; - - Ok(()) - } - - async fn drain_and_commit(&self, buffer: &mut HashMap) { - if buffer.is_empty() { - return; - } - let mut writer = self.index_writer.lock().await; - let mut operations = Vec::new(); - - for (eid, doc) in buffer.drain() { - let delete_term = Term::from_field_text(SchemaTools::fields().f_id, &eid); - operations.push(UserOperation::Delete(delete_term)); - operations.push(UserOperation::Add(doc)); - } - if let Err(e) = writer.run(operations) { - eprintln!("[FATAL] Tantivy run failed: {e:?}"); - std::process::exit(1); - } - - fatal_commit(&mut writer); - } -} - -fn fatal_commit(writer: &mut IndexWriter) { - const MAX_RETRIES: usize = 3; - const RETRY_DELAY_MS: u64 = 1000; - - for attempt in 0..=MAX_RETRIES { - match writer.commit() { - Ok(_) => { - if attempt > 0 { - eprintln!("[INFO] Commit succeeded on attempt {}", attempt + 1); - } - return; - } - Err(e) => match &e { - tantivy::TantivyError::IoError(io_error) => { - if attempt < MAX_RETRIES { - eprintln!( - "[WARN] Commit failed (attempt {}/{}): {:?}. Retrying in {}ms...", - attempt + 1, - MAX_RETRIES + 1, - io_error, - RETRY_DELAY_MS * (attempt as u64 + 1) - ); - std::thread::sleep(std::time::Duration::from_millis( - RETRY_DELAY_MS * (attempt as u64 + 1), - )); - } else { - eprintln!( - "[FATAL] Tantivy commit failed after {} attempts: {:?}", - MAX_RETRIES + 1, - io_error - ); - std::process::exit(1); - } - } - _ => { - eprintln!("[FATAL] Tantivy commit failed with non-IO error: {e:?}"); - std::process::exit(1); - } - }, - } - } -} - -fn open(index_dir: &PathBuf) -> Index { - Index::open_in_dir(index_dir) - .unwrap_or_else(|e| panic!("Failed to open index in {:?}: {}", index_dir, e)) -} diff --git a/src/modules/indexer/mod.rs b/src/modules/indexer/mod.rs index 979558a..8cab6d1 100644 --- a/src/modules/indexer/mod.rs +++ b/src/modules/indexer/mod.rs @@ -16,7 +16,16 @@ // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . +use tantivy::TantivyDocument; + +pub mod attachment; +pub mod eml; pub mod envelope; pub mod fields; pub mod manager; pub mod schema; + +pub enum DocumentOp { + Document((String, TantivyDocument)), + Shutdown, +} diff --git a/src/modules/indexer/schema.rs b/src/modules/indexer/schema.rs index 594b9bd..139e302 100644 --- a/src/modules/indexer/schema.rs +++ b/src/modules/indexer/schema.rs @@ -19,8 +19,8 @@ use std::sync::{Arc, LazyLock}; use crate::modules::indexer::fields::*; +use tantivy::schema::STRING; use tantivy::schema::{Schema, FAST, STORED}; -use tantivy::schema::{INDEXED, STRING}; static BLOB_FIELDS: LazyLock> = LazyLock::new(|| { let (_, fields) = SchemaTools::create_schema(); @@ -42,15 +42,8 @@ impl SchemaTools { pub fn create_schema() -> (Schema, BlobFields) { let mut builder = Schema::builder(); let f_id = builder.add_text_field(F_ID, STRING | FAST); - let f_account_id = builder.add_u64_field(F_ACCOUNT_ID, INDEXED | STORED | FAST); - let f_mailbox_id = builder.add_u64_field(F_MAILBOX_ID, INDEXED | STORED | FAST); let f_blob = builder.add_bytes_field(F_BLOB, STORED); - let fields = BlobFields { - f_id, - f_account_id, - f_mailbox_id, - f_blob, - }; + let fields = BlobFields { f_id, f_blob }; (builder.build(), fields) } } diff --git a/src/modules/mailbox/delete.rs b/src/modules/mailbox/delete.rs index 5f1c98d..822a7ac 100644 --- a/src/modules/mailbox/delete.rs +++ b/src/modules/mailbox/delete.rs @@ -1,7 +1,10 @@ use crate::modules::{ cache::imap::mailbox::MailBox, error::BichonResult, - indexer::manager::{EML_INDEX_MANAGER, ENVELOPE_INDEX_MANAGER}, + indexer::{ + attachment::ATTACHMENT_INDEX_MANAGER, eml::EML_INDEX_MANAGER, + manager::ENVELOPE_INDEX_MANAGER, + }, }; pub async fn delete_mailbox_impl(account_id: u64, mailbox_id: u64) -> BichonResult<()> { @@ -26,13 +29,11 @@ pub async fn delete_mailbox_impl(account_id: u64, mailbox_id: u64) -> BichonResu MailBox::delete(*id).await?; } - ENVELOPE_INDEX_MANAGER + let content_hashes = ENVELOPE_INDEX_MANAGER .delete_mailbox_envelopes(account_id, ids_to_delete.clone()) .await?; - EML_INDEX_MANAGER - .delete_mailbox_envelopes(account_id, ids_to_delete) - .await?; - + EML_INDEX_MANAGER.delete(&content_hashes).await?; + ATTACHMENT_INDEX_MANAGER.delete(&content_hashes).await?; Ok(()) } diff --git a/src/modules/message/append.rs b/src/modules/message/append.rs index 9d0d05a..1544a1f 100644 --- a/src/modules/message/append.rs +++ b/src/modules/message/append.rs @@ -2,9 +2,9 @@ use crate::{ encode_mailbox_name, modules::{ account::migration::{AccountModel, AccountType}, + envelope::extractor::reattach_eml_content, error::{code::ErrorCode, BichonResult}, imap::executor::ImapExecutor, - indexer::manager::{EML_INDEX_MANAGER, ENVELOPE_INDEX_MANAGER}, }, raise_error, }; @@ -43,32 +43,7 @@ pub async fn restore_emails(account_id: u64, envelope_ids: Vec) -> Bicho let mut session = ImapExecutor::create_connection(account_id).await?; for envelope_id in envelope_ids { let result: BichonResult<()> = async { - let eid = envelope_id.clone(); - let envelope = ENVELOPE_INDEX_MANAGER - .get_envelope_by_id(account_id, eid) - .await? - .ok_or_else(|| { - raise_error!( - format!( - "Envelope not found: account_id={} message_id={}", - account_id, &envelope_id - ), - ErrorCode::ResourceNotFound - ) - })?; - let eml = EML_INDEX_MANAGER - .get(account_id, &envelope.content_hash) - .await? - .ok_or_else(|| { - raise_error!( - format!( - "Eml not found: account_id={} id={}", - account_id, &envelope_id - ), - ErrorCode::ResourceNotFound - ) - })?; - + let (envelope, eml) = reattach_eml_content(account_id, envelope_id.clone()).await?; if let Some(mailbox_name) = envelope.mailbox_name { ImapExecutor::append( &mut session, diff --git a/src/modules/message/attachment.rs b/src/modules/message/attachment.rs index 9c8ed16..0b43d25 100644 --- a/src/modules/message/attachment.rs +++ b/src/modules/message/attachment.rs @@ -1,7 +1,19 @@ use std::collections::HashSet; +use crate::{ + modules::{ + envelope::extractor::reattach_eml_content, + error::{code::ErrorCode, BichonResult}, + settings::dir::DATA_DIR_MANAGER, + utils::compute_content_hash, + }, + raise_error, +}; +use mail_parser::MessageParser; use poem_openapi::Object; use serde::{Deserialize, Serialize}; +use tokio::fs::File; +use tokio::io::AsyncWriteExt; #[derive(Clone, Debug, Default, Eq, PartialEq, Deserialize, Serialize, Object)] pub struct AttachmentMetadata { @@ -17,3 +29,103 @@ pub struct AttachmentMetadata { /// Example: ["application/pdf", "image/jpeg"] pub content_types: HashSet, } + +pub async fn retrieve_attachment_content( + account_id: u64, + envelope_id: String, + content_hash: &str, +) -> BichonResult { + let (envelope, eml) = reattach_eml_content(account_id, envelope_id).await?; + let message = MessageParser::default().parse(&eml).ok_or_else(|| { + raise_error!( + "Failed to parse parent EML".into(), + ErrorCode::InternalError + ) + })?; + + let attachment_content: &[u8] = message + .attachments() + .find(|att| compute_content_hash(att.contents()) == content_hash) + .map(|att| att.contents()) + .ok_or_else(|| { + raise_error!( + "Target nested EML not found".into(), + ErrorCode::ResourceNotFound + ) + })?; + let mut path = DATA_DIR_MANAGER.temp_dir.clone(); + path.push(format!("{}.eml", envelope.content_hash)); + { + let mut file = File::create(&path) + .await + .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; + file.write_all(attachment_content) + .await + .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; + } + let file = File::open(&path) + .await + .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; + Ok(file) +} + +pub async fn retrieve_nested_attachment_content( + account_id: u64, + envelope_id: String, + content_hash: &str, + nested_content_hash: &str, +) -> BichonResult { + let (_, eml) = reattach_eml_content(account_id, envelope_id).await?; + let parent_message = MessageParser::default().parse(&eml).ok_or_else(|| { + raise_error!( + "Failed to parse parent EML".into(), + ErrorCode::InternalError + ) + })?; + + let attachment_content = parent_message + .attachments() + .find(|att| compute_content_hash(att.contents()) == content_hash) + .map(|att| att.contents()) + .ok_or_else(|| { + raise_error!( + "Target nested EML not found".into(), + ErrorCode::ResourceNotFound + ) + })?; + + let nested_message = MessageParser::default() + .parse(attachment_content) + .ok_or_else(|| { + raise_error!( + "Failed to parse nested EML".into(), + ErrorCode::InternalError + ) + })?; + + let attachment_content = nested_message + .attachments() + .find(|att| compute_content_hash(att.contents()) == nested_content_hash) + .map(|att| att.contents()) + .ok_or_else(|| { + raise_error!( + "Target nested EML not found".into(), + ErrorCode::ResourceNotFound + ) + })?; + + let mut path = DATA_DIR_MANAGER.temp_dir.clone(); + path.push(format!("{}.eml", nested_content_hash)); + { + let mut file = File::create(&path) + .await + .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; + file.write_all(attachment_content) + .await + .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; + } + let file = File::open(&path) + .await + .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; + Ok(file) +} diff --git a/src/modules/message/content.rs b/src/modules/message/content.rs index c97dedc..3654670 100644 --- a/src/modules/message/content.rs +++ b/src/modules/message/content.rs @@ -18,16 +18,16 @@ use crate::base64_encode; use crate::modules::account::migration::AccountModel; -use crate::modules::envelope::extractor::extract_envelope_from_message; +use crate::modules::envelope::extractor::{ + extract_envelope_from_nested_message, reattach_eml_content, +}; use crate::modules::error::code::ErrorCode; use crate::modules::indexer::envelope::Envelope; -use crate::modules::indexer::manager::{EML_INDEX_MANAGER, ENVELOPE_INDEX_MANAGER}; +use crate::modules::utils::compute_content_hash; use crate::{modules::error::BichonResult, raise_error}; use mail_parser::{MessageParser, MimeHeaders}; - use poem_openapi::Object; use serde::{Deserialize, Serialize}; - /// Represents metadata of an attachment in a Gmail message. /// /// This struct stores information required to identify, download, @@ -40,53 +40,56 @@ pub struct AttachmentInfo { /// Whether the attachment is marked as inline (true) or a regular file (false). pub inline: bool, /// Original filename of the attachment, if provided. - pub filename: String, + pub filename: Option, /// Size of the attachment in bytes. pub size: usize, pub content_id: Option, + /// Hash of the content. + pub content_hash: String, + pub is_message: bool, } impl AttachmentInfo { - pub fn get_extension(&self) -> String { - std::path::Path::new(&self.filename) - .extension() + pub fn get_extension(&self) -> Option { + self.filename + .as_deref() + .and_then(|f| std::path::Path::new(f).extension()) .and_then(|ext| ext.to_str()) .map(|ext| ext.to_ascii_lowercase()) - .unwrap_or_default() } pub fn get_category(&self) -> &'static str { - let ext = self.get_extension(); + if let Some(ext) = self.get_extension() { + let category = match ext.as_str() { + "doc" | "docx" | "pdf" | "rtf" | "odt" | "pages" | "pptx" | "ppt" => { + Some("document") + } + "xls" | "xlsx" | "ods" | "numbers" | "csv" => Some("spreadsheet"), + "ical" | "ics" | "vcs" | "ifb" | "icalendar" => Some("event"), + "txt" | "log" | "md" => Some("text"), + "jpg" | "jpeg" | "png" | "gif" | "bmp" | "tiff" | "avif" | "heic" | "heif" + | "webp" => Some("image"), + "mp4" | "mkv" | "mov" | "avi" | "webm" => Some("video"), + "wav" | "mp3" | "aac" | "ogg" | "wma" | "flac" | "aiff" => Some("audio"), + "psd" | "eps" | "svg" | "cdr" | "ai" => Some("graphics_2d"), + "stl" | "obj" | "3mf" | "amf" | "f3d" | "sldprt" | "stp" | "step" | "dwg" + | "x_t" | "x_b" | "sat" | "ipt" => Some("graphics_3d"), + "c" | "h" | "html" | "css" | "js" | "ts" | "vue" | "tsx" | "svelte" | "py" + | "java" | "cs" | "go" | "rb" | "php" | "swift" | "rs" | "r" | "jl" | "lua" + | "sql" => Some("code"), + "tsv" | "xml" | "json" | "yml" | "yaml" | "toml" | "env" | "ini" => Some("data"), + "ps1" | "sh" | "bat" | "cmd" | "exe" | "msi" | "dmg" | "pkg" | "deb" | "rpm" => { + Some("executable") + } + "zip" | "gz" | "tgz" | "7z" | "rar" | "tar" | "bz2" | "zst" | "xz" | "iso" + | "img" => Some("archive"), + "eml" | "msg" => Some("message"), + _ => None, + }; - let category = match ext.as_str() { - "doc" | "docx" | "pdf" | "rtf" | "odt" | "pages" | "pptx" | "ppt" => Some("document"), - "xls" | "xlsx" | "ods" | "numbers" | "csv" => Some("spreadsheet"), - "ical" | "ics" | "vcs" | "ifb" | "icalendar" => Some("event"), - "txt" | "log" | "md" => Some("text"), - "jpg" | "jpeg" | "png" | "gif" | "bmp" | "tiff" | "avif" | "heic" | "heif" | "webp" => { - Some("image") + if let Some(cat) = category { + return cat; } - "mp4" | "mkv" | "mov" | "avi" | "webm" => Some("video"), - "wav" | "mp3" | "aac" | "ogg" | "wma" | "flac" | "aiff" => Some("audio"), - "psd" | "eps" | "svg" | "cdr" | "ai" => Some("graphics_2d"), - "stl" | "obj" | "3mf" | "amf" | "f3d" | "sldprt" | "stp" | "step" | "dwg" | "x_t" - | "x_b" | "sat" | "ipt" => Some("graphics_3d"), - "c" | "h" | "html" | "css" | "js" | "ts" | "vue" | "tsx" | "svelte" | "py" | "java" - | "cs" | "go" | "rb" | "php" | "swift" | "rs" | "r" | "jl" | "lua" | "sql" => { - Some("code") - } - "tsv" | "xml" | "json" | "yml" | "yaml" | "toml" | "env" | "ini" => Some("data"), - "ps1" | "sh" | "bat" | "cmd" | "exe" | "msi" | "dmg" | "pkg" | "deb" | "rpm" => { - Some("executable") - } - "zip" | "gz" | "tgz" | "7z" | "rar" | "tar" | "bz2" | "zst" | "xz" | "iso" | "img" => { - Some("archive") - } - _ => None, - }; - - if let Some(cat) = category { - return cat; } let mime = self.file_type.to_lowercase(); @@ -102,16 +105,46 @@ impl AttachmentInfo { if mime.starts_with("text/") { return "text"; } + if mime == "message/rfc822" { + return "message"; + } if mime.contains("compressed") || mime.contains("zip") || mime.contains("archive") { return "archive"; } if mime.contains("pdf") || mime.contains("msword") || mime.contains("officedocument") { return "document"; } + if mime.contains("spreadsheet") || mime.contains("excel") { + return "spreadsheet"; + } "other" } } + +#[derive(Clone, Debug, Default, Eq, PartialEq)] +pub struct AttachmentDetail { + pub shard_id: usize, + pub info: AttachmentInfo, +} + +impl AttachmentDetail { + pub fn from_row(row: &duckdb::Row) -> duckdb::Result { + Ok(Self { + shard_id: row.get("shard_id")?, + info: AttachmentInfo { + file_type: row.get("content_type")?, + inline: row.get("is_inline")?, + filename: row.get("filename")?, + size: row.get("size_bytes")?, + content_id: row.get("cid")?, + content_hash: row.get("content_hash")?, + is_message: row.get("is_message")?, + }, + }) + } +} + /// Represents the content of an email message in both plain text and HTML formats. /// /// This struct contains optional fields for plain text and HTML versions of @@ -148,37 +181,10 @@ pub async fn retrieve_email_content( envelope_id: String, ) -> BichonResult { AccountModel::check_account_exists(account_id).await?; - let envelope = ENVELOPE_INDEX_MANAGER - .get_envelope_by_id(account_id, envelope_id.clone()) - .await? - .ok_or_else(|| { - raise_error!( - format!( - "Email record not found: account_id={} id={}", - account_id, &envelope_id - ), - ErrorCode::ResourceNotFound - ) - })?; - - let eml = EML_INDEX_MANAGER - .get(account_id, &envelope.content_hash) - .await? - .ok_or_else(|| { - raise_error!( - format!( - "Email record not found: account_id={} id={}", - account_id, &envelope_id - ), - ErrorCode::ResourceNotFound - ) - })?; + let (envelope, eml) = reattach_eml_content(account_id, envelope_id).await?; let message = MessageParser::default().parse(&eml).ok_or_else(|| { raise_error!( - format!( - "Failed to parse EML data (id={}) — the message may be corrupted.", - &envelope_id - ), + "Failed to parse EML data — the message may be corrupted.".into(), ErrorCode::InternalError ) })?; @@ -190,24 +196,13 @@ pub async fn retrieve_email_content( raise_error!( format!( "Attachment is missing Content-Type (email id={})", - &envelope_id + &envelope.id ), ErrorCode::InternalError ) })?; - let filename = attachment - .attachment_name() - .map(|name| name.to_string()) - .unwrap_or_else(|| { - format!( - "email{}_attachment{}", - &envelope_id, - attachment.raw_body_offset() - ) - }); - + let filename = attachment.attachment_name().map(|name| name.to_string()); let disposition = attachment.content_disposition(); - let file_type = format!( "{}/{}", content_type.c_type.as_ref(), @@ -235,12 +230,15 @@ pub async fn retrieve_email_content( if inline && attachment.content_id().is_some() { continue; } - + let is_message = attachment.is_message(); + let content_hash = compute_content_hash(attachment.contents()); attachments.push(AttachmentInfo { - filename, + filename: filename.or(Some(content_hash.clone())), // Fallback to content_hash as the default filename if it is not provided. size: attachment.contents().len(), inline, file_type, + is_message, + content_hash, content_id: attachment.content_id().map(Into::into), }); } @@ -254,65 +252,87 @@ pub async fn retrieve_email_content( pub async fn retrieve_nested_eml_content( account_id: u64, envelope_id: String, - name: &str, + content_hash: &str, ) -> BichonResult { - let attachment_content = EML_INDEX_MANAGER - .get_attachment_content(account_id, envelope_id, name) - .await?; - let message = MessageParser::default().parse(&attachment_content).ok_or_else(|| { + let (_, eml) = reattach_eml_content(account_id, envelope_id).await?; + let parent_message = MessageParser::default().parse(&eml).ok_or_else(|| { raise_error!( - format!( - "Unable to parse '{}' as an email. It may not be in RFC822 format or the file is corrupted.", - name - ), + "Failed to parse parent EML".into(), ErrorCode::InternalError ) })?; - let mut html: Option = message.body_html(0).map(|cow| cow.into_owned()); - let text: Option = message.body_text(0).map(|cow| cow.into_owned()); + let attachment_content = parent_message + .attachments() + .find(|att| compute_content_hash(att.contents()) == content_hash) + .map(|att| att.contents()) + .ok_or_else(|| { + raise_error!( + "Target nested EML not found".into(), + ErrorCode::ResourceNotFound + ) + })?; + + let nested_message = MessageParser::default() + .parse(attachment_content) + .ok_or_else(|| { + raise_error!( + "Failed to parse nested EML".into(), + ErrorCode::InternalError + ) + })?; + + let mut html = nested_message.body_html(0).map(|c| c.into_owned()); + let text = nested_message.body_text(0).map(|c| c.into_owned()); + let mut attachments = Vec::new(); - for attachment in message.attachments() { - let content_type = attachment.content_type(); - let file_type = content_type.map_or_else( + let has_html = html.is_some(); + + for attachment in nested_message.attachments() { + let cid = attachment.content_id(); + let disposition = attachment.content_disposition(); + let is_inline = disposition.map(|d| d.is_inline()).unwrap_or(false); + + if has_html && is_inline && cid.is_some() { + let content_id = cid.unwrap(); + let html_ref = html.as_mut().unwrap(); + + let cid_pattern = format!("cid:{}", content_id); + if html_ref.contains(&cid_pattern) { + let data = attachment.contents(); + let ct = attachment + .content_type() + .map(|ct| format!("{}/{}", ct.c_type, ct.c_subtype.as_deref().unwrap_or(""))) + .unwrap_or_else(|| "image/png".to_string()); + + let base64_data = format!("data:{};base64,{}", ct, base64_encode!(data)); + *html_ref = html_ref.replace(&cid_pattern, &base64_data); + continue; + } + } + + let file_type = attachment.content_type().map_or_else( || "application/octet-stream".to_string(), |ct| format!("{}/{}", ct.c_type, ct.c_subtype.as_deref().unwrap_or("")), ); - - let filename = attachment - .attachment_name() - .map(|n| n.to_string()) - .unwrap_or_else(|| format!("attached_file_{}", attachment.raw_body_offset())); - - let disposition = attachment.content_disposition(); - let is_inline = disposition.map(|d| d.is_inline()).unwrap_or(false); - let cid = attachment.content_id(); - - if is_inline && cid.is_some() { - if let (Some(html_str), Some(content_id)) = (html.as_mut(), cid) { - if html_str.contains(content_id) { - let data = attachment.contents(); - let base64_encoded = base64_encode!(data); - *html_str = html_str.replace( - &format!("cid:{}", content_id), - &format!("data:{};base64,{}", file_type, base64_encoded), - ); - } - } - continue; - } - + let content_hash = compute_content_hash(attachment.contents()); attachments.push(AttachmentInfo { - filename, + filename: attachment + .attachment_name() + .map(|n| n.to_string()) + .or(Some(content_hash.clone())), // Fallback to content_hash as the default filename if it is not provided. size: attachment.contents().len(), inline: is_inline, file_type, + content_hash, + is_message: attachment.is_message(), content_id: cid.map(Into::into), }); } - let envelope = extract_envelope_from_message(message, account_id)?; + let envelope = extract_envelope_from_nested_message(nested_message, account_id)?; + Ok(FullNestedMessageContent { text, html, diff --git a/src/modules/message/delete.rs b/src/modules/message/delete.rs index e80dda7..bbd5c3d 100644 --- a/src/modules/message/delete.rs +++ b/src/modules/message/delete.rs @@ -17,13 +17,19 @@ // along with this program. If not, see . use crate::modules::error::BichonResult; -use crate::modules::indexer::manager::{EML_INDEX_MANAGER, ENVELOPE_INDEX_MANAGER}; +use crate::modules::indexer::attachment::ATTACHMENT_INDEX_MANAGER; +use crate::modules::indexer::eml::EML_INDEX_MANAGER; +use crate::modules::indexer::manager::ENVELOPE_INDEX_MANAGER; use std::collections::HashMap; pub async fn delete_messages_impl(request: HashMap>) -> BichonResult<()> { - EML_INDEX_MANAGER - .delete_email_multi_account(&request) + let content_hashes = ENVELOPE_INDEX_MANAGER + .get_orphan_hashes_in_memory(request.clone()) .await?; + if !content_hashes.is_empty() { + EML_INDEX_MANAGER.delete(&content_hashes).await?; + ATTACHMENT_INDEX_MANAGER.delete(&content_hashes).await?; + } ENVELOPE_INDEX_MANAGER .delete_envelopes_multi_account(request) .await diff --git a/src/modules/rest/api/message.rs b/src/modules/rest/api/message.rs index 0057c70..80244b1 100644 --- a/src/modules/rest/api/message.rs +++ b/src/modules/rest/api/message.rs @@ -18,11 +18,13 @@ use crate::modules::account::migration::AccountModel; use crate::modules::common::auth::ClientContext; +use crate::modules::indexer::eml::EML_INDEX_MANAGER; use crate::modules::indexer::envelope::Envelope; -use crate::modules::indexer::manager::EML_INDEX_MANAGER; use crate::modules::indexer::manager::ENVELOPE_INDEX_MANAGER; use crate::modules::message::append::restore_emails; use crate::modules::message::append::RestoreMessagesRequest; +use crate::modules::message::attachment::retrieve_attachment_content; +use crate::modules::message::attachment::retrieve_nested_attachment_content; use crate::modules::message::attachment::AttachmentMetadata; use crate::modules::message::content::retrieve_nested_eml_content; use crate::modules::message::content::FullNestedMessageContent; @@ -183,16 +185,16 @@ impl MessageApi { account_id: Path, /// The ID of the message to fetch. envelope_id: Path, - name: Query, + content_hash: Query, context: ClientContext, ) -> ApiResult> { let account_id = account_id.0; context .require_permission(Some(account_id), Permission::DATA_READ) .await?; - let name = name.0.trim(); + let content_hash = content_hash.0.trim(); Ok(Json( - retrieve_nested_eml_content(account_id, envelope_id.0, name).await?, + retrieve_nested_eml_content(account_id, envelope_id.0, content_hash).await?, )) } @@ -292,23 +294,22 @@ impl MessageApi { account_id: Path, /// The ID of the message containing the attachment. envelope_id: Path, - /// The filename of the attachment to download. - name: Query, + /// The content_hash of the attachment to download. + content_hash: Query, context: ClientContext, ) -> ApiResult> { let account_id = account_id.0; + let envelope_id = envelope_id.0.trim().to_string(); AccountModel::check_account_exists(account_id).await?; context .require_permission(Some(account_id), Permission::DATA_READ) .await?; - let name = name.0.trim(); - let reader = EML_INDEX_MANAGER - .get_attachment(account_id, envelope_id.0, name) - .await?; + let content_hash = content_hash.0.trim(); + let reader = retrieve_attachment_content(account_id, envelope_id, content_hash).await?; let body = Body::from_async_read(reader); let attachment = Attachment::new(body) .attachment_type(AttachmentType::Attachment) - .filename(name); + .filename(content_hash); Ok(attachment) } @@ -325,24 +326,29 @@ impl MessageApi { /// The ID of the message containing the attachment. envelope_id: Path, /// The filename of the attachment to download. - name: Query, - nested_name: Query, + content_hash: Query, + nested_content_hash: Query, context: ClientContext, ) -> ApiResult> { let account_id = account_id.0; + let envelope_id = envelope_id.0.trim().to_string(); AccountModel::check_account_exists(account_id).await?; context .require_permission(Some(account_id), Permission::DATA_READ) .await?; - let name = name.0.trim(); - let nested_name = nested_name.0.trim(); - let reader = EML_INDEX_MANAGER - .get_nested_attachment(account_id, envelope_id.0, name, nested_name) - .await?; + let content_hash = content_hash.0.trim(); + let nested_content_hash = nested_content_hash.0.trim(); + let reader = retrieve_nested_attachment_content( + account_id, + envelope_id, + content_hash, + nested_content_hash, + ) + .await?; let body = Body::from_async_read(reader); let attachment = Attachment::new(body) .attachment_type(AttachmentType::Attachment) - .filename(name); + .filename(nested_content_hash); Ok(attachment) } diff --git a/src/modules/settings/dir.rs b/src/modules/settings/dir.rs index 502a2e1..60c765c 100644 --- a/src/modules/settings/dir.rs +++ b/src/modules/settings/dir.rs @@ -29,6 +29,7 @@ pub const META_FILE: &str = "meta.db"; pub const MAILBOX_FILE: &str = "mailbox.db"; const ENVELOPE_DIR: &str = "envelope"; const EML_DIR: &str = "eml"; +const ATTACHMENT_DIR: &str = "attachment"; const TMP_DIR: &str = "tmp"; const LOG_DIR: &str = "logs"; const TLS_CERT: &str = "cert.pem"; @@ -47,6 +48,7 @@ pub struct DataDirManager { pub tls_key: PathBuf, pub envelope_dir: PathBuf, pub eml_dir: PathBuf, + pub attachment_dir: PathBuf, pub log_dir: PathBuf, } @@ -71,11 +73,17 @@ impl DataDirManager { }; let eml_dir = if let Some(ref data_dir) = SETTINGS.bichon_data_dir { - PathBuf::from(data_dir) + PathBuf::from(data_dir).join(EML_DIR) } else { root_dir.join(EML_DIR) }; + let attachment_dir = if let Some(ref data_dir) = SETTINGS.bichon_data_dir { + PathBuf::from(data_dir).join(ATTACHMENT_DIR) + } else { + root_dir.join(ATTACHMENT_DIR) + }; + Self { root_dir: root_dir.clone(), meta_db: root_dir.join(META_FILE), @@ -86,6 +94,7 @@ impl DataDirManager { envelope_dir, temp_dir: root_dir.join(TMP_DIR), eml_dir, + attachment_dir, } } } diff --git a/src/modules/smtp/server.rs b/src/modules/smtp/server.rs index 60b04a6..8238864 100644 --- a/src/modules/smtp/server.rs +++ b/src/modules/smtp/server.rs @@ -22,9 +22,6 @@ use std::time::Duration; use crate::modules::cache::imap::mailbox::{Attribute, AttributeEnum}; use crate::modules::envelope::extractor::extract_envelope_from_smtp; use crate::modules::error::BichonResult; -use crate::modules::indexer::manager::EML_INDEX_MANAGER; -use crate::modules::indexer::manager::ENVELOPE_INDEX_MANAGER; -use crate::modules::indexer::schema::SchemaTools; use crate::modules::utils::create_hash; use crate::modules::{ account::migration::AccountModel, @@ -35,7 +32,6 @@ use crate::modules::{ users::{permissions::Permission, UserModel}, }; use base64::{prelude::BASE64_STANDARD, Engine as _}; -use tantivy::doc; use tokio::time::timeout; use tokio::{ io::{AsyncBufReadExt, AsyncRead, AsyncWrite, AsyncWriteExt}, @@ -606,7 +602,6 @@ async fn read_data(reader: &mut R) -> io::Result BichonResult<()> { - let fields = SchemaTools::fields(); let rcpt = match session.rcpt_to.first() { Some(r) => r, None => { @@ -637,29 +632,14 @@ async fn parse_email(data: &[u8], session: &Session) -> BichonResult<()> { return Err(e.into()); } - let envelope = extract_envelope_from_smtp(data, rcpt.id, mailbox_id).map_err(|e| { - tracing::error!( - "SMTP: Envelope extraction failed for {}: {:?}", - rcpt.email, + extract_envelope_from_smtp(data, rcpt.id, mailbox_id) + .await + .map_err(|e| { + tracing::error!( + "SMTP: Envelope extraction failed for {}: {:?}", + rcpt.email, + e + ); e - ); - e - })?; - - let content_hash = envelope.0.content_hash.clone(); - ENVELOPE_INDEX_MANAGER.add_document(envelope).await; - - EML_INDEX_MANAGER - .add_document( - content_hash.clone(), - doc!( - fields.f_id => content_hash, - fields.f_account_id => rcpt.id, - fields.f_mailbox_id => mailbox_id, - fields.f_blob => data - ), - ) - .await; - - Ok(()) + }) } diff --git a/src/modules/utils/mod.rs b/src/modules/utils/mod.rs index ee75843..03cbe1a 100644 --- a/src/modules/utils/mod.rs +++ b/src/modules/utils/mod.rs @@ -372,7 +372,7 @@ pub fn validate_tag(tag: &str) -> Result<(), String> { Ok(()) } -pub fn content_hash(content: &[u8]) -> String { +pub fn compute_content_hash(content: &[u8]) -> String { let hash = blake3::hash(content); hash.to_hex().to_string() } diff --git a/web/src/api/index.ts b/web/src/api/index.ts index 0f721f8..69ca136 100644 --- a/web/src/api/index.ts +++ b/web/src/api/index.ts @@ -45,6 +45,7 @@ export interface EmailEnvelope { size: number; thread_id: string, attachment_count: number; + regular_attachment_count: number; tags: string[]; content_hash: string; } \ No newline at end of file diff --git a/web/src/api/mailbox/envelope/api.ts b/web/src/api/mailbox/envelope/api.ts index 473650a..f92d44a 100644 --- a/web/src/api/mailbox/envelope/api.ts +++ b/web/src/api/mailbox/envelope/api.ts @@ -34,16 +34,16 @@ export const get_thread_messages = async (accountId: number, thread_id: string, return response.data; } -export const download_attachment = async (accountId: number, id: string, attachmentFileName: string) => { - const response = await axiosInstance.get(`api/v1/download-attachment/${accountId}/${id}?name=${attachmentFileName}`, { responseType: 'blob' }); +export const download_attachment = async (accountId: number, id: string, content_hash: string, fileName: string) => { + const response = await axiosInstance.get(`api/v1/download-attachment/${accountId}/${id}?content_hash=${content_hash}`, { responseType: 'blob' }); const blob = new Blob([response.data]); - saveAs(blob, attachmentFileName); + saveAs(blob, fileName); }; -export const download_nested_attachment = async (accountId: number, id: string, attachmentFileName: string, nestedAttachmentFileName: string) => { - const response = await axiosInstance.get(`api/v1/download-nested-attachment/${accountId}/${id}?name=${attachmentFileName}&nested_name=${nestedAttachmentFileName}`, { responseType: 'blob' }); +export const download_nested_attachment = async (accountId: number, id: string, content_hash: string, nested_content_hash: string) => { + const response = await axiosInstance.get(`api/v1/download-nested-attachment/${accountId}/${id}?content_hash=${content_hash}&nested_content_hash=${nested_content_hash}`, { responseType: 'blob' }); const blob = new Blob([response.data]); - saveAs(blob, nestedAttachmentFileName); + saveAs(blob, nested_content_hash); }; export interface AttachmentInfo { /** MIME content type of the attachment (e.g., `image/png`, `application/pdf`). */ @@ -56,7 +56,10 @@ export interface AttachmentInfo { filename: string; /** Size of the attachment in bytes. */ size: number; + content_hash: string; + is_message: boolean } + export interface MessageContentResponse { text?: string; html?: string; @@ -84,8 +87,8 @@ export const load_message = async (accountId: number, id: string) => { return response.data; }; -export const load_nested_message = async (accountId: number, id: string, attachmentFileName: string) => { - const response = await axiosInstance.get(`api/v1/nested-message-content/${accountId}/${id}?name=${attachmentFileName}`); +export const load_nested_message = async (accountId: number, id: string, content_hash: string) => { + const response = await axiosInstance.get(`api/v1/nested-message-content/${accountId}/${id}?content_hash=${content_hash}`); return response.data; }; diff --git a/web/src/features/search/mail-list-table.tsx b/web/src/features/search/mail-list-table.tsx index 5a89c87..03451d3 100755 --- a/web/src/features/search/mail-list-table.tsx +++ b/web/src/features/search/mail-list-table.tsx @@ -282,7 +282,7 @@ export function MailListTable({ { id: "attachment_count", header: () => , - cell: ({ row }) => {row.original.attachment_count}, + cell: ({ row }) => {row.original.regular_attachment_count}, meta: { className: 'text-left text-xs' }, minSize: 40, maxSize: 40 diff --git a/web/src/features/search/mail-list.tsx b/web/src/features/search/mail-list.tsx index 82acf49..a3c747f 100644 --- a/web/src/features/search/mail-list.tsx +++ b/web/src/features/search/mail-list.tsx @@ -154,7 +154,7 @@ export function MailList({ )} {items.map((item, index) => { - const hasAttachments = item.attachment_count > 0 + const hasAttachments = item.regular_attachment_count > 0 const isSelectedRow = currentEnvelope?.id === item.id const isChecked = hasSelected(item.account_id, item.id) @@ -209,7 +209,7 @@ export function MailList({ {hasAttachments && (
- {item.attachment_count} + {item.regular_attachment_count}
)} diff --git a/web/src/features/search/mail-message-view.tsx b/web/src/features/search/mail-message-view.tsx index 125d7b6..bd3c24e 100644 --- a/web/src/features/search/mail-message-view.tsx +++ b/web/src/features/search/mail-message-view.tsx @@ -126,13 +126,13 @@ export function MailMessageView({ const [attachments, setAttachments] = useState(null); const [loading, setLoading] = useState(true); const [downloadingAttachmentFileName, setDownloadingAttachmentFileName] = useState(null); - const [nestedEmlFile, setNestedEmlFile] = useState(null); + const [nestedEmlFile, setNestedEmlFile] = useState(null); const { getEmailById } = useMinimalAccountList(); const [threadOpen, setThreadOpen] = useState(false); const downloadAttachmentMutation = useMutation({ - mutationFn: ({ fileName }: { fileName: string }) => - download_attachment(envelope.account_id, envelope.id, fileName), + mutationFn: ({ content_hash }: { content_hash: string }) => + download_attachment(envelope.account_id, envelope.id, content_hash, downloadingAttachmentFileName!), onSuccess: () => setDownloadingAttachmentFileName(null), onError: (error: any) => { setDownloadingAttachmentFileName(null); @@ -168,8 +168,8 @@ export function MailMessageView({ }, [envelope.id]); - const handleViewNestedEml = (filename: string) => { - setNestedEmlFile(filename); + const handleViewNestedEml = (attachment: AttachmentInfo) => { + setNestedEmlFile(attachment); }; const toggleToDelete = (accountId: number, mailId: string) => { @@ -317,7 +317,7 @@ export function MailMessageView({
{nonInline.map((attachment, i) => { const { icon, color } = getFileConfig(attachment.file_type); - const isNestedEmail = attachment.file_type.toLowerCase() === 'message/rfc822'; + const is_message = attachment.is_message; return
@@ -337,14 +337,16 @@ export function MailMessageView({
- {isNestedEmail && ( + {is_message && ( @@ -362,7 +364,7 @@ export function MailMessageView({ className="w-4 h-4 cursor-pointer" onClick={() => { setDownloadingAttachmentFileName(attachment.filename); - downloadAttachmentMutation.mutate({ fileName: attachment.filename }); + downloadAttachmentMutation.mutate({ content_hash: attachment.content_hash }); }} /> )} @@ -407,7 +409,8 @@ export function MailMessageView({ onOpenChange={(open: boolean) => !open && setNestedEmlFile(null)} accountId={envelope.account_id} envelopeId={envelope.id} - fileName={nestedEmlFile || ''} + fileName={nestedEmlFile?.filename || ''} + content_hash={nestedEmlFile?.content_hash} />
); diff --git a/web/src/features/search/nested-email-dialog.tsx b/web/src/features/search/nested-email-dialog.tsx index 237d717..9a5bde3 100644 --- a/web/src/features/search/nested-email-dialog.tsx +++ b/web/src/features/search/nested-email-dialog.tsx @@ -17,7 +17,7 @@ const MessageHeader = ({ }: { envelope: EmailEnvelope, attachments?: AttachmentInfo[], - onDownload: (fileName: string) => void + onDownload: (nested_content_hash: string) => void }) => { const { t } = useTranslation(); const displayAttachments = attachments || []; @@ -100,7 +100,7 @@ const MessageHeader = ({