refactor!: replace Tantivy search engine with DuckDB

This commit is contained in:
rustmailer
2026-03-03 12:30:26 +08:00
parent ef4ab3496e
commit d2936ed4a7
52 changed files with 3594 additions and 1821 deletions
Generated
+1158 -111
View File
File diff suppressed because it is too large Load Diff
+27 -16
View File
@@ -1,6 +1,6 @@
[package]
name = "bichon"
version = "0.3.7"
version = "1.0.0"
edition = "2021"
[[bin]]
@@ -27,8 +27,8 @@ opt-level = 3
codegen-units = 1
[dependencies]
chrono = "0.4.43"
clap = { version = "4.5.56", features = ["derive", "env"] }
chrono = "0.4.44"
clap = { version = "4.5.60", features = ["derive", "env"] }
mimalloc = "0.1.48"
native_db = "0.8.2"
itertools = "0.14.0"
@@ -62,35 +62,35 @@ reqwest = { version = "0.12.24", default-features = false, features = [
] }
tokio-socks = "0.5.2"
http = "1.4.0"
regex = "1.12.2"
regex = "1.12.3"
email_address = "0.2.9"
futures = "0.3.31"
futures = "0.3.32"
utf7-imap = "0.3.2"
imap-proto = "0.16.6"
mail-parser = { version = '0.11.1', features = ["serde"] }
mail-parser = { version = '0.11.2', features = ["serde"] }
# mail-send = "0.5.2"
tokio-rustls = { version = "0.26.4", default-features = false, features = [
"ring",
"tls12",
] }
timeago = "0.5.0"
timeago = "0.6.0"
ahash = "0.8.12"
oauth2 = { version = "5.0.0", features = ["reqwest-blocking"] }
url = { version = "2.5.8", features = ["serde"] }
sysinfo = "0.38.0"
sysinfo = "0.38.2"
num_cpus = "1.17.0"
cacache = { version = "13.1.0", default-features = false, features = [
"tokio-runtime",
"mmap",
] }
rand = "0.9.2"
rand = "0.10.0"
encoding_rs = "0.8.35"
async-imap = { version = "0.11.1", default-features = false, features = [
async-imap = { version = "0.11.2", default-features = false, features = [
"runtime-tokio",
"compress",
] }
webpki-roots = "1.0.5"
rustls = { version = "0.23.36", default-features = false, features = ["ring"] }
webpki-roots = "1.0.6"
rustls = { version = "0.23.37", default-features = false, features = ["ring"] }
rustls-pki-types = "1.14.0"
tokio-io-timeout = "1.2.1"
semver = "1.0.27"
@@ -98,7 +98,7 @@ governor = "0.10.4"
lru = "0.16.3"
mime_guess = "2.0.5"
hex = "0.4.3"
time = { version = "0.3.46", features = [
time = { version = "0.3.47", features = [
"formatting",
"parsing",
"local-offset",
@@ -114,16 +114,27 @@ gethostname = "1.1.0"
tantivy = { version = "0.25.0", features = ["quickwit", "zstd-compression"] }
itoa = "1.0.17"
html2text = "0.16.7"
bytes = "1.11.0"
bytes = "1.11.1"
dialoguer = "0.12.0"
console = "0.16.2"
toml = "0.9.8"
memmap2 = "0.9.9"
memmap2 = "0.9.10"
outlook-pst = { git = "https://github.com/rustmailer/outlook-pst-rs.git", branch = "main" }
compressed-rtf = "1.0.0"
codepage-strings = "1.0.2"
mail-send = "0.5.2"
duckdb = { version = "1.4.4", features = [
"chrono",
"bundled",
"r2d2",
"appender-arrow",
] }
arrow = { version = "56.2.0", features = ["ffi"] }
r2d2 = { version = "0.8", default-features = false }
refinery = { version = "0.9", default-features = false }
refinery-core = { version = "0.9", default-features = false }
[dev-dependencies]
#bincode = "1.3.3"
#secret-lib = "1.0.0"
tempfile = "3.24.0"
tempfile = "3.26.0"
+3
View File
@@ -1,6 +1,9 @@
use std::{io::Result, process::Command};
fn main() -> Result<()> {
if cfg!(target_os = "windows") {
println!("cargo:rustc-link-lib=Rstrtmgr");
}
let output = Command::new("git")
.args(&["rev-parse", "--short", "HEAD"])
.output()
+2
View File
@@ -21,6 +21,7 @@ use bichon::{
modules::{
common::rustls::RustMailerTls,
context::{executors::BichonContext, Initialize},
duckdb::init::DuckDBManager,
error::BichonResult,
logger,
rest::start_http_server,
@@ -69,6 +70,7 @@ async fn initialize() -> BichonResult<()> {
// SETTINGS.validate()?;
SignalManager::initialize().await?;
DataDirManager::initialize().await?;
DuckDBManager::initialize().await?;
UserManager::initialize().await?;
RustMailerTls::initialize().await?;
BichonContext::initialize().await?;
+1 -1
View File
@@ -58,7 +58,7 @@ impl BatchAccountRoleRequest {
}
for id in &self.account_ids {
let exists = AccountModel::find(*id).await?; // Assuming an exists helper
let exists = AccountModel::async_find(*id).await?; // Assuming an exists helper
if exists.is_none() {
return Err(raise_error!(
format!("Account ID {} not found", id),
+146 -24
View File
@@ -33,7 +33,7 @@ use crate::{
state::AccountRunningState,
},
cache::imap::mailbox::MailBox,
database::{list_all_impl, with_transaction},
database::{list_all_impl, secondary_find_impl, with_transaction},
error::BichonResult,
indexer::manager::{EML_INDEX_MANAGER, ENVELOPE_INDEX_MANAGER},
users::{role::DEFAULT_ACCOUNT_MANAGER_ROLE_ID, UserModel, DEFAULT_ADMIN_USER_ID},
@@ -51,14 +51,14 @@ use crate::modules::database::count_by_unique_secondary_key_impl;
use crate::modules::database::delete_impl;
use crate::modules::database::manager::DB_MANAGER;
use crate::modules::database::{
paginate_query_primary_scan_all_impl, secondary_find_impl, update_impl,
async_secondary_find_impl, paginate_query_primary_scan_all_impl, update_impl,
};
use crate::modules::error::code::ErrorCode;
use crate::modules::oauth2::token::OAuth2AccessToken;
use crate::modules::rest::response::DataPage;
use crate::raise_error;
pub type AccountModel = AccountV3;
pub type AccountModel = AccountV4;
#[derive(Clone, Debug, Default, Eq, PartialEq, Deserialize, Serialize, Enum)]
pub enum AccountType {
@@ -158,6 +158,42 @@ impl AccountV3 {
fn pk(&self) -> String {
format!("{}_{}", self.created_at, self.id)
}
}
#[derive(Clone, Debug, Default, Eq, PartialEq, Deserialize, Serialize, Object)]
#[native_model(id = 4, version = 4, from = AccountV3)]
#[native_db(primary_key(pk -> String))]
pub struct AccountV4 {
#[secondary_key(unique)]
pub id: u64,
pub imap: Option<ImapConfig>,
pub enabled: bool,
#[oai(validator(custom = "crate::modules::common::validator::EmailValidator"))]
pub email: String,
pub name: Option<String>,
pub capabilities: Option<Vec<String>>,
pub date_since: Option<DateSince>,
pub date_before: Option<RelativeDate>,
pub folder_limit: Option<u32>,
pub sync_folders: Option<Vec<String>>,
pub account_type: AccountType,
pub sync_interval_min: Option<i64>,
pub sync_batch_size: Option<u32>,
pub known_folders: Option<BTreeSet<String>>,
pub created_at: i64,
pub updated_at: i64,
pub created_by: u64, //user id
pub use_proxy: Option<u64>,
pub use_dangerous: bool,
pub pgp_key: Option<String>,
pub imap_daily_quota_bytes: Option<u32>,
pub auto_sync_new_mailboxes: Option<bool>,
}
impl AccountV4 {
fn pk(&self) -> String {
format!("{}_{}", self.created_at, self.id)
}
pub fn new(user_id: u64, request: AccountCreateRequest) -> BichonResult<Self> {
Ok(Self {
@@ -181,25 +217,30 @@ impl AccountV3 {
created_by: user_id,
sync_batch_size: request.sync_batch_size,
date_before: request.date_before,
imap_daily_quota_bytes: request.imap_daily_quota_bytes,
auto_sync_new_mailboxes: request.auto_sync_new_mailboxes,
})
}
pub async fn check_account_exists(account_id: u64) -> BichonResult<AccountModel> {
let account =
secondary_find_impl::<AccountModel>(DB_MANAGER.meta_db(), AccountV3Key::id, account_id)
.await?
.ok_or_else(|| {
raise_error!(
format!("Account id='{account_id}' not found"),
ErrorCode::ResourceNotFound
)
})?;
let account = async_secondary_find_impl::<AccountModel>(
DB_MANAGER.meta_db(),
AccountV4Key::id,
account_id,
)
.await?
.ok_or_else(|| {
raise_error!(
format!("Account id='{account_id}' not found"),
ErrorCode::ResourceNotFound
)
})?;
Ok(account)
}
/// Fetches an `AccountEntity` by its `id`.
pub async fn get(account_id: u64) -> BichonResult<AccountModel> {
let result: AccountModel = Self::find(account_id).await?.ok_or_else(|| {
pub async fn async_get(account_id: u64) -> BichonResult<AccountModel> {
let result: AccountModel = Self::async_find(account_id).await?.ok_or_else(|| {
raise_error!(
format!("Account with ID '{account_id}' not found"),
ErrorCode::ResourceNotFound
@@ -208,9 +249,27 @@ impl AccountV3 {
Ok(result)
}
pub async fn find(account_id: u64) -> BichonResult<Option<AccountModel>> {
secondary_find_impl::<AccountModel>(DB_MANAGER.meta_db(), AccountV3Key::id, account_id)
.await
pub fn get(account_id: u64) -> BichonResult<AccountModel> {
let result: AccountModel = Self::find(account_id)?.ok_or_else(|| {
raise_error!(
format!("Account with ID '{account_id}' not found"),
ErrorCode::ResourceNotFound
)
})?;
Ok(result)
}
pub async fn async_find(account_id: u64) -> BichonResult<Option<AccountModel>> {
async_secondary_find_impl::<AccountModel>(
DB_MANAGER.meta_db(),
AccountV4Key::id,
account_id,
)
.await
}
pub fn find(account_id: u64) -> BichonResult<Option<AccountModel>> {
secondary_find_impl::<AccountModel>(DB_MANAGER.meta_db(), AccountV4Key::id, account_id)
}
pub async fn create_account(
@@ -258,7 +317,7 @@ impl AccountV3 {
request: AccountUpdateRequest,
validate: bool,
) -> BichonResult<()> {
let account = AccountModel::get(account_id).await?;
let account = AccountModel::async_get(account_id).await?;
if validate {
request.validate_update_request(&account)?;
}
@@ -273,7 +332,7 @@ impl AccountV3 {
}
pub async fn delete(account_id: u64) -> BichonResult<()> {
let account = Self::get(account_id).await?;
let account = Self::async_get(account_id).await?;
if let Err(error) = Self::cleanup_account_resources_sequential(&account).await {
tracing::error!(
"[CLEANUP_ACCOUNT_ERROR] Account {}: failed to cleanup resources: {:#?}",
@@ -287,7 +346,7 @@ impl AccountV3 {
async fn delete_account(account_id: u64) -> BichonResult<()> {
delete_impl(DB_MANAGER.meta_db(), move|rw|{
rw.get().secondary::<AccountModel>(AccountV3Key::id, account_id).map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
rw.get().secondary::<AccountModel>(AccountV4Key::id, account_id).map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
.ok_or_else(||raise_error!(format!("The account entity with id={account_id} that you want to delete was not found."), ErrorCode::ResourceNotFound))
}).await
}
@@ -317,7 +376,7 @@ impl AccountV3 {
sync_folders: Vec<String>,
) -> BichonResult<()> {
update_impl(DB_MANAGER.meta_db(), move |rw| {
rw.get().secondary::<AccountModel>(AccountV3Key::id, account_id).map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
rw.get().secondary::<AccountModel>(AccountV4Key::id, account_id).map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
.ok_or_else(|| raise_error!(format!("When trying to update account sync_folders, the corresponding record was not found. account_id={}", account_id), ErrorCode::ResourceNotFound))
}, |current|{
let mut updated = current.clone();
@@ -332,7 +391,7 @@ impl AccountV3 {
known_folders: BTreeSet<String>,
) -> BichonResult<()> {
update_impl(DB_MANAGER.meta_db(), move |rw| {
rw.get().secondary::<AccountModel>(AccountV3Key::id, account_id).map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
rw.get().secondary::<AccountModel>(AccountV4Key::id, account_id).map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
.ok_or_else(|| raise_error!(format!("When trying to update account known_folders, the corresponding record was not found. account_id={}", account_id), ErrorCode::ResourceNotFound))
}, |current|{
let mut updated = current.clone();
@@ -347,7 +406,7 @@ impl AccountV3 {
capabilities: Vec<String>,
) -> BichonResult<()> {
update_impl(DB_MANAGER.meta_db(), move |rw| {
rw.get().secondary::<AccountModel>(AccountV3Key::id, account_id).map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
rw.get().secondary::<AccountModel>(AccountV4Key::id, account_id).map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
.ok_or_else(|| raise_error!(format!("When trying to update account capabilities, the corresponding record was not found. account_id={}", account_id), ErrorCode::ResourceNotFound))
}, |current|{
let mut updated = current.clone();
@@ -378,7 +437,7 @@ impl AccountV3 {
}
pub async fn count() -> BichonResult<usize> {
count_by_unique_secondary_key_impl::<AccountModel>(DB_MANAGER.meta_db(), AccountV3Key::id)
count_by_unique_secondary_key_impl::<AccountModel>(DB_MANAGER.meta_db(), AccountV4Key::id)
.await
}
@@ -483,6 +542,13 @@ impl AccountV3 {
new.pgp_key = Some(pgp_key);
}
if let Some(imap_daily_quota_bytes) = request.imap_daily_quota_bytes {
new.imap_daily_quota_bytes = Some(imap_daily_quota_bytes);
}
if let Some(auto_sync_new_mailboxes) = request.auto_sync_new_mailboxes {
new.auto_sync_new_mailboxes = Some(auto_sync_new_mailboxes);
}
new.updated_at = utc_now!();
Ok(new)
}
@@ -584,3 +650,59 @@ impl From<AccountV2> for AccountV3 {
}
}
}
impl From<AccountV4> for AccountV3 {
fn from(value: AccountV4) -> Self {
Self {
id: value.id,
imap: value.imap,
enabled: value.enabled,
email: value.email,
name: value.name,
capabilities: value.capabilities,
date_since: value.date_since,
date_before: value.date_before,
folder_limit: value.folder_limit,
sync_folders: value.sync_folders,
account_type: value.account_type,
sync_interval_min: value.sync_interval_min,
sync_batch_size: value.sync_batch_size,
known_folders: value.known_folders,
created_at: value.created_at,
updated_at: value.updated_at,
created_by: value.created_by,
use_proxy: value.use_proxy,
use_dangerous: value.use_dangerous,
pgp_key: value.pgp_key,
}
}
}
impl From<AccountV3> for AccountV4 {
fn from(value: AccountV3) -> Self {
Self {
id: value.id,
imap: value.imap,
enabled: value.enabled,
email: value.email,
name: value.name,
capabilities: value.capabilities,
date_since: value.date_since,
date_before: value.date_before,
folder_limit: value.folder_limit,
sync_folders: value.sync_folders,
account_type: value.account_type,
sync_interval_min: value.sync_interval_min,
sync_batch_size: value.sync_batch_size,
known_folders: value.known_folders,
created_at: value.created_at,
updated_at: value.updated_at,
created_by: value.created_by,
use_proxy: value.use_proxy,
use_dangerous: value.use_dangerous,
pgp_key: value.pgp_key,
imap_daily_quota_bytes: None,
auto_sync_new_mailboxes: None,
}
}
}
+4
View File
@@ -44,6 +44,8 @@ pub struct AccountCreateRequest {
pub use_proxy: Option<u64>,
pub use_dangerous: bool,
pub pgp_key: Option<String>,
pub imap_daily_quota_bytes: Option<u32>,
pub auto_sync_new_mailboxes: Option<bool>,
}
impl AccountCreateRequest {
@@ -155,6 +157,8 @@ pub struct AccountUpdateRequest {
pub use_dangerous: Option<bool>,
pub pgp_key: Option<String>,
pub imap_daily_quota_bytes: Option<u32>,
pub auto_sync_new_mailboxes: Option<bool>,
}
impl AccountUpdateRequest {
+2
View File
@@ -54,6 +54,7 @@ pub struct AccountResp {
pub use_proxy: Option<u64>,
pub use_dangerous: bool,
pub pgp_key: Option<String>,
pub imap_daily_quota_bytes: Option<u32>,
}
impl AccountResp {
@@ -86,6 +87,7 @@ impl AccountResp {
use_proxy: account.use_proxy,
use_dangerous: account.use_dangerous,
pgp_key: account.pgp_key,
imap_daily_quota_bytes: account.imap_daily_quota_bytes,
}
}
}
+18 -22
View File
@@ -20,8 +20,9 @@ use crate::{
decode_mailbox_name, encode_mailbox_name,
modules::{
database::{
async_find_impl, batch_delete_impl, batch_insert_impl, batch_upsert_impl, delete_impl,
filter_by_secondary_key_impl, manager::DB_MANAGER,
async_filter_by_secondary_key_impl, async_find_impl, batch_delete_impl,
batch_insert_impl, batch_upsert_impl, delete_impl, filter_by_secondary_key_impl,
manager::DB_MANAGER,
},
error::{code::ErrorCode, BichonResult},
},
@@ -71,24 +72,6 @@ impl MailBox {
encode_mailbox_name!(&self.name)
}
// pub async fn batch_delete(mailboxes: Vec<MailBox>) -> BichonResult<()> {
// batch_delete_impl(DB_MANAGER.envelope_db(), move |rw| {
// let mut to_deleted = Vec::new();
// for mailbox in mailboxes {
// let retrived = rw
// .get()
// .primary::<MailBox>(mailbox.id)
// .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?;
// if let Some(retrived) = retrived {
// to_deleted.push(retrived);
// }
// }
// Ok(to_deleted)
// })
// .await?;
// Ok(())
// }
pub async fn get(id: u64) -> BichonResult<MailBox> {
let result = async_find_impl::<MailBox>(DB_MANAGER.envelope_db(), id).await?;
Ok(result.ok_or_else(|| {
@@ -110,8 +93,21 @@ impl MailBox {
}
pub async fn list_all(account_id: u64) -> BichonResult<Vec<MailBox>> {
filter_by_secondary_key_impl(DB_MANAGER.envelope_db(), MailBoxKey::account_id, account_id)
.await
async_filter_by_secondary_key_impl(
DB_MANAGER.envelope_db(),
MailBoxKey::account_id,
account_id,
)
.await
}
pub fn find_mailbox(account_id: u64, mailbox_id: u64) -> BichonResult<Option<MailBox>> {
let all: Vec<MailBox> = filter_by_secondary_key_impl(
DB_MANAGER.envelope_db(),
MailBoxKey::account_id,
account_id,
)?;
Ok(all.into_iter().find(|m| m.id == mailbox_id))
}
pub async fn batch_insert(mailboxes: &[MailBox]) -> BichonResult<()> {
+1 -1
View File
@@ -62,7 +62,7 @@ pub async fn get_sync_folders(
mailboxes.iter().map(|(m, _)| m.name.clone()).collect(),
)
.await?;
let account = AccountModel::get(account.id).await?;
let account = AccountModel::async_get(account.id).await?;
let subscribed = &account.sync_folders.unwrap_or_default();
let is_noselect = |mailbox: &MailBox| {
mailbox
+1 -1
View File
@@ -54,7 +54,7 @@ impl AccountSyncTask {
let task = move |param: Option<u64>| {
let account_id = param.unwrap();
Box::pin(async move {
let account = AccountModel::get(account_id).await.ok();
let account = AccountModel::async_get(account_id).await.ok();
match account {
Some(account) => {
if !account.enabled {
+1 -4
View File
@@ -16,12 +16,9 @@
// 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 std::sync::LazyLock;
use crate::modules::{
context::Initialize, error::BichonResult, utils::shutdown::shutdown_signal,
};
use crate::modules::{context::Initialize, error::BichonResult, utils::shutdown::shutdown_signal};
use tokio::sync::broadcast;
pub static SIGNAL_MANAGER: LazyLock<SignalManager> = LazyLock::new(SignalManager::new);
+5 -34
View File
@@ -20,7 +20,6 @@ use crate::modules::users::permissions::Permission;
use poem_openapi::Object;
use serde::{Deserialize, Serialize};
use std::collections::HashSet;
use tantivy::{schema::Value, TantivyDocument};
use crate::{
bichon_version,
@@ -28,7 +27,7 @@ use crate::{
account::migration::AccountModel,
common::auth::ClientContext,
error::{code::ErrorCode, BichonResult},
indexer::{manager::ENVELOPE_INDEX_MANAGER, schema::SchemaTools},
indexer::manager::ENVELOPE_INDEX_MANAGER,
settings::dir::DATA_DIR_MANAGER,
utils::get_total_size,
},
@@ -65,11 +64,11 @@ impl DashboardStats {
};
let mut stat = ENVELOPE_INDEX_MANAGER
.get_dashboard_stats(&authorized_ids)
.get_dashboard_stats(authorized_ids.clone())
.await?;
stat.top_largest_emails = ENVELOPE_INDEX_MANAGER
.top_10_largest_emails(&authorized_ids)
.top_10_largest_emails(authorized_ids.clone())
.await?;
stat.account_count = if has_all_accounts {
@@ -78,7 +77,7 @@ impl DashboardStats {
authorized_ids.as_ref().map(|ids| ids.len()).unwrap_or(0)
};
stat.email_count = ENVELOPE_INDEX_MANAGER.total_emails(&authorized_ids)?;
stat.email_count = ENVELOPE_INDEX_MANAGER.total_emails(authorized_ids).await?;
if has_all_accounts {
stat.storage_usage_bytes = get_total_size(&DATA_DIR_MANAGER.eml_dir)
@@ -114,33 +113,5 @@ pub struct Group {
pub struct LargestEmail {
pub subject: String, // Email subject
pub size_bytes: u64, // Email size in bytes
}
impl LargestEmail {
pub fn from_tantivy_doc(document: &TantivyDocument) -> BichonResult<Self> {
let fields = SchemaTools::envelope_fields();
let value = document.get_first(fields.f_size).ok_or_else(|| {
raise_error!(
"miss 'size' field in tantivy document".into(),
ErrorCode::InternalError
)
})?;
let size_bytes = value.as_u64().ok_or_else(|| {
raise_error!("'size' field is not a u64".into(), ErrorCode::InternalError)
})?;
let value = document.get_first(fields.f_subject).ok_or_else(|| {
raise_error!("'subject' field not found".into(), ErrorCode::InternalError)
})?;
let subject = value.as_str().map(|s| s.to_string()).ok_or_else(|| {
raise_error!(
"'subject' field is not a string".into(),
ErrorCode::InternalError
)
})?;
let envelope = LargestEmail {
subject,
size_bytes,
};
Ok(envelope)
}
pub id: u64,
}
+40 -42
View File
@@ -16,7 +16,7 @@
// 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::{AccountV1, AccountV2, AccountV3};
use crate::modules::account::migration::{AccountV1, AccountV2, AccountV3, AccountV4};
use crate::modules::autoconfig::CachedMailSettings;
use crate::modules::error::code::ErrorCode;
use crate::modules::error::BichonResult;
@@ -67,6 +67,7 @@ impl ModelsAdapter {
self.register_model::<AccountV1>();
self.register_model::<AccountV2>();
self.register_model::<AccountV3>();
self.register_model::<AccountV4>();
self.register_model::<OAuth2>();
self.register_model::<OAuth2PendingEntity>();
self.register_model::<OAuth2AccessToken>();
@@ -188,30 +189,6 @@ pub async fn update_impl<T: ToInput + Clone + std::fmt::Debug + Send + 'static>(
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
}
// pub async fn batch_update_impl<T: ToInput + Clone + std::fmt::Debug + Send + 'static>(
// database: &Arc<Database<'static>>,
// filter: impl FnOnce(&RwTransaction) -> RustMailerResult<Vec<T>> + Send + 'static,
// updated: impl FnOnce(&Vec<T>) -> RustMailerResult<Vec<(T, T)>> + Send + 'static,
// ) -> RustMailerResult<Vec<T>> {
// let db = database.clone();
// tokio::task::spawn_blocking(move || {
// let rw = db
// .rw_transaction()
// .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?;
// let targets = filter(&rw)?;
// let tuples = updated(&targets)?;
// for (old, updated) in tuples {
// rw.update(old, updated)
// .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?;
// }
// rw.commit()
// .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?;
// Ok(targets)
// })
// .await
// .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
// }
pub async fn async_find_impl<T: ToInput + Clone + Send + 'static>(
database: &Arc<Database<'static>>,
key: impl ToKey + Send + 'static,
@@ -231,21 +208,6 @@ pub async fn async_find_impl<T: ToInput + Clone + Send + 'static>(
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
}
// pub fn find_impl<T: ToInput + Clone + Send + 'static>(
// database: &Arc<Database<'static>>,
// key: &str,
// ) -> BichonResult<Option<T>> {
// let db = database.clone();
// let r_transaction = db
// .r_transaction()
// .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?;
// let entity: Option<T> = r_transaction
// .get()
// .primary(key)
// .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?;
// Ok(entity)
// }
pub async fn delete_impl<T: ToInput + Clone + Send + 'static>(
database: &Arc<Database<'static>>,
delete: impl FnOnce(&RwTransaction) -> BichonResult<T> + Send + 'static,
@@ -423,7 +385,7 @@ pub async fn paginate_query_primary_scan_all_impl<
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
}
pub async fn filter_by_secondary_key_impl<T: ToInput + Clone + Send + 'static>(
pub async fn async_filter_by_secondary_key_impl<T: ToInput + Clone + Send + 'static>(
database: &Arc<Database<'static>>,
key_def: impl ToKeyDefinition<KeyOptions> + Send + 'static,
start_with: impl ToKey + Send + 'static,
@@ -447,6 +409,26 @@ pub async fn filter_by_secondary_key_impl<T: ToInput + Clone + Send + 'static>(
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
}
pub fn filter_by_secondary_key_impl<T: ToInput + Clone + Send + 'static>(
database: &Arc<Database<'static>>,
key_def: impl ToKeyDefinition<KeyOptions> + Send + 'static,
start_with: impl ToKey + Send + 'static,
) -> BichonResult<Vec<T>> {
let db = database.clone();
let r_transaction = db
.r_transaction()
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?;
let entities: Vec<T> = r_transaction
.scan()
.secondary(key_def)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
.start_with(start_with)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
.try_collect()
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?;
Ok(entities)
}
pub async fn count_by_unique_secondary_key_impl<T: ToInput + Clone + Send + 'static>(
database: &Arc<Database<'static>>,
key_def: impl ToKeyDefinition<KeyOptions> + Send + 'static,
@@ -469,7 +451,7 @@ pub async fn count_by_unique_secondary_key_impl<T: ToInput + Clone + Send + 'sta
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
}
pub async fn secondary_find_impl<T: ToInput + Clone + Send + 'static>(
pub async fn async_secondary_find_impl<T: ToInput + Clone + Send + 'static>(
database: &Arc<Database<'static>>,
key_def: impl ToKeyDefinition<KeyOptions> + Send + 'static,
key: impl ToKey + Send + 'static,
@@ -491,6 +473,22 @@ pub async fn secondary_find_impl<T: ToInput + Clone + Send + 'static>(
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
}
pub fn secondary_find_impl<T: ToInput + Clone + Send + 'static>(
database: &Arc<Database<'static>>,
key_def: impl ToKeyDefinition<KeyOptions> + Send + 'static,
key: impl ToKey + Send + 'static,
) -> BichonResult<Option<T>> {
let db = database.clone();
let r_transaction = db
.r_transaction()
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?;
let entities: Option<T> = r_transaction
.get()
.secondary(key_def, key)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?;
Ok(entities)
}
#[derive(Debug)]
pub struct Paginated<T> {
pub page: Option<u64>,
+240
View File
@@ -0,0 +1,240 @@
use arrow::array::{BooleanArray, Int32Array, Int64Array, ListBuilder, StringBuilder, UInt64Array};
use arrow::datatypes::{DataType, Field, Schema};
use arrow::record_batch::RecordBatch;
use std::sync::Arc;
use crate::modules::indexer::envelope::Envelope;
pub const DEFAULT_SHARD_ID: u64 = 0;
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("account_id", DataType::UInt64, false),
Field::new("mailbox_id", DataType::UInt64, false),
Field::new("uid", DataType::UInt64, false),
Field::new("subject", DataType::Utf8, true),
Field::new("body", DataType::Utf8, true),
Field::new("sender", DataType::Utf8, true),
Field::new(
"recipients",
DataType::List(Arc::new(Field::new("item", DataType::Utf8, true))),
true,
),
Field::new(
"cc",
DataType::List(Arc::new(Field::new("item", DataType::Utf8, true))),
true,
),
Field::new(
"bcc",
DataType::List(Arc::new(Field::new("item", DataType::Utf8, true))),
true,
),
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("message_id", DataType::Utf8, true),
Field::new("has_attachment", DataType::Boolean, false),
Field::new("attachment_count", DataType::Int32, false),
Field::new(
"tags",
DataType::List(Arc::new(Field::new("item", DataType::Utf8, true))),
true,
),
Field::new("shard_id", DataType::UInt64, false),
]));
let mut id_b = UInt64Array::builder(capacity);
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 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);
let mut to_b = ListBuilder::new(StringBuilder::new());
let mut cc_b = ListBuilder::new(StringBuilder::new());
let mut bcc_b = ListBuilder::new(StringBuilder::new());
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 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 tags_b = ListBuilder::new(StringBuilder::new());
let mut shard_id_b = UInt64Array::builder(capacity);
for e in items {
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);
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);
att_count_b.append_value(e.attachment_count as i32);
tags_b.append(true);
shard_id_b.append_value(DEFAULT_SHARD_ID);
}
RecordBatch::try_new(
schema,
vec![
Arc::new(id_b.finish()),
Arc::new(account_id_b.finish()),
Arc::new(mailbox_id_b.finish()),
Arc::new(uid_b.finish()),
Arc::new(subject_b.finish()),
Arc::new(body_b.finish()),
Arc::new(from_b.finish()),
Arc::new(to_b.finish()),
Arc::new(cc_b.finish()),
Arc::new(bcc_b.finish()),
Arc::new(date_b.finish()),
Arc::new(internal_date_b.finish()),
Arc::new(size_b.finish()),
Arc::new(thread_id_b.finish()),
Arc::new(msg_id_b.finish()),
Arc::new(has_att_b.finish()),
Arc::new(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 (
-- internal id (tantivy f_id)
id UBIGINT NOT NULL,
-- account / mailbox / uid
account_id UBIGINT NOT NULL,
mailbox_id UBIGINT NOT NULL,
uid UBIGINT NOT NULL,
-- headers / content
subject TEXT,
body TEXT,
sender TEXT,
recipients VARCHAR[],
cc VARCHAR[],
bcc VARCHAR[],
-- dates
sent_at BIGINT,
received_at BIGINT,
-- size
size_bytes UBIGINT,
-- thread
thread_id UBIGINT,
-- message-id
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
);
"#,
)?;
let items = vec![Envelope {
id: 101,
account_id: 1,
mailbox_id: 1,
uid: 50,
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,
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 FROM envelopes WHERE id = 101")?;
let mut rows = stmt.query([])?;
if let Some(row) = rows.next()? {
let subject: String = row.get(0)?;
let size: u64 = row.get(1)?;
assert_eq!(subject, "Testing Arrow");
assert_eq!(size, 2048);
}
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 = 101",
[],
|r| r.get(0),
)?;
assert!(has_att);
println!("Integration test for ingestion and query passed!");
Ok(())
}
}
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,70 @@
-- =========================
-- envelopes (main table)
-- =========================
CREATE TABLE IF NOT EXISTS envelopes (
-- internal id (tantivy f_id)
id UBIGINT NOT NULL,
-- account / mailbox / uid
account_id UBIGINT NOT NULL,
mailbox_id UBIGINT NOT NULL,
uid UBIGINT NOT NULL,
-- headers / content
subject TEXT,
body TEXT,
sender TEXT,
recipients VARCHAR[],
cc VARCHAR[],
bcc VARCHAR[],
-- dates
sent_at BIGINT,
received_at BIGINT,
-- size
size_bytes UBIGINT,
-- thread
thread_id UBIGINT,
-- message-id
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
);
CREATE INDEX idx_env_mailbox_sent ON envelopes(account_id, mailbox_id, sent_at);
-- =========================
-- envelope_attachments
--
-- Normalized attachment metadata for fast filtering and UI display
-- =========================
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,
-- Original attachment filename (for display)
filename TEXT NOT NULL,
-- Normalized file extension (lowercase, without dot)
extension TEXT NOT NULL,
-- Extension category (document / image / archive / ...)
ext_category TEXT NOT NULL,
content_type TEXT NOT NULL,
-- Attachment size in bytes
-- 0 if unknown
size_bytes UBIGINT NOT NULL,
shard_id UBIGINT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_attachments_mailbox_id ON envelope_attachments (mailbox_id);
+3
View File
@@ -0,0 +1,3 @@
pub mod build;
pub mod init;
pub mod refinery;
+74
View File
@@ -0,0 +1,74 @@
use chrono::{DateTime, Utc};
use refinery::{error::WrapMigrationError, Migration};
use refinery_core::{
traits::sync::{Query, Transaction},
Migrate,
};
use std::{ops::DerefMut, time::SystemTime};
pub struct DuckDBConnection<T: DerefMut<Target = duckdb::Connection>>(pub T);
impl<T: DerefMut<Target = duckdb::Connection>> From<T> for DuckDBConnection<T> {
fn from(conn: T) -> Self {
Self(conn)
}
}
impl<Conn: DerefMut<Target = duckdb::Connection>> Transaction for DuckDBConnection<Conn> {
type Error = duckdb::Error;
fn execute<'a, T: Iterator<Item = &'a str>>(
&mut self,
mut queries: T,
) -> std::result::Result<usize, Self::Error> {
let transaction = self.0.transaction()?;
let count = queries.try_fold(0, |count, query| {
transaction.execute_batch(query)?;
Ok::<_, Self::Error>(count + 1)
})?;
transaction.commit()?;
Ok(count)
}
}
impl<T: DerefMut<Target = duckdb::Connection>> Query<Vec<Migration>> for DuckDBConnection<T> {
fn query(&mut self, query: &str) -> Result<Vec<Migration>, Self::Error> {
let mut stmt = self.0.prepare(query)?;
let applied: Vec<Migration> = stmt
.query_map([], |row| {
let version = row.get(0)?;
let name: String = row.get(1)?;
let applied_on: DateTime<Utc> = row.get(2)?;
let checksum: u64 = row.get(3)?;
let applied_on: SystemTime = applied_on.into();
Ok(Migration::applied(
version,
name,
applied_on.into(),
checksum,
))
})?
.collect::<Result<Vec<_>, _>>()?;
Ok(applied)
}
}
impl<T: DerefMut<Target = duckdb::Connection>> Migrate for DuckDBConnection<T> {
fn assert_migrations_table(
&mut self,
migration_table_name: &str,
) -> std::result::Result<usize, refinery::Error> {
let query = format!(
"CREATE TABLE IF NOT EXISTS {migration_table_name} (
version INT primary key,
name TEXT not null,
applied_on TIMESTAMP not null,
checksum TEXT not null
)"
);
self.execute(std::iter::once(query.as_str()))
.migration_err("failed to create or verify DuckDB migrations table", None)?;
Ok(0)
}
}
+88 -135
View File
@@ -20,124 +20,55 @@ 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::utils::create_hash;
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::{id, modules::indexer::envelope::Envelope};
use async_imap::types::Fetch;
use mail_parser::{HeaderName, Message, MessageParser, MimeHeaders};
use mail_parser::{Address, HeaderName, Message, MessageParser, MimeHeaders};
pub fn extract_envelope(fetch: &Fetch, account_id: u64, mailbox_id: u64) -> BichonResult<Envelope> {
pub fn extract_envelope(
fetch: &Fetch,
account_id: u64,
mailbox_id: u64,
) -> BichonResult<(Envelope, Vec<AttachmentInfo>)> {
let internal_date = fetch
.internal_date()
.map(|d| d.timestamp_millis())
.unwrap_or(0);
let uid = fetch.uid.unwrap_or(0);
let body = fetch
.body()
.ok_or_else(|| raise_error!("No body available".into(), ErrorCode::InternalError))?;
let size = fetch.size.unwrap_or(body.len() as u32);
let message = MessageParser::new().parse(body).ok_or_else(|| {
raise_error!(
"Email header parse result is not available".into(),
ErrorCode::InternalError
)
})?;
let text = if let Some(text) = message.body_text(0).map(|cow| cow.into_owned()) {
text
} else if let Some(html) = message.body_html(0).map(|cow| cow.into_owned()) {
extract_text(html)
} else {
String::new()
};
let message_id = message
.message_id()
.map(String::from)
.unwrap_or(generate_message_id());
let in_reply_to = message.in_reply_to().as_text().map(String::from);
let references = extract_references(&message);
let thread_id = compute_thread_id(in_reply_to, references, &message_id);
let mut subject = message.subject().map(String::from).unwrap_or_default();
if subject.contains('\u{FFFD}') {
subject = normalize_subject(message.header_raw(HeaderName::Subject));
}
let date = message.date().map(|d| d.to_timestamp() * 1000).unwrap_or(0);
let bcc: Option<Vec<String>> = message.bcc().map(|addr| {
AddrVec::from(addr)
.0
.into_iter()
.filter_map(|a| a.address)
.collect()
});
let cc: Option<Vec<String>> = message.cc().map(|addr| {
AddrVec::from(addr)
.0
.into_iter()
.filter_map(|a| a.address)
.collect()
});
let to: Option<Vec<String>> = message.to().map(|addr| {
AddrVec::from(addr)
.0
.into_iter()
.filter_map(|a| a.address)
.collect()
});
let from = message
.from()
.and_then(|addr| AddrVec::from(addr).0.into_iter().next())
.and_then(|add| add.address)
.unwrap_or_else(|| "unknown".to_string());
let attachments: Vec<String> = message
.attachments()
.filter(|att| {
let disp = att.content_disposition();
let is_inline = disp.map(|d| d.is_inline()).unwrap_or(false);
let has_filename = att.attachment_name().is_some();
has_filename && !is_inline
})
.filter_map(|att| att.attachment_name())
.map(|name| name.to_string())
.collect();
let envelope = Envelope {
id: create_hash(account_id, &message_id),
message_id,
account_id,
mailbox_id,
uid,
subject,
text,
from,
to: to.unwrap_or_default(),
cc: cc.unwrap_or_default(),
bcc: bcc.unwrap_or_default(),
date,
internal_date,
size,
thread_id,
attachments,
tags: None,
account_email: None,
mailbox_name: None,
};
Ok(envelope)
extract_envelope_core(body, uid, size, internal_date, account_id, mailbox_id)
}
pub fn extract_envelope_from_eml(
body: &[u8],
account_id: u64,
mailbox_id: u64,
) -> BichonResult<Envelope> {
let uid = 0;
let size = body.len() as u32;
) -> BichonResult<(Envelope, Vec<AttachmentInfo>)> {
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)
},
)
}
fn extract_envelope_core(
body: &[u8],
uid: u32,
size: u32,
internal_date: i64,
account_id: u64,
mailbox_id: u64,
) -> BichonResult<(Envelope, Vec<AttachmentInfo>)> {
let message = MessageParser::new().parse(body).ok_or_else(|| {
raise_error!(
"Email header parse result is not available".into(),
@@ -156,7 +87,8 @@ pub fn extract_envelope_from_eml(
let message_id = message
.message_id()
.map(String::from)
.unwrap_or(generate_message_id());
.unwrap_or_else(generate_message_id);
let in_reply_to = message.in_reply_to().as_text().map(String::from);
let references = extract_references(&message);
let thread_id = compute_thread_id(in_reply_to, references, &message_id);
@@ -167,46 +99,66 @@ pub fn extract_envelope_from_eml(
}
let date = message.date().map(|d| d.to_timestamp() * 1000).unwrap_or(0);
let bcc: Option<Vec<String>> = message.bcc().map(|addr| {
AddrVec::from(addr)
.0
.into_iter()
.filter_map(|a| a.address)
.collect()
});
let cc: Option<Vec<String>> = message.cc().map(|addr| {
AddrVec::from(addr)
.0
.into_iter()
.filter_map(|a| a.address)
.collect()
});
let to: Option<Vec<String>> = message.to().map(|addr| {
AddrVec::from(addr)
.0
.into_iter()
.filter_map(|a| a.address)
.collect()
});
let parse_addrs = |addrs: Option<&Address<'_>>| {
addrs
.map(|addr| {
AddrVec::from(addr)
.0
.into_iter()
.filter_map(|a| a.address)
.collect()
})
.unwrap_or_default()
};
let bcc = parse_addrs(message.bcc());
let cc = parse_addrs(message.cc());
let to = parse_addrs(message.to());
let from = message
.from()
.and_then(|addr| AddrVec::from(addr).0.into_iter().next())
.and_then(|add| add.address)
.unwrap_or_else(|| "unknown".to_string());
let attachments: Vec<String> = message
let attachments: Vec<AttachmentInfo> = message
.attachments()
.filter(|att| {
let disp = att.content_disposition();
let is_inline = disp.map(|d| d.is_inline()).unwrap_or(false);
let has_filename = att.attachment_name().is_some();
has_filename && !is_inline
.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());
Some(AttachmentInfo {
filename: attachment
.attachment_name()
.map(|name| name.to_string())
.unwrap_or_default(),
size: attachment.contents().len(),
inline,
file_type,
content_id,
})
})
.filter_map(|att| att.attachment_name())
.map(|name| name.to_string())
.collect();
let envelope = Envelope {
id: create_hash(account_id, &message_id),
id: create_hash2(account_id, mailbox_id, &message_id),
message_id,
account_id,
mailbox_id,
@@ -214,19 +166,20 @@ pub fn extract_envelope_from_eml(
subject,
text,
from,
to: to.unwrap_or_default(),
cc: cc.unwrap_or_default(),
bcc: bcc.unwrap_or_default(),
to,
cc,
bcc,
date,
internal_date: date,
internal_date,
size,
thread_id,
attachments,
attachment_count: attachments.len(),
tags: None,
account_email: None,
mailbox_name: None,
};
Ok(envelope)
Ok((envelope, attachments))
}
pub fn compute_thread_id(
+8 -4
View File
@@ -25,6 +25,7 @@ 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;
@@ -207,13 +208,15 @@ impl ImapExecutor {
.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.id, envelope.to_document(mailbox_id)?)
.add_document(envelope.0.id, envelope)
.await;
let body = fetch.body().ok_or_else(|| {
raise_error!("missing a body".into(), ErrorCode::ImapUnexpectedResult)
})?;
EML_INDEX_MANAGER.add_document( envelope.id, doc!(fields.f_id => envelope.id, fields.f_account_id => account_id, fields.f_mailbox_id => mailbox_id, fields.f_eml => body)).await;
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;
count += 1;
}
Ok(count)
@@ -242,13 +245,14 @@ impl ImapExecutor {
.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.id, envelope.to_document(mailbox_id)?)
.add_document(envelope.0.id, envelope)
.await;
let body = fetch.body().ok_or_else(|| {
raise_error!("missing a body".into(), ErrorCode::ImapUnexpectedResult)
})?;
EML_INDEX_MANAGER.add_document( envelope.id, doc!(fields.f_id => envelope.id, fields.f_account_id => account_id, fields.f_mailbox_id => mailbox_id, fields.f_eml => body)).await;
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;
}
Ok(())
}
+1 -1
View File
@@ -94,7 +94,7 @@ impl ImapConnectionManager {
}
pub async fn build(account_id: u64) -> BichonResult<Session<Box<dyn SessionStream>>> {
let account = AccountModel::get(account_id).await?;
let account = AccountModel::async_get(account_id).await?;
let client = match Self::create_client(&account).await {
Ok(client) => client,
Err(error) => {
+24 -10
View File
@@ -16,13 +16,11 @@
// 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 mail_parser::{parsers::MessageStream, HeaderName, MessageParser};
use mail_parser::{parsers::MessageStream, MessageParser, MimeHeaders};
use crate::{
base64_encode_url_safe,
modules::{
account::entity::Encryption, envelope::utils::normalize_subject, imap::client::Client,
},
modules::{account::entity::Encryption, imap::client::Client},
};
#[tokio::test]
@@ -109,13 +107,29 @@ R0lGODlhAQABAIAAAAAAAP///yH5BAEAAAAALAAAAAABAAEAAAIBRAA7
#[tokio::test]
async fn test44() {
let path = r"C:\Users\polly\Downloads\test222.eml";
let path = r"C:\Users\polly\Downloads\test333.eml";
let input = std::fs::read(path).unwrap();
let message = MessageParser::default().parse(&input).unwrap();
let subject = message.subject().unwrap();
println!("Subject: {}", subject);
if subject.contains('\u{FFFD}') {
let subject = normalize_subject(message.header_raw(HeaderName::Subject));
println!("Subject: {}", subject);
for attachment in message.attachments() {
let content_type = attachment.content_type().unwrap();
let filename = attachment
.attachment_name()
.map(|name| name.to_string())
.unwrap_or_else(|| "unknown".to_string());
let disposition = attachment.content_disposition();
let file_type = format!(
"{}/{}",
content_type.c_type.as_ref(),
content_type.c_subtype.as_deref().unwrap_or("")
);
let inline = disposition.map(|d| d.is_inline()).unwrap_or(false);
println!(
"filename: {}, file_type: {}, inline: {}",
filename, file_type, inline
);
}
}
+4 -4
View File
@@ -129,16 +129,16 @@ impl ImportEmls {
continue;
}
};
let eml_id = create_hash(account_id, &envelope.0.message_id);
ENVELOPE_INDEX_MANAGER
.add_document(envelope.id, envelope.to_document(mailbox_id).unwrap())
.add_document(envelope.0.id, envelope)
.await;
EML_INDEX_MANAGER
.add_document(
envelope.id,
eml_id,
doc!(
fields.f_id => envelope.id,
fields.f_id => eml_id,
fields.f_account_id => account_id,
fields.f_mailbox_id => mailbox_id,
fields.f_eml => decoded
+60 -181
View File
@@ -16,18 +16,11 @@
// 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 std::collections::HashSet;
use crate::modules::account::migration::AccountModel;
use crate::modules::cache::imap::mailbox::MailBox;
use crate::modules::error::code::ErrorCode;
use crate::modules::utils::create_hash;
use crate::modules::{error::BichonResult, indexer::schema::SchemaTools};
use crate::raise_error;
use duckdb::types::Value;
use poem_openapi::Object;
use serde::{Deserialize, Serialize};
use tantivy::schema::Facet;
use tantivy::{doc, schema::Value, TantivyDocument};
use crate::modules::{account::migration::AccountModel, cache::imap::mailbox::MailBox};
#[derive(Clone, Debug, Default, Eq, PartialEq, Deserialize, Serialize, Object)]
pub struct Envelope {
@@ -48,186 +41,72 @@ pub struct Envelope {
pub internal_date: i64,
pub size: u32,
pub thread_id: u64,
pub attachments: Vec<String>,
pub attachment_count: usize,
pub tags: Option<Vec<String>>,
}
fn extract_u64_field(
document: &TantivyDocument,
field: tantivy::schema::Field,
) -> BichonResult<u64> {
let value = document.get_first(field).ok_or_else(|| {
raise_error!(
format!("miss '{}' field in tantivy document", stringify!(field)),
ErrorCode::InternalError
)
})?;
value.as_u64().ok_or_else(|| {
raise_error!(
format!("'{}' field is not a u64", stringify!(field)),
ErrorCode::InternalError
)
})
}
fn extract_i64_field(
document: &TantivyDocument,
field: tantivy::schema::Field,
) -> BichonResult<i64> {
let value = document.get_first(field).ok_or_else(|| {
raise_error!(
format!("miss '{}' field in tantivy document", stringify!(field)),
ErrorCode::InternalError
)
})?;
value.as_i64().ok_or_else(|| {
raise_error!(
format!("'{}' field is not a i64", stringify!(field)),
ErrorCode::InternalError
)
})
}
fn extract_string_field(
document: &TantivyDocument,
field: tantivy::schema::Field,
) -> BichonResult<String> {
let value = document.get_first(field).ok_or_else(|| {
raise_error!(
format!("'{}' field not found", stringify!(field)),
ErrorCode::InternalError
)
})?;
value.as_str().map(|s| s.to_string()).ok_or_else(|| {
raise_error!(
format!("'{}' field is not a string", stringify!(field)),
ErrorCode::InternalError
)
})
}
fn extract_vec_string_field(
document: &TantivyDocument,
field: tantivy::schema::Field,
) -> BichonResult<Vec<String>> {
let value = document
.get_all(field)
.filter_map(|v| v.as_str().map(|s| s.to_string()))
.collect();
Ok(value)
}
impl Envelope {
pub fn to_document(&self, mailbox_id: u64) -> BichonResult<TantivyDocument> {
let fields = SchemaTools::envelope_fields();
let mut doc = doc!();
doc.add_u64(fields.f_id, self.id);
doc.add_text(fields.f_message_id, &self.message_id);
doc.add_u64(fields.f_account_id, self.account_id);
doc.add_u64(fields.f_mailbox_id, mailbox_id);
doc.add_u64(fields.f_uid, self.uid as u64);
doc.add_text(fields.f_subject, &self.subject);
doc.add_text(fields.f_text, &self.text);
doc.add_text(fields.f_from, &self.from);
for to in &self.to {
doc.add_text(fields.f_to, to);
}
for cc in &self.cc {
doc.add_text(fields.f_cc, cc);
}
for bcc in &self.bcc {
doc.add_text(fields.f_bcc, bcc);
}
doc.add_i64(fields.f_date, self.date);
doc.add_i64(fields.f_internal_date, self.internal_date);
doc.add_u64(fields.f_size, self.size as u64);
doc.add_u64(fields.f_thread_id, self.thread_id);
for att in &self.attachments {
doc.add_text(fields.f_attachments, att);
}
doc.add_bool(fields.f_has_attachment, self.attachments.len() > 0);
Ok(doc)
}
pub async fn from_tantivy_doc(doc: &TantivyDocument) -> BichonResult<Self> {
let fields = SchemaTools::envelope_fields();
let account_id = extract_u64_field(doc, fields.f_account_id)?;
let message_id = extract_string_field(doc, fields.f_message_id)?;
let mailbox_id = extract_u64_field(doc, fields.f_mailbox_id)?;
let id = create_hash(account_id, &message_id);
let full_text = extract_string_field(doc, fields.f_text)?;
// Take up to the first 500 characters as a preview;
let preview = if full_text.chars().count() > 500 {
full_text.chars().take(500).collect::<String>() + "..."
} else {
full_text
pub fn from_row(row: &duckdb::Row) -> duckdb::Result<Self> {
let get_list = |col_name: &str| -> Vec<String> {
row.get::<_, Value>(col_name)
.map(|v| {
if let Value::List(inner_list) = v {
inner_list
.into_iter()
.filter_map(|item| {
if let Value::Text(s) = item {
Some(s)
} else {
None
}
})
.collect()
} else {
vec![]
}
})
.unwrap_or_default()
};
let account_id = row.get("account_id")?;
let mailbox_id = row.get("mailbox_id")?;
let email = match AccountModel::get(account_id) {
Ok(account) => account.email,
Err(_) => "unknown".to_string(),
};
let tags: Vec<String> = doc
.get_all(fields.f_tags)
.filter_map(|value| value.as_facet())
.map(|facet_encoded_str| {
Facet::from_encoded(facet_encoded_str.as_bytes().to_vec())
.ok()
.map(|facet| facet.to_string())
})
.flatten()
.collect();
let account_email = AccountModel::find(account_id).await?.map(|a| a.email);
let mailbox_name = MailBox::find_mailbox(account_id, mailbox_id)
.ok()
.and_then(|m| m)
.map(|m| m.name)
.unwrap_or_else(|| "unknown".to_string());
let mailboxes = MailBox::list_all(account_id).await?;
let mailbox_name = mailboxes
.iter()
.find(|m| m.id == mailbox_id)
.map(|m| m.name.clone());
let envelope = Envelope {
id,
Ok(Self {
id: row.get("id")?,
message_id: row.get("message_id").unwrap_or_default(),
account_id,
account_email,
account_email: Some(email),
mailbox_id,
mailbox_name,
message_id: extract_string_field(doc, fields.f_message_id)?,
uid: extract_u64_field(doc, fields.f_uid)? as u32,
subject: extract_string_field(doc, fields.f_subject)?,
text: preview,
from: extract_string_field(doc, fields.f_from)?,
to: extract_vec_string_field(doc, fields.f_to)?,
cc: extract_vec_string_field(doc, fields.f_cc)?,
bcc: extract_vec_string_field(doc, fields.f_bcc)?,
date: extract_i64_field(doc, fields.f_date)?,
internal_date: extract_i64_field(doc, fields.f_internal_date)?,
size: extract_u64_field(doc, fields.f_size)? as u32,
thread_id: extract_u64_field(doc, fields.f_thread_id)?,
attachments: extract_vec_string_field(doc, fields.f_attachments)?,
tags: Some(tags),
};
Ok(envelope)
}
}
pub async fn extract_contacts(doc: &TantivyDocument) -> BichonResult<HashSet<String>> {
let fields = SchemaTools::envelope_fields();
let mut all_contacts = HashSet::new();
if let Ok(from_val) = extract_string_field(doc, fields.f_from) {
if !from_val.is_empty() {
all_contacts.insert(from_val);
}
}
let multi_fields = [fields.f_to, fields.f_cc, fields.f_bcc];
for field in multi_fields {
if let Ok(vals) = extract_vec_string_field(doc, field) {
for v in vals {
if !v.is_empty() {
all_contacts.insert(v);
mailbox_name: Some(mailbox_name),
uid: row.get::<_, u64>("uid")? as u32,
subject: row.get("subject").unwrap_or_default(),
text: row.get("body").unwrap_or_default(),
from: row.get("sender").unwrap_or_default(),
to: get_list("recipients"),
cc: get_list("cc"),
bcc: get_list("bcc"),
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),
attachment_count: row.get::<_, i32>("attachment_count")? as usize,
tags: {
let t = get_list("tags");
if t.is_empty() {
None
} else {
Some(t)
}
}
}
},
})
}
Ok(all_contacts)
}
File diff suppressed because it is too large Load Diff
+84 -8
View File
@@ -16,11 +16,11 @@
// 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::base64_encode;
use crate::modules::account::migration::AccountModel;
use crate::modules::error::code::ErrorCode;
use crate::modules::indexer::manager::EML_INDEX_MANAGER;
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};
@@ -45,6 +45,72 @@ pub struct AttachmentInfo {
pub content_id: Option<String>,
}
impl AttachmentInfo {
pub fn get_extension(&self) -> String {
std::path::Path::new(&self.filename)
.extension()
.and_then(|ext| ext.to_str())
.map(|ext| ext.to_lowercase())
.unwrap_or_default()
}
pub fn get_category(&self) -> &'static str {
let 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")
}
_ => None,
};
if let Some(cat) = category {
return cat;
}
let mime = self.file_type.to_lowercase();
if mime.starts_with("image/") {
return "image";
}
if mime.starts_with("video/") {
return "video";
}
if mime.starts_with("audio/") {
return "audio";
}
if mime.starts_with("text/") {
return "text";
}
if mime.contains("compressed") || mime.contains("zip") || mime.contains("archive") {
return "archive";
}
if mime.contains("pdf") || mime.contains("msword") || mime.contains("officedocument") {
return "document";
}
"other"
}
}
/// 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
@@ -64,13 +130,23 @@ pub struct FullMessageContent {
pub attachments: Option<Vec<AttachmentInfo>>,
}
pub async fn retrieve_email_content(
account_id: u64,
id: u64,
) -> BichonResult<FullMessageContent> {
pub async fn retrieve_email_content(account_id: u64, id: u64) -> BichonResult<FullMessageContent> {
AccountModel::check_account_exists(account_id).await?;
let envelope = ENVELOPE_INDEX_MANAGER
.get_envelope_by_id(account_id, id)
.await?
.ok_or_else(|| {
raise_error!(
format!(
"Email record not found: account_id={} id={}",
account_id, id
),
ErrorCode::ResourceNotFound
)
})?;
let eml_id = create_hash(account_id, &envelope.message_id);
let eml = EML_INDEX_MANAGER
.get(account_id, id)
.get(account_id, eml_id)
.await?
.ok_or_else(|| {
raise_error!(
@@ -133,7 +209,7 @@ pub async fn retrieve_email_content(
attachments.push(AttachmentInfo {
filename,
size: attachment.len(),
size: attachment.contents().len(),
inline,
file_type,
content_id: attachment.content_id().map(Into::into),
+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::error::BichonResult;
use crate::modules::indexer::manager::{EML_INDEX_MANAGER, ENVELOPE_INDEX_MANAGER};
use std::collections::HashMap;
@@ -26,6 +25,6 @@ pub async fn delete_messages_impl(request: HashMap<u64, Vec<u64>>) -> BichonResu
.delete_email_multi_account(&request)
.await?;
ENVELOPE_INDEX_MANAGER
.delete_envelopes_multi_account(&request)
.delete_envelopes_multi_account(request)
.await
}
+3 -3
View File
@@ -39,14 +39,14 @@ pub struct SearchFilter {
pub bcc: Option<String>,
pub since: Option<i64>,
pub before: Option<i64>,
pub account_ids: Option<Vec<u64>>,
pub mailbox_ids: Option<Vec<u64>>,
pub account_ids: Option<HashSet<u64>>,
pub mailbox_ids: Option<HashSet<u64>>,
pub min_size: Option<u64>,
pub max_size: Option<u64>,
pub message_id: Option<String>,
pub has_attachment: Option<bool>,
pub attachment_name: Option<String>,
pub tags: Option<Vec<String>>,
pub tags: Option<HashSet<String>>,
}
#[derive(Debug, Clone, Default, Eq, PartialEq, Serialize, Deserialize, Enum)]
+1
View File
@@ -24,6 +24,7 @@ pub mod common;
pub mod context;
pub mod dashboard;
pub mod database;
pub mod duckdb;
pub mod envelope;
pub mod error;
pub mod imap;
+2 -2
View File
@@ -21,7 +21,7 @@ use crate::{
modules::{
database::{
delete_impl, insert_impl, manager::DB_MANAGER, paginate_query_primary_scan_all_impl,
secondary_find_impl, update_impl,
async_secondary_find_impl, update_impl,
},
error::{code::ErrorCode, BichonResult},
rest::response::DataPage,
@@ -133,7 +133,7 @@ impl OAuth2 {
}
pub async fn get(id: u64) -> BichonResult<Option<OAuth2>> {
secondary_find_impl(DB_MANAGER.meta_db(), OAuth2Key::id, id).await
async_secondary_find_impl(DB_MANAGER.meta_db(), OAuth2Key::id, id).await
}
pub async fn delete(id: u64) -> BichonResult<()> {
+1 -1
View File
@@ -58,7 +58,7 @@ impl AccountApi {
context
.require_permission(Some(account_id), Permission::ACCOUNT_READ_DETAILS)
.await?;
Ok(Json(AccountModel::get(account_id).await?))
Ok(Json(AccountModel::async_get(account_id).await?))
}
/// Delete an account by ID - WARNING: This permanently removes the account and all associated resources
+11 -11
View File
@@ -34,6 +34,7 @@ use crate::modules::rest::response::DataPage;
use crate::modules::rest::ApiResult;
use crate::modules::rest::ErrorCode;
use crate::modules::users::permissions::Permission;
use crate::modules::utils::validate_tag;
use crate::raise_error;
use poem::Body;
use poem_openapi::param::{Path, Query};
@@ -41,7 +42,6 @@ use poem_openapi::payload::{Attachment, AttachmentType, Json};
use poem_openapi::OpenApi;
use std::collections::HashMap;
use std::collections::HashSet;
use tantivy::schema::Facet;
pub struct MessageApi;
@@ -147,7 +147,7 @@ impl MessageApi {
/// Fetches the content of a specific email.
#[oai(
path = "/message-content/:account_id/:message_id",
path = "/message-content/:account_id/:envelope_id",
method = "get",
operation_id = "fetch_message_content"
)]
@@ -156,7 +156,7 @@ impl MessageApi {
/// The ID of the account.
account_id: Path<u64>,
/// The ID of the message to fetch.
message_id: Path<u64>,
envelope_id: Path<u64>,
context: ClientContext,
) -> ApiResult<Json<FullMessageContent>> {
let account_id = account_id.0;
@@ -164,13 +164,13 @@ impl MessageApi {
.require_permission(Some(account_id), Permission::DATA_READ)
.await?;
Ok(Json(
retrieve_email_content(account_id, message_id.0).await?,
retrieve_email_content(account_id, envelope_id.0).await?,
))
}
/// Retrieves the envelope (metadata) of a specific message.
#[oai(
path = "/envelope/:account_id/:message_id",
path = "/envelope/:account_id/:envelope_id",
method = "get",
operation_id = "get_envelope"
)]
@@ -179,7 +179,7 @@ impl MessageApi {
/// The ID of the account.
account_id: Path<u64>,
/// The ID of the message.
message_id: Path<u64>,
envelope_id: Path<u64>,
context: ClientContext,
) -> ApiResult<Json<Envelope>> {
let account_id = account_id.0;
@@ -187,13 +187,13 @@ impl MessageApi {
.require_permission(Some(account_id), Permission::DATA_READ)
.await?;
let envelope = ENVELOPE_INDEX_MANAGER
.get_envelope_by_id(account_id, message_id.0)
.get_envelope_by_id(account_id, envelope_id.0)
.await?
.ok_or_else(|| {
raise_error!(
format!(
"Envelope not found: account_id={} message_id={}",
account_id, message_id.0
"Envelope not found: account_id={} envelope_id={}",
account_id, envelope_id.0
),
ErrorCode::ResourceNotFound
)
@@ -308,8 +308,8 @@ impl MessageApi {
) -> ApiResult<()> {
let req = req.0;
for tag in &req.tags {
Facet::from_text(tag)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InvalidParameter))?;
validate_tag(tag)
.map_err(|e| raise_error!(format!("{}", e), ErrorCode::InvalidParameter))?;
}
for account_id in req.updates.keys() {
+15
View File
@@ -247,6 +247,21 @@ pub struct Settings {
value_parser = clap::value_parser!(u16).range(1..)
)]
pub bichon_sync_concurrency: Option<u16>,
#[clap(
long,
env,
help = "Number of DuckDB execution threads (default: auto)",
value_parser = clap::value_parser!(u16).range(1..)
)]
pub bichon_duckdb_threads: Option<i64>,
#[clap(
long,
env,
help = "Maximum memory DuckDB may use (e.g. 512MB, 2GB, 80%)"
)]
pub bichon_duckdb_max_memory: Option<String>,
}
impl Settings {
+3 -3
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, filter_by_secondary_key_impl, with_transaction,
async_find_impl, delete_impl, async_filter_by_secondary_key_impl, with_transaction,
};
use crate::modules::database::{insert_impl, list_all_impl, update_impl};
use crate::modules::settings::cli::SETTINGS;
@@ -154,7 +154,7 @@ impl AccessTokenModel {
}
pub async fn get_user_webui_token(user_id: u64) -> BichonResult<Option<AccessTokenModel>> {
let tokens = filter_by_secondary_key_impl::<AccessTokenModel>(
let tokens = async_filter_by_secondary_key_impl::<AccessTokenModel>(
DB_MANAGER.meta_db(),
AccessTokenModelKey::user_id,
user_id,
@@ -167,7 +167,7 @@ impl AccessTokenModel {
}
pub async fn get_user_api_tokens(user_id: u64) -> BichonResult<Vec<AccessTokenModel>> {
let tokens = filter_by_secondary_key_impl::<AccessTokenModel>(
let tokens = async_filter_by_secondary_key_impl::<AccessTokenModel>(
DB_MANAGER.meta_db(),
AccessTokenModelKey::user_id,
user_id,
+7 -7
View File
@@ -21,7 +21,7 @@ use crate::{
modules::{
database::{
async_find_impl, batch_delete_impl, delete_impl, list_all_impl, manager::DB_MANAGER,
secondary_find_impl, update_impl, with_transaction,
async_secondary_find_impl, update_impl, with_transaction,
},
error::{code::ErrorCode, BichonResult},
token::{AccessTokenModel, AccessTokenModelKey, TokenType},
@@ -282,7 +282,7 @@ impl BichonUserV2 {
username: String,
password: String,
) -> BichonResult<LoginResult> {
let user_option = secondary_find_impl::<UserModel>(
let user_option = async_secondary_find_impl::<UserModel>(
DB_MANAGER.meta_db(),
BichonUserV2Key::username,
username.clone(),
@@ -292,7 +292,7 @@ impl BichonUserV2 {
let user = match user_option {
Some(u) => u,
None => {
match secondary_find_impl::<UserModel>(
match async_secondary_find_impl::<UserModel>(
DB_MANAGER.meta_db(),
BichonUserV2Key::email,
username,
@@ -366,7 +366,7 @@ impl BichonUserV2 {
pub async fn check_username_conflict(username: &str) -> BichonResult<()> {
// Check username duplicate
if secondary_find_impl::<UserModel>(
if async_secondary_find_impl::<UserModel>(
DB_MANAGER.meta_db(),
BichonUserV2Key::username,
username.to_string(),
@@ -385,7 +385,7 @@ impl BichonUserV2 {
pub async fn check_email_conflict(email: &str) -> BichonResult<()> {
// Check email duplicate
if secondary_find_impl::<UserModel>(
if async_secondary_find_impl::<UserModel>(
DB_MANAGER.meta_db(),
BichonUserV2Key::email,
email.to_string(),
@@ -520,7 +520,7 @@ impl BichonUserV2 {
}
if let Some(username) = &request.username {
let user_option = secondary_find_impl::<UserModel>(
let user_option = async_secondary_find_impl::<UserModel>(
DB_MANAGER.meta_db(),
BichonUserV2Key::username,
username.to_string(),
@@ -538,7 +538,7 @@ impl BichonUserV2 {
}
if let Some(email) = &request.email {
let user_option = secondary_find_impl::<UserModel>(
let user_option = async_secondary_find_impl::<UserModel>(
DB_MANAGER.meta_db(),
BichonUserV2Key::email,
email.to_string(),
+2 -2
View File
@@ -284,7 +284,7 @@ impl UserCreateRequest {
}
for (aid, rid) in &self.account_access_map {
if AccountModel::find(*aid).await?.is_none() {
if AccountModel::async_find(*aid).await?.is_none() {
return Err(raise_error!(
format!("Account {} not found", aid),
ErrorCode::InvalidParameter
@@ -404,7 +404,7 @@ impl UserUpdateRequest {
if let Some(account_access_map) = &self.account_access_map {
for (aid, rid) in account_access_map {
if AccountModel::find(*aid).await?.is_none() {
if AccountModel::async_find(*aid).await?.is_none() {
return Err(raise_error!(
format!("Account {} not found", aid),
ErrorCode::InvalidParameter
+40 -6
View File
@@ -21,7 +21,7 @@ use std::{fs, io, path::PathBuf};
use crate::modules::error::BichonResult;
use base64::engine::general_purpose::STANDARD;
use base64::{engine::general_purpose, Engine};
use rand::{rng, Rng};
use rand::{rng, RngExt};
use super::error::code::ErrorCode;
@@ -277,14 +277,23 @@ pub fn hash(s: &str) -> u64 {
}
pub fn create_hash(account_id: u64, field: &str) -> u64 {
// Construct a buffer of bytes from account_id and mailbox_name
let mut buffer = Vec::new();
buffer.extend_from_slice(&account_id.to_le_bytes()); // Convert u64 to bytes
buffer.extend_from_slice(&account_id.to_le_bytes());
buffer.push(b':'); // Separator
buffer.extend_from_slice(field.as_bytes()); // Add mailbox name
// Create a Cursor for the buffer
buffer.extend_from_slice(field.as_bytes());
let mut cursor = std::io::Cursor::new(buffer);
let hash = murmur3::murmur3_x64_128(&mut cursor, 0).unwrap();
(hash & 0x1F_FFFF_FFFF_FFFF) as u64
}
pub fn create_hash2(account_id: u64, field1: u64, field2: &str) -> u64 {
let mut buffer = Vec::new();
buffer.extend_from_slice(&account_id.to_le_bytes());
buffer.push(b':'); // Separator
buffer.extend_from_slice(&field1.to_le_bytes());
buffer.push(b':'); // Separator
buffer.extend_from_slice(field2.as_bytes());
let mut cursor = std::io::Cursor::new(buffer);
// Compute the 128-bit Murmur3 hash and cast to u64
let hash = murmur3::murmur3_x64_128(&mut cursor, 0).unwrap();
(hash & 0x1F_FFFF_FFFF_FFFF) as u64
}
@@ -335,3 +344,28 @@ pub fn decode_avatar_bytes(base64_str: &str) -> BichonResult<Vec<u8>> {
Ok(bytes)
}
pub fn validate_tag(tag: &str) -> Result<(), String> {
if tag.is_empty() {
return Err("Tag cannot be empty".to_string());
}
const INVALID: &[char] = &[
'\'', '"', '`', ';', ',', '(', ')', '[', ']', '{', '}', '<', '>',
];
let mut found = Vec::new();
for c in tag.chars() {
if INVALID.contains(&c) && !found.contains(&c) {
found.push(c);
}
}
if !found.is_empty() {
let chars: String = found.iter().collect();
return Err(format!("Tag contains invalid characters: {}", chars));
}
Ok(())
}
+2 -9
View File
@@ -38,13 +38,6 @@ pub(crate) async fn establish_tcp_connection_with_timeout(
) -> BichonResult<Pin<Box<TimeoutStream<TcpStream>>>> {
// Establish the TCP connection with a timeout
let tcp_stream = connect_with_optional_proxy(use_proxy, address).await?;
// // Disable Nagle's algorithm for more efficient network communication
// tcp_stream
// .set_nodelay(true)
// .map_err(|e| raise_error!(e.to_string(), ErrorCode::NetworkError))?;
// Wrap the TCP stream in a TimeoutStream for timeout management
let mut timeout_stream = TimeoutStream::new(tcp_stream);
// Set read and write timeouts
@@ -140,7 +133,7 @@ async fn connect_with_optional_proxy(
)
})?
.map(|s| s.into_inner())
.map_err(|e| raise_error!(e.to_string(), ErrorCode::NetworkError));
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::NetworkError));
}
// Fallback to direct TCP connection
timeout(TIMEOUT, TcpStream::connect(address))
@@ -160,5 +153,5 @@ async fn connect_with_optional_proxy(
ErrorCode::ConnectionTimeout
)
})?
.map_err(|e| raise_error!(e.to_string(), ErrorCode::NetworkError))
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::NetworkError))
}
+1 -1
View File
@@ -77,7 +77,7 @@ async fn establish_rustls_stream(
let tls_stream = tls_connector
.connect(server_name, stream)
.await
.map_err(|e| raise_error!(e.to_string(), ErrorCode::NetworkError))?;
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::NetworkError))?;
Ok(tls_stream)
}
+1 -1
View File
@@ -43,6 +43,6 @@ export interface EmailEnvelope {
internal_date: number;
size: number;
thread_id: number,
attachments: string[];
attachment_count: number;
tags: string[];
}
@@ -23,7 +23,7 @@ import {
CardHeader,
CardTitle,
} from '@/components/ui/card'
import { UserAuthForm } from './components/user-auth-form'
import { UserAuthForm } from './user-auth-form'
import { useTranslation } from 'react-i18next'
import { AuthLayout } from './auth-layout'
@@ -1,63 +0,0 @@
//
// Copyright (c) 2025 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 <http://www.gnu.org/licenses/>.
import { IconLogout } from '@tabler/icons-react'
import { ConfirmDialog } from '@/components/confirm-dialog'
import { useTranslation } from 'react-i18next'
interface Props {
open: boolean
onOpenChange: (open: boolean) => void
handleConfirm: () => void
}
export function LogoutConfirmDialog({ open, onOpenChange, handleConfirm }: Props) {
const { t } = useTranslation()
const handleLogout = () => {
handleConfirm();
onOpenChange(false);
};
return (
<ConfirmDialog
open={open}
onOpenChange={onOpenChange}
handleConfirm={handleLogout}
className="max-w-md"
title={
<span className='text-destructive'>
<IconLogout
className='mr-1 inline-block stroke-destructive'
size={18}
/>{' '}
{t('auth.logout')}
</span>
}
desc={
<p>
{t('auth.areYouSureYouWantToLogOut')}
<br />
{t('auth.youWillNeedToLogInAgain')}
</p>
}
confirmText={t('auth.logout')}
cancelBtnText={t('common.cancel')}
/>
)
}
+3
View File
@@ -29,6 +29,7 @@ import { get_dashboard_stats, TimeBucket } from '@/api/system/api';
import { Main } from '@/components/layout/main';
import { FixedHeader } from '@/components/layout/fixed-header';
import { useTranslation } from 'react-i18next';
import { getToken } from '@/stores/authStore';
interface DailyActivity {
date: string;
@@ -102,8 +103,10 @@ const EmptyTable = ({ title }: { title: string }) => (
);
export default function MailArchiveDashboard() {
const token = getToken();
const { data: stats, isLoading } = useQuery({
queryKey: ['dashboard-stats'],
enabled: !!token,
queryFn: get_dashboard_stats,
});
@@ -118,7 +118,7 @@ export function MailList({
)}
{items.map((item, index) => {
const hasAttachments = item.attachments && item.attachments.length > 0
const hasAttachments = item.attachment_count > 0
const isSelected = currentEnvelope?.id === item.id
const isChecked = hasSelected(item.id)
return (
@@ -171,7 +171,7 @@ export function MailList({
{hasAttachments && (
<div className="flex items-center gap-1">
<Paperclip className="h-3 w-3" />
<span>{item.attachments?.length}</span>
<span>{item.attachment_count}</span>
</div>
)}
<span className="hidden md:inline">{formatBytes(item.size)}</span>
+1 -1
View File
@@ -207,7 +207,7 @@ export function MailListTable({
{
id: "attachment_count",
header: () => <Paperclip size={16} />,
cell: ({ row }) => <span className='text-xs'>{(row.original.attachments ?? []).length}</span>,
cell: ({ row }) => <span className='text-xs'>{row.original.attachment_count}</span>,
meta: { className: 'text-left text-xs' },
minSize: 40,
maxSize: 40
+2 -2
View File
@@ -154,7 +154,7 @@ export function MailList({
)}
{items.map((item, index) => {
const hasAttachments = item.attachments && item.attachments.length > 0
const hasAttachments = item.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 && (
<div className="flex items-center gap-1">
<Paperclip className="h-3 w-3" />
<span>{item.attachments?.length}</span>
<span>{item.attachment_count}</span>
</div>
)}
+4
View File
@@ -21,10 +21,14 @@ import { useQuery } from '@tanstack/react-query'
import { AxiosError } from 'axios'
import { get_current_user, User } from '@/api/users/api'
import { useMemo } from 'react'
import { getToken } from '@/stores/authStore'
export function useCurrentUser() {
const token = getToken();
const query = useQuery<User | null, AxiosError>({
queryKey: ['current-user'],
enabled: !!token,
queryFn: get_current_user,
retry: false,
})
+4 -20
View File
@@ -88,28 +88,12 @@ export function validateTag(facetPath: string) {
};
}
if (!facetPath.startsWith('/')) {
const invalidChars = /['"`;,()[\]{}<>]/;
if (invalidChars.test(facetPath)) {
return {
valid: false,
error: "Tag path must start with '/'"
};
}
let escaped = false;
for (let i = 1; i < facetPath.length; i++) {
const char = facetPath[i];
if (escaped) {
escaped = false;
} else if (char === '\\') {
escaped = true;
}
}
if (escaped) {
return {
valid: false,
error: "Tag path has unmatched escape character at the end"
error: "Tag path contains invalid characters"
};
}
+1 -1
View File
@@ -18,7 +18,7 @@
import { createFileRoute } from '@tanstack/react-router'
import SignIn from '@/features/auth/sign-in'
import SignIn from '@/features/auth'
export const Route = createFileRoute('/(auth)/sign-in')({
component: SignIn,