External/bichon
1
0
Fork 0
mirror of https://github.com/rustmailer/bichon.git synced 2026-08-03 07:48:34 +02:00
Code Issues Packages Projects Releases Wiki Activity

refactor: use UUID for envelope id to prevent accidental deletion

Browse Source
This commit is contained in:
rustmailer
2026-03-18 01:04:41 +08:00
parent 8a42fcdb4a
commit d690f57290
37 changed files with 393 additions and 393 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/modules/common/mod.rs
-5
View File
@@ -92,11 +92,6 @@ impl<'x> From<&ImapAddress<'x>> for AddrVec {
}
}
// #[derive(Serialize)]
// pub struct ErrorResponse {
// pub message: String,
// }
#[inline]
fn create_rust_mailer_error(message: &str, code: ErrorCode) -> BichonError {
BichonError::Generic {
src/modules/dashboard/mod.rs
+1 -1
View File
@@ -113,5 +113,5 @@ pub struct Group {
pub struct LargestEmail {
pub subject: String, // Email subject
pub size_bytes: u64, // Email size in bytes
pub id: u64,
pub id: String,
}
src/modules/duckdb/build.rs
+39 -34
View File
@@ -16,7 +16,6 @@
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
use arrow::array::{BooleanArray, Int32Array, Int64Array, ListBuilder, StringBuilder, UInt64Array};
use arrow::datatypes::{DataType, Field, Schema};
use arrow::record_batch::RecordBatch;
@@ -30,10 +29,11 @@ pub fn build_record_batch(items: &[Envelope]) -> RecordBatch {
let capacity = items.len();
let schema = Arc::new(Schema::new(vec![
Field::new("id", DataType::UInt64, false),
Field::new("id", DataType::Utf8, false),
Field::new("account_id", DataType::UInt64, false),
Field::new("mailbox_id", DataType::UInt64, false),
Field::new("uid", DataType::UInt64, false),
Field::new("content_hash", DataType::Utf8, false),
Field::new("subject", DataType::Utf8, true),
Field::new("body", DataType::Utf8, true),
Field::new("sender", DataType::Utf8, true),
@@ -55,7 +55,7 @@ pub fn build_record_batch(items: &[Envelope]) -> RecordBatch {
Field::new("sent_at", DataType::Int64, true),
Field::new("received_at", DataType::Int64, true),
Field::new("size_bytes", DataType::UInt64, true),
Field::new("thread_id", DataType::UInt64, true),
Field::new("thread_id", DataType::Utf8, true),
Field::new("message_id", DataType::Utf8, true),
Field::new("has_attachment", DataType::Boolean, false),
Field::new("attachment_count", DataType::Int32, false),
@@ -67,10 +67,12 @@ pub fn build_record_batch(items: &[Envelope]) -> RecordBatch {
Field::new("shard_id", DataType::UInt64, false),
]));
let mut id_b = UInt64Array::builder(capacity);
let mut id_b = StringBuilder::with_capacity(capacity, capacity * 20);
let mut account_id_b = UInt64Array::builder(capacity);
let mut mailbox_id_b = UInt64Array::builder(capacity);
let mut uid_b = UInt64Array::builder(capacity);
let mut content_hash_b = StringBuilder::with_capacity(capacity, capacity * 64);
let mut subject_b = StringBuilder::with_capacity(capacity, capacity * 20);
let mut body_b = StringBuilder::with_capacity(capacity, capacity * 100);
let mut from_b = StringBuilder::with_capacity(capacity, capacity * 20);
@@ -82,7 +84,7 @@ pub fn build_record_batch(items: &[Envelope]) -> RecordBatch {
let mut date_b = Int64Array::builder(capacity);
let mut internal_date_b = Int64Array::builder(capacity);
let mut size_b = UInt64Array::builder(capacity);
let mut thread_id_b = UInt64Array::builder(capacity);
let mut thread_id_b = StringBuilder::with_capacity(capacity, capacity * 20);
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);
@@ -90,10 +92,11 @@ pub fn build_record_batch(items: &[Envelope]) -> RecordBatch {
let mut shard_id_b = UInt64Array::builder(capacity);
for e in items {
id_b.append_value(e.id);
id_b.append_value(&e.id);
account_id_b.append_value(e.account_id);
mailbox_id_b.append_value(e.mailbox_id);
uid_b.append_value(e.uid as u64);
content_hash_b.append_value(&e.content_hash);
subject_b.append_value(&e.subject);
body_b.append_value(&e.text);
from_b.append_value(&e.from);
@@ -116,7 +119,7 @@ pub fn build_record_batch(items: &[Envelope]) -> RecordBatch {
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);
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);
@@ -132,6 +135,7 @@ pub fn build_record_batch(items: &[Envelope]) -> RecordBatch {
Arc::new(account_id_b.finish()),
Arc::new(mailbox_id_b.finish()),
Arc::new(uid_b.finish()),
Arc::new(content_hash_b.finish()),
Arc::new(subject_b.finish()),
Arc::new(body_b.finish()),
Arc::new(from_b.finish()),
@@ -163,50 +167,46 @@ mod integration_tests {
conn.execute_batch(
r#"
CREATE TABLE IF NOT EXISTS envelopes (
-- internal id (tantivy f_id)
id UBIGINT NOT NULL,
-- account / mailbox / uid
id UUID PRIMARY KEY,
account_id UBIGINT NOT NULL,
mailbox_id UBIGINT NOT NULL,
uid UBIGINT NOT NULL,
-- headers / content
content_hash VARCHAR(64),
subject TEXT,
body TEXT,
sender TEXT,
recipients VARCHAR[],
cc VARCHAR[],
bcc VARCHAR[],
sender TEXT,
recipients VARCHAR[],
cc VARCHAR[],
bcc VARCHAR[],
-- dates
sent_at BIGINT,
received_at BIGINT,
received_at BIGINT,
-- size
size_bytes UBIGINT,
-- thread
thread_id UBIGINT,
-- message-id
size_bytes UBIGINT,
thread_id VARCHAR,
message_id TEXT,
-- attachment summary
has_attachment BOOLEAN NOT NULL,
attachment_count INTEGER NOT NULL CHECK (attachment_count >= 0),
tags VARCHAR[],
shard_id UBIGINT NOT NULL
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: 101,
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(),
@@ -216,7 +216,7 @@ mod integration_tests {
date: 1000,
internal_date: 1001,
size: 2048,
thread_id: 1,
thread_id: "1".to_string(),
message_id: "id123".to_string(),
attachment_count: 1,
tags: None,
@@ -231,13 +231,18 @@ mod integration_tests {
appender.flush()?;
}
let mut stmt = conn.prepare("SELECT subject, size_bytes FROM envelopes WHERE id = 101")?;
let mut rows = stmt.query([])?;
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(
@@ -247,13 +252,13 @@ mod integration_tests {
assert_eq!(count, 1);
let has_att: bool = conn.query_row(
"SELECT has_attachment FROM envelopes WHERE id = 101",
[],
"SELECT has_attachment FROM envelopes WHERE id = ?",
[&test_uuid],
|r| r.get(0),
)?;
assert!(has_att);
println!("Integration test for ingestion and query passed!");
println!("Integration test with content_hash passed!");
Ok(())
}
}
src/modules/duckdb/init.rs
+74 -44
View File
@@ -16,7 +16,6 @@
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
use chrono::{NaiveDateTime, Utc};
use duckdb::{params, types::Value, DuckdbConnectionManager};
use refinery::Runner;
@@ -249,6 +248,7 @@ impl DuckDBManager {
att.get_category(),
att.file_type,
att.size as u64,
env.content_hash.clone(), //这里是错误的，应该保存附件的content_hash
0,
])
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?;
@@ -291,7 +291,7 @@ impl DuckDBManager {
Ok(total)
}
pub fn num_messages_in_thread(&self, account_id: u64, thread_id: u64) -> BichonResult<u64> {
pub fn num_messages_in_thread(&self, account_id: u64, thread_id: String) -> BichonResult<u64> {
let conn = self.conn()?;
let count: u64 = conn
.query_row(
@@ -331,7 +331,7 @@ impl DuckDBManager {
pub fn get_envelope_by_id(
&self,
account_id: u64,
envelope_id: u64,
envelope_id: String,
) -> BichonResult<Option<Envelope>> {
let conn = self.conn()?;
let mut stmt = conn
@@ -352,36 +352,44 @@ impl DuckDBManager {
pub fn get_envelopes_by_ids(
&self,
account_id: u64,
envelope_ids: &[u64],
envelope_ids: &[&str],
) -> BichonResult<Vec<Envelope>> {
if envelope_ids.is_empty() {
return Ok(vec![]);
}
let conn = self.conn()?;
let ids_str = envelope_ids
.iter()
.map(|id| id.to_string())
.collect::<Vec<_>>()
.join(",");
let placeholders = vec!["?"; envelope_ids.len()].join(", ");
let query = format!(
"SELECT * FROM envelopes WHERE account_id = ? AND id IN ({})",
ids_str
placeholders
);
let mut stmt = conn
.prepare(&query)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?;
let mut stmt = conn.prepare(&query).map_err(|e| {
raise_error!(format!("Prepare error: {:#?}", e), ErrorCode::InternalError)
})?;
let mut params: Vec<Box<dyn duckdb::ToSql>> = Vec::new();
params.push(Box::new(account_id));
for id in envelope_ids {
params.push(Box::new(id.to_string()));
}
let param_refs: Vec<&dyn duckdb::ToSql> = params.iter().map(|p| p.as_ref()).collect();
let rows = stmt
.query_map(params![account_id], |row| Envelope::from_row(row))
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?;
.query_map(duckdb::params_from_iter(param_refs), |row| {
Envelope::from_row(row)
})
.map_err(|e| {
raise_error!(format!("Query error: {:#?}", e), ErrorCode::InternalError)
})?;
let mut result = Vec::new();
for row in rows {
result.push(
row.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?,
);
result.push(row.map_err(|e| {
raise_error!(format!("Row error: {:#?}", e), ErrorCode::InternalError)
})?);
}
Ok(result)
@@ -497,34 +505,46 @@ impl DuckDBManager {
pub fn update_envelope_tags(
&self,
updates: HashMap<u64, Vec<u64>>, // account_id -> [envelope_id1, ...]
updates: HashMap<u64, Vec<String>>,
tags: Vec<String>,
) -> BichonResult<()> {
let mut conn = self.conn()?;
let tags_json = serde_json::to_string(&tags)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?;
let tx = conn
.transaction()
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?;
for (account_id, ids) in updates {
if ids.is_empty() {
continue;
}
let ids_str = ids
.iter()
.map(|id| id.to_string())
.collect::<Vec<_>>()
.join(",");
let query = format!(
"UPDATE envelopes
SET tags = CAST(json(?) AS VARCHAR[])
WHERE account_id = ?
AND id IN ({})",
ids_str
);
tx.execute(&query, params![tags_json, account_id])
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?;
for chunk in ids.chunks(100) {
let placeholders = vec!["?"; chunk.len()].join(", ");
let query = format!(
"UPDATE envelopes
SET tags = CAST(json(?) AS VARCHAR[])
WHERE account_id = ?
AND id IN ({})",
placeholders
);
let mut params: Vec<Box<dyn duckdb::ToSql>> = Vec::new();
params.push(Box::new(tags_json.clone()));
params.push(Box::new(account_id));
for id in chunk {
params.push(Box::new(id.clone()));
}
let param_refs: Vec<&dyn duckdb::ToSql> =
params.iter().map(|p| p.as_ref()).collect();
tx.execute(&query, duckdb::params_from_iter(param_refs))
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?;
}
}
tx.commit()
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?;
Ok(())
@@ -532,31 +552,38 @@ impl DuckDBManager {
pub fn delete_envelopes_multi_account(
&self,
deletes: HashMap<u64, Vec<u64>>, // account_id -> [envelope_id1, ...]
deletes: HashMap<u64, Vec<String>>,
) -> BichonResult<()> {
let mut conn = self.conn()?;
let tx = conn
.transaction()
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?;
for (account_id, ids) in deletes {
if ids.is_empty() {
continue;
}
for chunk in ids.chunks(100) {
let ids_str = chunk
.iter()
.map(|id| id.to_string())
.collect::<Vec<_>>()
.join(",");
let placeholders = vec!["?"; chunk.len()].join(", ");
let mut params: Vec<Box<dyn duckdb::ToSql>> = Vec::new();
params.push(Box::new(account_id));
for id in chunk {
params.push(Box::new(id.clone()));
}
let param_refs: Vec<&dyn duckdb::ToSql> =
params.iter().map(|p| p.as_ref()).collect();
let query_params = duckdb::params_from_iter(param_refs);
let del_attachments_query = format!(
"DELETE FROM envelope_attachments
WHERE account_id = ?
AND envelope_id IN ({})",
ids_str
placeholders
);
tx.execute(&del_attachments_query, params![account_id])
tx.execute(&del_attachments_query, query_params.clone())
.map_err(|e| {
raise_error!(
format!("Delete attachments fail: {:#?}", e),
@@ -568,14 +595,17 @@ impl DuckDBManager {
"DELETE FROM envelopes
WHERE account_id = ?
AND id IN ({})",
ids_str
placeholders
);
tx.execute(&query, params![account_id])
tx.execute(&query, query_params)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?;
}
}
tx.commit()
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?;
Ok(())
}
@@ -1000,7 +1030,7 @@ impl DuckDBManager {
pub fn list_thread_envelopes(
&self,
account_id: u64,
thread_id: u64,
thread_id: String,
page: u64,
page_size: u64,
desc: bool,
src/modules/duckdb/migrations/V1__initial.sql
+9 -9
View File
@@ -3,7 +3,7 @@
-- =========================
CREATE TABLE IF NOT EXISTS envelopes (
-- internal id (tantivy f_id)
id UBIGINT NOT NULL,
id UUID PRIMARY KEY,
-- account / mailbox / uid
account_id UBIGINT NOT NULL,
@@ -11,6 +11,7 @@ CREATE TABLE IF NOT EXISTS envelopes (
uid UBIGINT NOT NULL,
-- headers / content
content_hash VARCHAR(64) NOT NULL,
subject TEXT,
body TEXT,
@@ -27,7 +28,7 @@ CREATE TABLE IF NOT EXISTS envelopes (
size_bytes UBIGINT,
-- thread
thread_id UBIGINT,
thread_id VARCHAR NOT NULL,
-- message-id
message_id TEXT,
@@ -39,8 +40,7 @@ CREATE TABLE IF NOT EXISTS envelopes (
shard_id UBIGINT NOT NULL
);
CREATE INDEX idx_env_mailbox_sent ON envelopes(account_id, mailbox_id, sent_at);
CREATE INDEX IF NOT EXISTS idx_env_mailbox_sent ON envelopes(account_id, mailbox_id, sent_at);
-- =========================
-- envelope_attachments
--
@@ -48,9 +48,9 @@ CREATE INDEX idx_env_mailbox_sent ON envelopes(account_id, mailbox_id, sent_at);
-- =========================
CREATE TABLE IF NOT EXISTS envelope_attachments (
-- Reference to envelopes.id
envelope_id UBIGINT NOT NULL,
account_id UBIGINT NOT NULL,
mailbox_id UBIGINT NOT NULL,
envelope_id UUID NOT NULL,
account_id UBIGINT NOT NULL,
mailbox_id UBIGINT NOT NULL,
-- Original attachment filename (for display)
filename TEXT NOT NULL,
@@ -63,8 +63,8 @@ CREATE TABLE IF NOT EXISTS envelope_attachments (
-- Attachment size in bytes
-- 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_mailbox_id ON envelope_attachments (mailbox_id);
CREATE INDEX IF NOT EXISTS idx_attachments_env_id ON envelope_attachments (envelope_id);
src/modules/duckdb/mod.rs
+2 -1
View File
@@ -16,7 +16,8 @@
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
pub mod build;
pub mod init;
pub mod refinery;
#[cfg(test)]
mod tests;
src/modules/duckdb/tests.rs
+28
View File
@@ -0,0 +1,28 @@
use duckdb::Connection;
#[test]
fn test_get_envelopes_with_params_iter() {
let conn = Connection::open_in_memory().unwrap();
conn.execute("CREATE TABLE envelopes (id VARCHAR, account_id UBIGINT)", []).unwrap();
conn.execute("INSERT INTO envelopes VALUES ('mail_1', 1), ('mail_2', 1), ('mail_3', 2)", []).unwrap();
let search_ids = vec!["mail_1", "mail_2"];
let acc_id = 1u64;
let placeholders = search_ids.iter().map(|_| "?").collect::<Vec<_>>().join(",");
let query = format!("SELECT id FROM envelopes WHERE account_id = ? AND id IN ({})", placeholders);
let mut params: Vec<&dyn duckdb::ToSql> = Vec::new();
params.push(&acc_id);
for id in &search_ids {
params.push(id);
}
let mut stmt = conn.prepare(&query).unwrap();
let res: Vec<String> = stmt.query_map(duckdb::params_from_iter(params), |r| r.get(0)).unwrap()
.map(|r| r.unwrap()).collect();
println!("{:#?}", &res);
assert_eq!(res.len(), 2);
assert!(res.contains(&"mail_1".to_string()));
}
src/modules/envelope/extractor.rs
+19 -15
View File
@@ -21,12 +21,13 @@ use crate::modules::envelope::utils::normalize_subject;
use crate::modules::error::code::ErrorCode;
use crate::modules::error::BichonResult;
use crate::modules::message::content::AttachmentInfo;
use crate::modules::utils::create_hash2;
use crate::modules::utils::html::extract_text;
use crate::{calculate_hash, raise_error, utc_now};
use crate::modules::utils::{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 uuid::Uuid;
pub fn extract_envelope(
fetch: &Fetch,
@@ -84,6 +85,7 @@ fn extract_envelope_core(
account_id: u64,
mailbox_id: u64,
) -> BichonResult<(Envelope, Vec<AttachmentInfo>)> {
let content_hash = content_hash(body);
let message = MessageParser::new().parse(body).ok_or_else(|| {
raise_error!(
"Email header parse result is not available".into(),
@@ -173,7 +175,7 @@ fn extract_envelope_core(
.collect();
let envelope = Envelope {
id: create_hash2(account_id, mailbox_id, &message_id),
id: Uuid::new_v4().to_string(),
message_id,
account_id,
mailbox_id,
@@ -192,6 +194,7 @@ fn extract_envelope_core(
tags: None,
account_email: None,
mailbox_name: None,
content_hash,
};
Ok((envelope, attachments))
@@ -248,11 +251,11 @@ pub fn extract_envelope_from_message(
.unwrap_or_else(|| "unknown".to_string());
let envelope = Envelope {
id: 0,
id: Default::default(),
message_id,
account_id,
mailbox_id: 0,
uid: 0,
mailbox_id: Default::default(),
uid: Default::default(),
subject,
text,
from,
@@ -260,13 +263,14 @@ pub fn extract_envelope_from_message(
cc,
bcc,
date,
internal_date: 0,
size: 0,
internal_date: Default::default(),
size: Default::default(),
thread_id,
attachment_count: 0,
tags: None,
account_email: None,
mailbox_name: None,
attachment_count: Default::default(),
tags: Default::default(),
account_email: Default::default(),
mailbox_name: Default::default(),
content_hash: Default::default(),
};
Ok(envelope)
@@ -276,11 +280,11 @@ pub fn compute_thread_id(
in_reply_to: Option<String>,
references: Option<Vec<String>>,
message_id: &str,
) -> u64 {
) -> String {
if in_reply_to.is_some() && references.as_ref().map_or(false, |r| !r.is_empty()) {
return calculate_hash!(&references.as_ref().unwrap()[0]);
return hex_hash(&references.as_ref().unwrap()[0]);
}
calculate_hash!(message_id)
hex_hash(message_id)
}
pub fn generate_message_id() -> String {
src/modules/imap/executor.rs
+8 -51
View File
@@ -25,7 +25,6 @@ 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::utils::create_hash;
use crate::modules::{error::BichonResult, imap::manager::ImapConnectionManager};
use crate::raise_error;
use async_imap::types::Name;
@@ -201,22 +200,19 @@ impl ImapExecutor {
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::ImapCommandFailed))?;
let mut count = 0;
let fields = SchemaTools::eml_fields();
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 eml_id = create_hash(account_id, &envelope.0.message_id);
ENVELOPE_INDEX_MANAGER
.add_document(envelope.0.id, envelope)
.await;
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( eml_id, doc!(fields.f_id => eml_id, fields.f_account_id => account_id, fields.f_mailbox_id => mailbox_id, fields.f_eml => body)).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 => body)).await;
count += 1;
}
Ok(count)
@@ -238,62 +234,23 @@ impl ImapExecutor {
.uid_fetch(uid_set, BODY_FETCH_COMMAND)
.await
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::ImapCommandFailed))?;
let fields = SchemaTools::eml_fields();
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 eml_id = create_hash(account_id, &envelope.0.message_id);
ENVELOPE_INDEX_MANAGER
.add_document(envelope.0.id, envelope)
.await;
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( eml_id, doc!(fields.f_id => eml_id, fields.f_account_id => account_id, fields.f_mailbox_id => mailbox_id, fields.f_eml => body)).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 => body)).await;
}
Ok(())
}
// async fn get_connection(
// &self,
// ) -> BichonResult<bb8::PooledConnection<'_, ImapConnectionManager>> {
// match self.pool.get().await {
// Ok(connection) => Ok(connection),
// Err(e) => match e {
// RunError::User(e) => Err(e),
// RunError::TimedOut => {
// let state = self.pool.state();
// tracing::warn!(
// "{}: connections={}, idle={}, \
// get_started={}, get_direct={}, get_waited={}, get_timed_out={}, \
// wait_time_ms={}, created={}, closed_broken={}, closed_invalid={}, \
// closed_lifetime={}, closed_idle={}",
// self.account_id,
// state.connections,
// state.idle_connections,
// state.statistics.get_started,
// state.statistics.get_direct,
// state.statistics.get_waited,
// state.statistics.get_timed_out,
// state.statistics.get_wait_time.as_millis(),
// state.statistics.connections_created,
// state.statistics.connections_closed_broken,
// state.statistics.connections_closed_invalid,
// state.statistics.connections_closed_max_lifetime,
// state.statistics.connections_closed_idle_timeout,
// );
// return Err(raise_error!(
// "Timed out while attempting to acquire a connection from the pool".into(),
// ErrorCode::ConnectionPoolTimeout
// ));
// }
// },
// }
// }
pub async fn create_connection(
account_id: u64,
) -> BichonResult<Session<Box<dyn SessionStream>>> {
src/modules/import/mod.rs
+6 -6
View File
@@ -112,7 +112,7 @@ impl ImportEmls {
},
};
let fields = SchemaTools::eml_fields();
let fields = SchemaTools::fields();
let account_id = account.id;
let mut success_count = 0;
let mut failed_details: Vec<FailedEmlDetail> = Vec::new(); // Store failure details
@@ -148,19 +148,19 @@ impl ImportEmls {
continue;
}
};
let eml_id = create_hash(account_id, &envelope.0.message_id);
let content_hash = envelope.0.content_hash.clone();
ENVELOPE_INDEX_MANAGER
.add_document(envelope.0.id, envelope)
.add_document(envelope)
.await;
EML_INDEX_MANAGER
.add_document(
eml_id,
content_hash.clone(),
doc!(
fields.f_id => eml_id,
fields.f_id => content_hash,
fields.f_account_id => account_id,
fields.f_mailbox_id => mailbox_id,
fields.f_eml => decoded
fields.f_blob => decoded
),
)
.await;
src/modules/indexer/envelope.rs
+5 -3
View File
@@ -24,7 +24,7 @@ use crate::modules::{account::migration::AccountModel, cache::imap::mailbox::Mai
#[derive(Clone, Debug, Default, Eq, PartialEq, Deserialize, Serialize, Object)]
pub struct Envelope {
pub id: u64,
pub id: String,
pub message_id: String,
pub account_id: u64,
pub account_email: Option<String>,
@@ -40,9 +40,10 @@ pub struct Envelope {
pub date: i64,
pub internal_date: i64,
pub size: u32,
pub thread_id: u64,
pub thread_id: String,
pub attachment_count: usize,
pub tags: Option<Vec<String>>,
pub content_hash: String,
}
impl Envelope {
@@ -97,7 +98,7 @@ impl Envelope {
date: row.get("sent_at").unwrap_or(0),
internal_date: row.get("received_at").unwrap_or(0),
size: row.get::<_, u64>("size_bytes")? as u32,
thread_id: row.get("thread_id").unwrap_or(0),
thread_id: row.get("thread_id")?,
attachment_count: row.get::<_, i32>("attachment_count")? as usize,
tags: {
let t = get_list("tags");
@@ -107,6 +108,7 @@ impl Envelope {
Some(t)
}
},
content_hash: row.get("content_hash")?,
})
}
}
src/modules/indexer/fields.rs
+3 -40
View File
@@ -16,54 +16,17 @@
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
use tantivy::schema::Field;
pub const F_MESSAGE_ID: &str = "message_id";
pub const F_ACCOUNT_ID: &str = "account_id";
pub const F_MAILBOX_ID: &str = "mailbox_id";
pub const F_UID: &str = "uid";
pub const F_SUBJECT: &str = "subject";
pub const F_TEXT: &str = "text";
pub const F_FROM: &str = "from";
pub const F_TO: &str = "to";
pub const F_CC: &str = "cc";
pub const F_BCC: &str = "bcc";
pub const F_DATE: &str = "date";
pub const F_INTERNAL_DATE: &str = "internal_date";
pub const F_SIZE: &str = "size";
pub const F_THREAD_ID: &str = "thread_id";
pub const F_ATTACHMENTS: &str = "attachments";
pub const F_HAS_ATTACHMENT: &str = "has_attachment";
pub const F_TAGS: &str = "tags";
pub const F_ID: &str = "id";
pub struct EnvelopeFields {
pub f_id: Field,
pub f_message_id: Field,
pub f_account_id: Field,
pub f_mailbox_id: Field,
pub f_uid: Field,
pub f_subject: Field,
pub f_text: Field,
pub f_from: Field,
pub f_to: Field,
pub f_cc: Field,
pub f_bcc: Field,
pub f_date: Field,
pub f_internal_date: Field,
pub f_size: Field,
pub f_thread_id: Field,
pub f_attachments: Field,
pub f_has_attachment: Field,
pub f_tags: Field,
}
pub const F_BLOB: &str = "blob";
pub const F_EML: &str = "eml";
pub struct EmlFields {
pub struct BlobFields {
pub f_id: Field,
pub f_account_id: Field,
pub f_mailbox_id: Field,
pub f_eml: Field,
pub f_blob: Field,
}
src/modules/indexer/manager.rs
+72 -64
View File
@@ -27,7 +27,6 @@ use crate::modules::{
duckdb::init::duckdb,
message::{content::AttachmentInfo, search::SortBy, tags::TagCount},
settings::cli::SETTINGS,
utils::create_hash,
};
use crate::{
modules::{
@@ -70,12 +69,12 @@ pub const EML_BATCH_SIZE: usize = 100;
const MAX_BUFFER_DURATION: Duration = Duration::from_secs(10);
pub enum MetadataOp {
Record((u64, (Envelope, Vec<AttachmentInfo>))),
Record((Envelope, Vec<AttachmentInfo>)),
Shutdown,
}
pub enum DocumentOp {
Document((u64, TantivyDocument)),
Document((String, TantivyDocument)),
Shutdown,
}
@@ -87,16 +86,16 @@ impl EnvelopeIndexManager {
pub fn new() -> Self {
let (sender, mut receiver) = mpsc::channel::<MetadataOp>(1000);
task::spawn(async move {
let mut buffer: HashMap<u64, (Envelope, Vec<AttachmentInfo>)> =
HashMap::with_capacity(ENVELOPE_BATCH_SIZE);
let mut buffer: Vec<(Envelope, Vec<AttachmentInfo>)> =
Vec::with_capacity(ENVELOPE_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(MetadataOp::Record((eid, doc))) => {
buffer.insert(eid, doc);
Some(MetadataOp::Record(doc)) => {
buffer.push(doc);
if buffer.len() >= ENVELOPE_BATCH_SIZE {
ENVELOPE_INDEX_MANAGER.drain_and_commit(&mut buffer).await;
}
@@ -122,16 +121,16 @@ impl EnvelopeIndexManager {
Self { sender }
}
pub async fn add_document(&self, eid: u64, doc: (Envelope, Vec<AttachmentInfo>)) {
let _ = self.sender.send(MetadataOp::Record((eid, doc))).await;
pub async fn add_document(&self, doc: (Envelope, Vec<AttachmentInfo>)) {
let _ = self.sender.send(MetadataOp::Record(doc)).await;
}
async fn drain_and_commit(&self, buffer: &mut HashMap<u64, (Envelope, Vec<AttachmentInfo>)>) {
async fn drain_and_commit(&self, buffer: &mut Vec<(Envelope, Vec<AttachmentInfo>)>) {
if buffer.is_empty() {
return;
}
let items: Vec<(Envelope, Vec<AttachmentInfo>)> = buffer.drain().map(|(_, v)| v).collect();
let items: Vec<(Envelope, Vec<AttachmentInfo>)> = buffer.drain(..).collect();
let result = (|| -> BichonResult<()> {
duckdb()?.append_envelopes_with_attachments(&items)?;
@@ -195,7 +194,7 @@ impl EnvelopeIndexManager {
pub async fn delete_envelopes_multi_account(
&self,
deletes: HashMap<u64, Vec<u64>>, // HashMap<account_id, envelope_ids>
deletes: HashMap<u64, Vec<String>>, // HashMap<account_id, envelope_ids>
) -> BichonResult<()> {
if deletes.is_empty() {
tracing::warn!("delete_envelopes_multi_account: deletes is empty, nothing to delete");
@@ -209,7 +208,7 @@ impl EnvelopeIndexManager {
pub async fn update_envelope_tags(
&self,
updates: HashMap<u64, Vec<u64>>, // HashMap<account_id, envelope_ids>
updates: HashMap<u64, Vec<String>>, // HashMap<account_id, envelope_ids>
tags: Vec<String>,
) -> BichonResult<()> {
if updates.is_empty() {
@@ -259,7 +258,7 @@ impl EnvelopeIndexManager {
pub async fn list_thread_envelopes(
&self,
account_id: u64,
thread_id: u64,
thread_id: String,
page: u64,
page_size: u64,
desc: bool,
@@ -276,7 +275,7 @@ impl EnvelopeIndexManager {
pub async fn get_envelope_by_id(
&self,
account_id: u64,
envelope_id: u64,
envelope_id: String,
) -> BichonResult<Option<Envelope>> {
tokio::task::spawn_blocking(move || duckdb()?.get_envelope_by_id(account_id, envelope_id))
.await
@@ -313,7 +312,7 @@ impl EnvelopeIndexManager {
pub async fn num_messages_in_thread(
&self,
account_id: u64,
thread_id: u64,
thread_id: String,
) -> BichonResult<u64> {
tokio::task::spawn_blocking(move || duckdb()?.num_messages_in_thread(account_id, thread_id))
.await
@@ -371,7 +370,8 @@ impl EmlIndexManager {
});
let (sender, mut receiver) = mpsc::channel::<DocumentOp>(100);
task::spawn(async move {
let mut buffer: HashMap<u64, TantivyDocument> = HashMap::with_capacity(EML_BATCH_SIZE);
let mut buffer: HashMap<String, TantivyDocument> =
HashMap::with_capacity(EML_BATCH_SIZE);
let mut interval = tokio::time::interval(MAX_BUFFER_DURATION);
let mut shutdown = SIGNAL_MANAGER.subscribe();
loop {
@@ -421,8 +421,11 @@ impl EmlIndexManager {
/// 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, eid: u64, doc: TantivyDocument) {
let _ = self.sender.send(DocumentOp::Document((eid, doc))).await;
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 {
@@ -441,7 +444,7 @@ impl EmlIndexManager {
panic!("Failed to create index directory {:?}: {}", index_dir, e)
});
IndexBuilder::new()
.schema(SchemaTools::eml_schema())
.schema(SchemaTools::schema())
.settings(IndexSettings {
docstore_compression: Compressor::Zstd(ZstdCompressor {
compression_level: Some(SETTINGS.bichon_eml_compression_level as i32),
@@ -460,13 +463,13 @@ impl EmlIndexManager {
}
}
fn envelope_query(&self, account_id: u64, eid: u64) -> Box<dyn Query> {
fn envelope_query(&self, account_id: u64, eid: &str) -> Box<dyn Query> {
let account_id_query = TermQuery::new(
Term::from_field_u64(SchemaTools::eml_fields().f_account_id, account_id),
Term::from_field_u64(SchemaTools::fields().f_account_id, account_id),
IndexRecordOption::Basic,
);
let envelope_id_query = TermQuery::new(
Term::from_field_u64(SchemaTools::eml_fields().f_id, eid),
Term::from_field_text(SchemaTools::fields().f_id, eid),
IndexRecordOption::Basic,
);
let boolean_query = BooleanQuery::new(vec![
@@ -476,7 +479,7 @@ impl EmlIndexManager {
Box::new(boolean_query)
}
pub async fn get(&self, account_id: u64, eml_id: u64) -> BichonResult<Option<Vec<u8>>> {
pub async fn get(&self, account_id: u64, eml_id: &str) -> BichonResult<Option<Vec<u8>>> {
let searcher = self.reader.searcher();
let query = self.envelope_query(account_id, eml_id);
let docs = searcher
@@ -492,8 +495,8 @@ impl EmlIndexManager {
.doc_async(*doc_address)
.await
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?;
let fields = SchemaTools::eml_fields();
let value = doc.get_first(fields.f_eml).ok_or_else(|| {
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
@@ -509,25 +512,27 @@ impl EmlIndexManager {
Ok(Some(bytes.to_vec()))
}
pub async fn get_reader(&self, account_id: u64, eid: u64) -> BichonResult<File> {
pub async fn get_reader(&self, account_id: u64, eid: String) -> BichonResult<File> {
let envelope = duckdb()?
.get_envelope_by_id(account_id, eid)?
.get_envelope_by_id(account_id, eid.clone())?
.ok_or_else(|| {
raise_error!(
format!(
"Email envelope not found: account_id={} id={}",
account_id, eid
account_id, &eid
),
ErrorCode::ResourceNotFound
)
})?;
let eml_id = create_hash(account_id, &envelope.message_id);
let data = self.get(account_id, eml_id).await?.ok_or_else(|| {
raise_error!(
format!("Eml not found: account_id={}, eid={}", 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"));
@@ -548,32 +553,34 @@ impl EmlIndexManager {
pub async fn get_attachment_content(
&self,
account_id: u64,
eid: u64,
eid: String,
file_name: &str,
) -> BichonResult<Vec<u8>> {
let envelope = duckdb()?
.get_envelope_by_id(account_id, eid)?
.get_envelope_by_id(account_id, eid.clone())?
.ok_or_else(|| {
raise_error!(
format!(
"Email envelope not found: account_id={} id={}",
account_id, eid
account_id, &eid
),
ErrorCode::ResourceNotFound
)
})?;
let eml_id = create_hash(account_id, &envelope.message_id);
let data = self.get(account_id, eml_id).await?.ok_or_else(|| {
raise_error!(
format!("Email not found: account_id={}, eid={}", 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
account_id, &eid
),
ErrorCode::InternalError
)
@@ -600,11 +607,11 @@ impl EmlIndexManager {
pub async fn get_attachment(
&self,
account_id: u64,
eid: u64,
eid: String,
file_name: &str,
) -> BichonResult<File> {
let content = self
.get_attachment_content(account_id, eid, file_name)
.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"));
@@ -625,19 +632,19 @@ impl EmlIndexManager {
pub async fn get_nested_attachment(
&self,
account_id: u64,
eid: u64,
eid: String,
file_name: &str,
nested_file_name: &str,
) -> BichonResult<File> {
let content = self
.get_attachment_content(account_id, eid, file_name)
.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
account_id, &eid
),
ErrorCode::InternalError
)
@@ -655,7 +662,7 @@ impl EmlIndexManager {
raise_error!(
format!(
"Nested attachment '{}' not found in email {}",
nested_file_name, eid
nested_file_name, &eid
),
ErrorCode::ResourceNotFound
)
@@ -678,7 +685,7 @@ impl EmlIndexManager {
}
fn account_query(&self, account_id: u64) -> Box<TermQuery> {
let account_term = Term::from_field_u64(SchemaTools::eml_fields().f_account_id, account_id);
let account_term = Term::from_field_u64(SchemaTools::fields().f_account_id, account_id);
Box::new(TermQuery::new(account_term, IndexRecordOption::Basic))
}
@@ -696,11 +703,11 @@ impl EmlIndexManager {
fn mailbox_query(&self, account_id: u64, mailbox_id: u64) -> Box<dyn Query> {
let account_query = TermQuery::new(
Term::from_field_u64(SchemaTools::eml_fields().f_account_id, account_id),
Term::from_field_u64(SchemaTools::fields().f_account_id, account_id),
IndexRecordOption::Basic,
);
let mailbox_query = TermQuery::new(
Term::from_field_u64(SchemaTools::eml_fields().f_mailbox_id, mailbox_id),
Term::from_field_u64(SchemaTools::fields().f_mailbox_id, mailbox_id),
IndexRecordOption::Basic,
);
let boolean_query = BooleanQuery::new(vec![
@@ -737,7 +744,7 @@ impl EmlIndexManager {
pub async fn delete_email_multi_account(
&self,
deletes: &HashMap<u64, Vec<u64>>, // HashMap<account_id, envelope_ids>
deletes: &HashMap<u64, Vec<String>>, // HashMap<account_id, envelope_ids>
) -> BichonResult<()> {
if deletes.is_empty() {
tracing::warn!("delete_email_multi_account: deletes is empty, nothing to delete");
@@ -747,10 +754,10 @@ impl EmlIndexManager {
let mut writer = self.index_writer.lock().await;
for (account_id, envelope_ids) in deletes {
let unique_ids: Vec<u64> = envelope_ids
let unique_ids: Vec<&str> = envelope_ids
.iter()
.copied()
.collect::<HashSet<_>>()
.map(|s| s.as_str())
.collect::<HashSet<&str>>()
.into_iter()
.collect();
if unique_ids.is_empty() {
@@ -759,7 +766,8 @@ impl EmlIndexManager {
for chunk in unique_ids.chunks(100) {
let envelopes = duckdb()?.get_envelopes_by_ids(*account_id, chunk)?;
let found_ids_set: HashSet<u64> = envelopes.iter().map(|e| e.id).collect();
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!(
@@ -770,7 +778,7 @@ impl EmlIndexManager {
}
for envelope in envelopes {
let hashed_id = create_hash(*account_id, &envelope.message_id);
let hashed_id = &envelope.content_hash;
let query = self.envelope_query(*account_id, hashed_id);
writer
@@ -786,7 +794,7 @@ impl EmlIndexManager {
Ok(())
}
async fn drain_and_commit(&self, buffer: &mut HashMap<u64, TantivyDocument>) {
async fn drain_and_commit(&self, buffer: &mut HashMap<String, TantivyDocument>) {
if buffer.is_empty() {
return;
}
@@ -794,7 +802,7 @@ impl EmlIndexManager {
let mut operations = Vec::new();
for (eid, doc) in buffer.drain() {
let delete_term = Term::from_field_u64(SchemaTools::eml_fields().f_id, eid);
let delete_term = Term::from_field_text(SchemaTools::fields().f_id, &eid);
operations.push(UserOperation::Delete(delete_term));
operations.push(UserOperation::Add(doc));
}
src/modules/indexer/schema.rs
+12 -12
View File
@@ -19,37 +19,37 @@
use std::sync::{Arc, LazyLock};
use crate::modules::indexer::fields::*;
use tantivy::schema::INDEXED;
use tantivy::schema::{Schema, FAST, STORED};
use tantivy::schema::{INDEXED, STRING};
static EML_FIELDS: LazyLock<Arc<EmlFields>> = LazyLock::new(|| {
let (_, fields) = SchemaTools::create_eml_schema();
static BLOB_FIELDS: LazyLock<Arc<BlobFields>> = LazyLock::new(|| {
let (_, fields) = SchemaTools::create_schema();
Arc::new(fields)
});
pub struct SchemaTools;
impl SchemaTools {
pub fn eml_schema() -> Schema {
let (schema, _) = Self::create_eml_schema();
pub fn schema() -> Schema {
let (schema, _) = Self::create_schema();
schema
}
pub fn eml_fields() -> &'static EmlFields {
&EML_FIELDS
pub fn fields() -> &'static BlobFields {
&BLOB_FIELDS
}
pub fn create_eml_schema() -> (Schema, EmlFields) {
pub fn create_schema() -> (Schema, BlobFields) {
let mut builder = Schema::builder();
let f_id = builder.add_u64_field(F_ID, INDEXED | FAST);
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_eml = builder.add_bytes_field(F_EML, STORED);
let fields = EmlFields {
let f_blob = builder.add_bytes_field(F_BLOB, STORED);
let fields = BlobFields {
f_id,
f_account_id,
f_mailbox_id,
f_eml,
f_blob,
};
(builder.build(), fields)
}
src/modules/message/append.rs
+16 -14
View File
@@ -5,7 +5,6 @@ use crate::{
error::{code::ErrorCode, BichonResult},
imap::executor::ImapExecutor,
indexer::manager::{EML_INDEX_MANAGER, ENVELOPE_INDEX_MANAGER},
utils::create_hash,
},
raise_error,
};
@@ -16,16 +15,16 @@ const MAX_RESTORE_COUNT: usize = 100;
#[derive(Clone, Debug, Default, Eq, PartialEq, Deserialize, Serialize, Object)]
pub struct RestoreMessagesRequest {
/// Message IDs to restore (max 100)
pub message_ids: Vec<u64>,
/// envelope IDs to restore (max 100)
pub envelope_ids: Vec<String>,
}
pub async fn restore_emails(account_id: u64, message_ids: Vec<u64>) -> BichonResult<()> {
if message_ids.len() > MAX_RESTORE_COUNT {
pub async fn restore_emails(account_id: u64, envelope_ids: Vec<String>) -> BichonResult<()> {
if envelope_ids.len() > MAX_RESTORE_COUNT {
return Err(raise_error!(
format!(
"Too many messages to restore: {} (max {})",
message_ids.len(),
envelope_ids.len(),
MAX_RESTORE_COUNT
),
ErrorCode::InvalidParameter
@@ -42,27 +41,30 @@ pub async fn restore_emails(account_id: u64, message_ids: Vec<u64>) -> BichonRes
let mut failed = Vec::new();
let mut session = ImapExecutor::create_connection(account_id).await?;
for message_id in message_ids {
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, message_id)
.get_envelope_by_id(account_id, eid)
.await?
.ok_or_else(|| {
raise_error!(
format!(
"Envelope not found: account_id={} message_id={}",
account_id, message_id
account_id, &envelope_id
),
ErrorCode::ResourceNotFound
)
})?;
let eml_id = create_hash(account_id, &envelope.message_id);
let eml = EML_INDEX_MANAGER
.get(account_id, eml_id)
.get(account_id, &envelope.content_hash)
.await?
.ok_or_else(|| {
raise_error!(
format!("Eml not found: account_id={} id={}", account_id, message_id),
format!(
"Eml not found: account_id={} id={}",
account_id, &envelope_id
),
ErrorCode::ResourceNotFound
)
})?;
@@ -83,13 +85,13 @@ pub async fn restore_emails(account_id: u64, message_ids: Vec<u64>) -> BichonRes
.await;
if let Err(err) = result {
failed.push(message_id);
tracing::warn!(
account_id = account_id,
message_id = message_id,
message_id = &envelope_id,
error = ?err,
"Failed to restore email"
);
failed.push(envelope_id);
}
}
src/modules/message/content.rs
+22 -11
View File
@@ -22,7 +22,6 @@ use crate::modules::envelope::extractor::extract_envelope_from_message;
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::create_hash;
use crate::{modules::error::BichonResult, raise_error};
use mail_parser::{MessageParser, MimeHeaders};
@@ -144,29 +143,32 @@ pub struct FullNestedMessageContent {
pub envelope: Envelope,
}
pub async fn retrieve_email_content(account_id: u64, id: u64) -> BichonResult<FullMessageContent> {
pub async fn retrieve_email_content(
account_id: u64,
envelope_id: String,
) -> BichonResult<FullMessageContent> {
AccountModel::check_account_exists(account_id).await?;
let envelope = ENVELOPE_INDEX_MANAGER
.get_envelope_by_id(account_id, id)
.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, id
account_id, &envelope_id
),
ErrorCode::ResourceNotFound
)
})?;
let eml_id = create_hash(account_id, &envelope.message_id);
let eml = EML_INDEX_MANAGER
.get(account_id, eml_id)
.get(account_id, &envelope.content_hash)
.await?
.ok_or_else(|| {
raise_error!(
format!(
"Email record not found: account_id={} id={}",
account_id, id
account_id, &envelope_id
),
ErrorCode::ResourceNotFound
)
@@ -175,7 +177,7 @@ pub async fn retrieve_email_content(account_id: u64, id: u64) -> BichonResult<Fu
raise_error!(
format!(
"Failed to parse EML data (id={}) — the message may be corrupted.",
id
&envelope_id
),
ErrorCode::InternalError
)
@@ -186,14 +188,23 @@ pub async fn retrieve_email_content(account_id: u64, id: u64) -> BichonResult<Fu
for attachment in message.attachments() {
let content_type = attachment.content_type().ok_or_else(|| {
raise_error!(
format!("Attachment is missing Content-Type (email id={})", id),
format!(
"Attachment is missing Content-Type (email id={})",
&envelope_id
),
ErrorCode::InternalError
)
})?;
let filename = attachment
.attachment_name()
.map(|name| name.to_string())
.unwrap_or_else(|| format!("email{}_attachment{}", id, attachment.raw_body_offset()));
.unwrap_or_else(|| {
format!(
"email{}_attachment{}",
&envelope_id,
attachment.raw_body_offset()
)
});
let disposition = attachment.content_disposition();
@@ -242,7 +253,7 @@ pub async fn retrieve_email_content(account_id: u64, id: u64) -> BichonResult<Fu
pub async fn retrieve_nested_eml_content(
account_id: u64,
envelope_id: u64,
envelope_id: String,
name: &str,
) -> BichonResult<FullNestedMessageContent> {
let attachment_content = EML_INDEX_MANAGER
src/modules/message/delete.rs
+1 -1
View File
@@ -20,7 +20,7 @@ use crate::modules::error::BichonResult;
use crate::modules::indexer::manager::{EML_INDEX_MANAGER, ENVELOPE_INDEX_MANAGER};
use std::collections::HashMap;
pub async fn delete_messages_impl(request: HashMap<u64, Vec<u64>>) -> BichonResult<()> {
pub async fn delete_messages_impl(request: HashMap<u64, Vec<String>>) -> BichonResult<()> {
EML_INDEX_MANAGER
.delete_email_multi_account(&request)
.await?;
src/modules/message/list.rs
+1 -2
View File
@@ -16,7 +16,6 @@
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
use crate::{
modules::{
account::migration::AccountModel,
@@ -58,7 +57,7 @@ fn validate_pagination_params(page: u64, page_size: u64) -> BichonResult<()> {
pub async fn get_thread_messages(
account_id: u64,
thread_id: u64,
thread_id: String,
page: u64,
page_size: u64,
) -> BichonResult<DataPage<Envelope>> {
src/modules/message/tags.rs
+1 -1
View File
@@ -24,7 +24,7 @@ use serde::{Deserialize, Serialize};
#[derive(Debug, Clone, Default, Eq, PartialEq, Serialize, Deserialize, Object)]
pub struct UpdateTagsRequest {
pub updates: HashMap<u64, Vec<u64>>, // account_id -> envelope_ids
pub updates: HashMap<u64, Vec<String>>, // account_id -> envelope_ids
pub tags: Vec<String>,
}
src/modules/rest/api/message.rs
+13 -12
View File
@@ -58,7 +58,7 @@ impl MessageApi {
async fn delete_messages(
&self,
/// specifying the mailbox and messages to delete.
payload: Json<HashMap<u64, Vec<u64>>>,
payload: Json<HashMap<u64, Vec<String>>>,
context: ClientContext,
) -> ApiResult<()> {
let request = payload.0;
@@ -130,7 +130,7 @@ impl MessageApi {
/// The ID of the account owning the mailbox.
account_id: Path<u64>,
// Thread ID
thread_id: Query<u64>,
thread_id: Query<String>,
/// The page number for pagination (1-based).
page: Query<u64>,
/// The number of messages per page.
@@ -158,7 +158,7 @@ impl MessageApi {
/// The ID of the account.
account_id: Path<u64>,
/// The ID of the message to fetch.
envelope_id: Path<u64>,
envelope_id: Path<String>,
context: ClientContext,
) -> ApiResult<Json<FullMessageContent>> {
let account_id = account_id.0;
@@ -181,7 +181,7 @@ impl MessageApi {
/// The ID of the account.
account_id: Path<u64>,
/// The ID of the message to fetch.
envelope_id: Path<u64>,
envelope_id: Path<String>,
name: Query<String>,
context: ClientContext,
) -> ApiResult<Json<FullNestedMessageContent>> {
@@ -206,21 +206,22 @@ impl MessageApi {
/// The ID of the account.
account_id: Path<u64>,
/// The ID of the message.
envelope_id: Path<u64>,
envelope_id: Path<String>,
context: ClientContext,
) -> ApiResult<Json<Envelope>> {
let account_id = account_id.0;
context
.require_permission(Some(account_id), Permission::DATA_READ)
.await?;
let envelope_id = envelope_id.0;
let envelope = ENVELOPE_INDEX_MANAGER
.get_envelope_by_id(account_id, envelope_id.0)
.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.0
account_id, &envelope_id
),
ErrorCode::ResourceNotFound
)
@@ -239,7 +240,7 @@ impl MessageApi {
/// The ID of the account.
account_id: Path<u64>,
/// The ID of the message to download.
envelope_id: Path<u64>,
envelope_id: Path<String>,
context: ClientContext,
) -> ApiResult<Attachment<Body>> {
let account_id = account_id.0;
@@ -249,7 +250,7 @@ impl MessageApi {
.await?;
let envelope_id = envelope_id.0;
let reader = EML_INDEX_MANAGER
.get_reader(account_id, envelope_id)
.get_reader(account_id, envelope_id.clone())
.await?;
let body = Body::from_async_read(reader);
let attachment = Attachment::new(body)
@@ -275,7 +276,7 @@ impl MessageApi {
context
.require_permission(Some(account_id), Permission::DATA_EXPORT_BATCH)
.await?;
Ok(restore_emails(account_id, payload.0.message_ids).await?)
Ok(restore_emails(account_id, payload.0.envelope_ids).await?)
}
/// Downloads a specific attachment from an email. Requires `name` query parameter.
@@ -289,7 +290,7 @@ impl MessageApi {
/// The ID of the account.
account_id: Path<u64>,
/// The ID of the message containing the attachment.
envelope_id: Path<u64>,
envelope_id: Path<String>,
/// The filename of the attachment to download.
name: Query<String>,
context: ClientContext,
@@ -321,7 +322,7 @@ impl MessageApi {
/// The ID of the account.
account_id: Path<u64>,
/// The ID of the message containing the attachment.
envelope_id: Path<u64>,
envelope_id: Path<String>,
/// The filename of the attachment to download.
name: Query<String>,
nested_name: Query<String>,
src/modules/smtp/server.rs
+6 -8
View File
@@ -606,7 +606,7 @@ async fn read_data<R: AsyncBufReadExt + Unpin>(reader: &mut R) -> io::Result<Vec
}
async fn parse_email(data: &[u8], session: &Session) -> BichonResult<()> {
let fields = SchemaTools::eml_fields();
let fields = SchemaTools::fields();
let rcpt = match session.rcpt_to.first() {
Some(r) => r,
None => {
@@ -646,19 +646,17 @@ async fn parse_email(data: &[u8], session: &Session) -> BichonResult<()> {
e
})?;
let eml_id = create_hash(rcpt.id, &envelope.0.message_id);
ENVELOPE_INDEX_MANAGER
.add_document(envelope.0.id, envelope)
.await;
let content_hash = envelope.0.content_hash.clone();
ENVELOPE_INDEX_MANAGER.add_document(envelope).await;
EML_INDEX_MANAGER
.add_document(
eml_id,
content_hash.clone(),
doc!(
fields.f_id => eml_id,
fields.f_id => content_hash,
fields.f_account_id => rcpt.id,
fields.f_mailbox_id => mailbox_id,
fields.f_eml => data
fields.f_blob => data
),
)
.await;
src/modules/token/mod.rs
+2 -2
View File
@@ -21,7 +21,7 @@ use std::collections::HashMap;
use super::error::code::ErrorCode;
use crate::modules::database::manager::DB_MANAGER;
use crate::modules::database::{
async_find_impl, delete_impl, async_filter_by_secondary_key_impl, with_transaction,
async_filter_by_secondary_key_impl, async_find_impl, delete_impl, with_transaction,
};
use crate::modules::database::{insert_impl, list_all_impl, update_impl};
use crate::modules::settings::cli::SETTINGS;
@@ -187,7 +187,7 @@ impl AccessTokenModel {
Some(token) => token,
None => {
return Err(raise_error!(
"Permission denied: no valid access token provided.".into(),
"Invalid access token provided. Please check your credentials.".into(),
ErrorCode::PermissionDenied
))
}
src/modules/utils/mod.rs
+14 -8
View File
@@ -107,7 +107,7 @@ macro_rules! raise_error {
$crate::modules::error::BichonError::Generic {
message: $msg,
code: $code,
location: snafu::location!()
location: snafu::location!(),
}
};
}
@@ -243,13 +243,6 @@ pub fn validate_email(email: &str) -> crate::modules::error::BichonResult<()> {
Ok(())
}
#[macro_export]
macro_rules! calculate_hash {
($name:expr) => {
$crate::modules::utils::hash($name)
};
}
#[macro_export]
macro_rules! id {
($bit_strength:expr) => {{
@@ -277,6 +270,14 @@ pub fn hash(s: &str) -> u64 {
(hash & 0x1F_FFFF_FFFF_FFFF) as u64
}
pub fn hex_hash(s: &str) -> String {
let mut cursor = Vec::new();
cursor.extend_from_slice(s.as_bytes());
let mut cursor = std::io::Cursor::new(cursor);
let hash = murmur3::murmur3_x64_128(&mut cursor, 0).unwrap();
format!("{:032x}", hash)
}
pub fn create_hash(account_id: u64, field: &str) -> u64 {
let mut buffer = Vec::new();
buffer.extend_from_slice(&account_id.to_le_bytes());
@@ -370,3 +371,8 @@ pub fn validate_tag(tag: &str) -> Result<(), String> {
Ok(())
}
pub fn content_hash(content: &[u8]) -> String {
let hash = blake3::hash(content);
hash.to_hex().to_string()
}