Compare commits

...
17 Commits
Author SHA1 Message Date
rustmailer a4f8e674c3 Update Cargo.lock 2026-05-18 20:46:31 +08:00
rustmailer dde6b990da bump to v1.1.0 2026-05-18 20:46:19 +08:00
rustmailerandGitHub 6b1f843bd5 Merge pull request #237 from rustmailer/fix/cli-mbox-memory
fix: bichon-cli OOMs on import #233
2026-05-18 20:27:03 +08:00
rustmailer 66fd50bc23 fix: bichon-cli OOMs on import #233 2026-05-18 20:25:13 +08:00
rustmailerandGitHub 9daab241b0 Merge pull request #236 from rustmailer/fix/deduplication
feat: add async index deduplication task
2026-05-18 18:03:14 +08:00
rustmailer 85d5490834 feat: add async index deduplication task 2026-05-18 15:30:37 +08:00
rustmailer 6e984f376c Update Cargo.lock 2026-05-17 12:29:48 +08:00
rustmailer 7470125a23 bump to v1.0.2 2026-05-17 12:29:42 +08:00
rustmailer 469d254e2b fix: can't select folders & scroll issue in Choose Mailboxes #222 #217 2026-05-17 12:26:49 +08:00
rustmailer 7fde7ee19a update 2026-05-17 10:35:32 +08:00
rustmailer d543508a23 fix: Overviews are breaking out of their boxes on the dashboard (v1.0.0) #218 2026-05-17 10:35:25 +08:00
rustmailer f440069912 add debug info in bichon-cli #224 2026-05-17 10:35:07 +08:00
rustmailer 9f9fc71d16 bump to v1.0.1 2026-05-16 22:19:15 +08:00
rustmailer b2a75643da fix(bichon-admin): reduce memory usage during data migration 2026-05-16 22:17:03 +08:00
rustmailer 37a38a2910 Update README.md 2026-05-15 12:21:06 +08:00
rustmailer 8817ed96f6 Update README.md 2026-05-15 12:19:50 +08:00
rustmailer 1ee2eade3a fix: rename bichonctl to bichon-cli 2026-05-15 12:07:26 +08:00
29 changed files with 1917 additions and 317 deletions
Generated
+6 -5
View File
@@ -293,7 +293,7 @@ checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6"
[[package]] [[package]]
name = "bichon-admin" name = "bichon-admin"
version = "1.0.0" version = "1.1.0"
dependencies = [ dependencies = [
"bichon-core", "bichon-core",
"console", "console",
@@ -301,6 +301,7 @@ dependencies = [
"indicatif", "indicatif",
"itertools", "itertools",
"memdb", "memdb",
"mimalloc",
"native_db", "native_db",
"native_model", "native_model",
"serde", "serde",
@@ -311,7 +312,7 @@ dependencies = [
[[package]] [[package]]
name = "bichon-cli" name = "bichon-cli"
version = "1.0.0" version = "1.1.0"
dependencies = [ dependencies = [
"base64 0.22.1", "base64 0.22.1",
"bichon-core", "bichon-core",
@@ -337,7 +338,7 @@ dependencies = [
[[package]] [[package]]
name = "bichon-core" name = "bichon-core"
version = "1.0.0" version = "1.1.0"
dependencies = [ dependencies = [
"async-imap", "async-imap",
"base64 0.22.1", "base64 0.22.1",
@@ -395,7 +396,7 @@ dependencies = [
[[package]] [[package]]
name = "bichon-server" name = "bichon-server"
version = "1.0.0" version = "1.1.0"
dependencies = [ dependencies = [
"bichon-core", "bichon-core",
"bichon-smtp", "bichon-smtp",
@@ -420,7 +421,7 @@ dependencies = [
[[package]] [[package]]
name = "bichon-smtp" name = "bichon-smtp"
version = "1.0.0" version = "1.1.0"
dependencies = [ dependencies = [
"base64 0.22.1", "base64 0.22.1",
"bichon-core", "bichon-core",
+1 -1
View File
@@ -11,7 +11,7 @@ members = [
resolver = "2" resolver = "2"
[workspace.package] [workspace.package]
version = "1.0.0" version = "1.1.0"
edition = "2021" edition = "2021"
[workspace.dependencies] [workspace.dependencies]
+16 -10
View File
@@ -39,7 +39,13 @@
<p align="center">A self-hosted email archiving server built in Rust. Download emails from IMAP accounts, builds a full-text search index, and serves a REST API with an embedded WebUI. Purpose-built for long-term preservation, unified cross-account search, and programmatic access to archived email.</p> <p align="center">A self-hosted email archiving server built in Rust. Download emails from IMAP accounts, builds a full-text search index, and serves a REST API with an embedded WebUI. Purpose-built for long-term preservation, unified cross-account search, and programmatic access to archived email.</p>
[![Watch the demo](https://img.youtube.com/vi/fMlayXo3Bo0/maxresdefault.jpg)](https://www.youtube.com/watch?v=fMlayXo3Bo0) <p align="center">
<a href="https://www.youtube.com/watch?v=fMlayXo3Bo0">
<img src="https://img.youtube.com/vi/fMlayXo3Bo0/maxresdefault.jpg" alt="Watch the demo"/>
</a>
<br/>
<em>▶ Click to watch the demo</em>
</p>
> [!NOTE] > [!NOTE]
> Bichon is an **archiver**, not an email client. It does not send, compose, forward, or reply to emails. Its optional SMTP server is for **receiving** emails only. > Bichon is an **archiver**, not an email client. It does not send, compose, forward, or reply to emails. Its optional SMTP server is for **receiving** emails only.
@@ -94,7 +100,7 @@
- **Multi-User RBAC**: 5 built-in roles (Admin, Manager, Member, AccountManager, AccountViewer) plus custom roles with 22 granular permissions. - **Multi-User RBAC**: 5 built-in roles (Admin, Manager, Member, AccountManager, AccountViewer) plus custom roles with 22 granular permissions.
- **Account-Level Isolation**: Grant users access to specific accounts with scoped roles. Permissions enforced at the API layer. - **Account-Level Isolation**: Grant users access to specific accounts with scoped roles. Permissions enforced at the API layer.
- **CLI Import Tools**: Import from EML directories, MBOX files (including Gmail variants), Thunderbird profiles, and Outlook PST files. - **CLI Import Tools**: Import from EML directories, MBOX files (including Gmail variants), Thunderbird profiles, and Outlook PST files.
- **CLI Export**: Download account data as MBOX via `bichonctl`. - **CLI Export**: Download account data as MBOX via `bichon-cli`.
- **Bulk Restore**: Restore emails in bulk back to their original IMAP accounts. - **Bulk Restore**: Restore emails in bulk back to their original IMAP accounts.
- **Embedded SMTP Server**: Receive emails directly at the gateway level. STARTTLS or TLS encryption. AUTH PLAIN/LOGIN with API token authentication. - **Embedded SMTP Server**: Receive emails directly at the gateway level. STARTTLS or TLS encryption. AUTH PLAIN/LOGIN with API token authentication.
- **Admin Tooling**: Password reset for locked-out admins. Non-destructive v0.3.7 to v1.0 data migration. - **Admin Tooling**: Password reset for locked-out admins. Non-destructive v0.3.7 to v1.0 data migration.
@@ -340,10 +346,10 @@ On first start, Bichon creates a built-in admin user:
## CLI Tools ## CLI Tools
### bichonctl — Import & Export ### bichon-cli — Import & Export
```bash ```bash
./bichonctl --config config.toml ./bichon-cli --config config.toml
``` ```
Creates a `config.toml` on first run with your server URL and API token. Creates a `config.toml` on first run with your server URL and API token.
@@ -391,12 +397,12 @@ All `/api/v1/*` endpoints require `Authorization: Bearer <token>`.
| Format | Tool | Notes | | Format | Tool | Notes |
|--------|------|-------| |--------|------|-------|
| **EML Directory** | `bichonctl` | Recursive `.eml` scan; preserves folder hierarchy | | **EML Directory** | `bichon-cli` | Recursive `.eml` scan; preserves folder hierarchy |
| **MBOX** | `bichonctl` | Single-file streaming import; supports Gmail's MBOX variant | | **MBOX** | `bichon-cli` | Single-file streaming import; supports Gmail's MBOX variant |
| **Thunderbird** | `bichonctl` | Reads directly from local Thunderbird profile directory | | **Thunderbird** | `bichon-cli` | Reads directly from local Thunderbird profile directory |
| **PST** | `bichonctl` | Outlook Personal Storage (`.pst`) file parsing | | **PST** | `bichon-cli` | Outlook Personal Storage (`.pst`) file parsing |
| **API Import** | `POST /api/v1/import` | Base64-encoded EML payloads for programmatic use | | **API Import** | `POST /api/v1/import` | Base64-encoded EML payloads for programmatic use |
| **MBOX Export** | `bichonctl` | Download account data as `.mbox` file | | **MBOX Export** | `bichon-cli` | Download account data as `.mbox` file |
All imports flow through the Bichon REST API. The server parses MIME, extracts metadata, indexes content into Tantivy, deduplicates by BLAKE3 content hash, and stores raw blobs in Fjall. All imports flow through the Bichon REST API. The server parses MIME, extracts metadata, indexes content into Tantivy, deduplicates by BLAKE3 content hash, and stores raw blobs in Fjall.
@@ -410,7 +416,7 @@ bichon/
│ ├── memdb/ Embedded key-value database layer (WAL, transactions) │ ├── memdb/ Embedded key-value database layer (WAL, transactions)
│ ├── core/ Library — IMAP sync, search, storage, auth, models │ ├── core/ Library — IMAP sync, search, storage, auth, models
│ ├── server/ Binary — Poem web server + embedded WebUI (rust-embed) │ ├── server/ Binary — Poem web server + embedded WebUI (rust-embed)
│ ├── cli/ Binary — bichonctl import/export CLI │ ├── cli/ Binary — bichon-cli import/export CLI
│ └── admin/ Binary — bichon-admin password reset & migration │ └── admin/ Binary — bichon-admin password reset & migration
└── web/ React + TypeScript + Vite + ShadCN UI frontend └── web/ React + TypeScript + Vite + ShadCN UI frontend
``` ```
+2 -1
View File
@@ -17,4 +17,5 @@ serde_json.workspace = true
itertools.workspace = true itertools.workspace = true
snafu.workspace = true snafu.workspace = true
memdb.workspace = true memdb.workspace = true
mimalloc = "0.1.50"
+10 -2
View File
@@ -18,6 +18,7 @@
use console::style; use console::style;
use dialoguer::{theme::ColorfulTheme, Select}; use dialoguer::{theme::ColorfulTheme, Select};
use mimalloc::MiMalloc;
use crate::{migrate::handle_migration, reset::handle_reset_password}; use crate::{migrate::handle_migration, reset::handle_reset_password};
@@ -25,8 +26,15 @@ pub mod meta;
pub mod migrate; pub mod migrate;
pub mod reset; pub mod reset;
#[global_allocator]
static GLOBAL: MiMalloc = MiMalloc;
fn main() {
run_interactive();
}
#[tokio::main] #[tokio::main]
async fn main() { async fn run_interactive() {
let theme = ColorfulTheme::default(); let theme = ColorfulTheme::default();
println!( println!(
"\n{}\n", "\n{}\n",
@@ -35,7 +43,7 @@ async fn main() {
let main_options = vec![ let main_options = vec![
"Reset Admin Password", "Reset Admin Password",
"Migrate Legacy v0.3.7 Storage to v1.0.0", "Migrate Legacy v0.3.7 Storage to v1.0.x",
"Exit", "Exit",
]; ];
+175 -45
View File
@@ -1,7 +1,7 @@
use std::path::{Path, PathBuf}; use std::path::{Path, PathBuf};
use bichon_core::migrate::{ use bichon_core::migrate::{
do_migrate, is_tantivy_index_dir, count_eml_segments, do_migrate_segment, is_tantivy_index_dir,
store::{LegacyDirs, NewDirs}, store::{LegacyDirs, NewDirs},
}; };
use console::style; use console::style;
@@ -11,7 +11,7 @@ use indicatif::{ProgressBar, ProgressStyle};
pub fn handle_migration(theme: &ColorfulTheme) { pub fn handle_migration(theme: &ColorfulTheme) {
println!( println!(
"\n{}", "\n{}",
style("MIGRATION: Bichon v0.3.7 Storage Architecture → v1.0.0") style("MIGRATION: Bichon v0.3.7 Storage Architecture → v1.0.x")
.bold() .bold()
.yellow() .yellow()
); );
@@ -20,7 +20,7 @@ pub fn handle_migration(theme: &ColorfulTheme) {
"{}", "{}",
style( style(
"This tool migrates data from the legacy v0.3.7 Tantivy-based storage \ "This tool migrates data from the legacy v0.3.7 Tantivy-based storage \
architecture to the new v1.0.0 \ architecture to the new v1.0.x \
separated index and Fjall-backed storage format." separated index and Fjall-backed storage format."
) )
.dim() .dim()
@@ -32,7 +32,7 @@ pub fn handle_migration(theme: &ColorfulTheme) {
"Legacy v0.3.7 architecture:\n\ "Legacy v0.3.7 architecture:\n\
• envelope metadata stored in Tantivy\n\ • envelope metadata stored in Tantivy\n\
• message data stored in Tantivy\n\n\ • message data stored in Tantivy\n\n\
New v1.0.0 architecture:\n\ New v1.0.x architecture:\n\
• mail indexes stored in Tantivy\n\ • mail indexes stored in Tantivy\n\
• attachment indexes stored in Tantivy\n\ • attachment indexes stored in Tantivy\n\
• raw message data stored in Fjall\n\ • raw message data stored in Fjall\n\
@@ -172,7 +172,7 @@ pub fn handle_migration(theme: &ColorfulTheme) {
println!( println!(
"{} {}", "{} {}",
style("").green(), style("").green(),
style("Legacy v0.x Tantivy-based storage detected. Migration to v1.0 is required.") style("Legacy v0.3.7 Tantivy-based storage detected. Migration to v1.0 is required.")
.yellow() .yellow()
); );
} }
@@ -180,7 +180,7 @@ pub fn handle_migration(theme: &ColorfulTheme) {
println!( println!(
"{} {}", "{} {}",
style("").green(), style("").green(),
style("No legacy v0.x storage layout was detected at the specified paths.").green() style("No legacy v0.3.7 storage layout was detected at the specified paths.").green()
); );
println!( println!(
@@ -234,8 +234,7 @@ pub fn handle_migration(theme: &ColorfulTheme) {
); );
eprintln!( eprintln!(
"{}", "{}",
style("Aborting migration. No changes have been made to Tantivy data.") style("Aborting migration. No changes have been made to Tantivy data.").yellow()
.yellow()
); );
return; return;
} }
@@ -246,48 +245,179 @@ pub fn handle_migration(theme: &ColorfulTheme) {
style("").yellow(), style("").yellow(),
style("Step 2: Migrating email index and blob data...").cyan() style("Step 2: Migrating email index and blob data...").cyan()
); );
let pb = ProgressBar::new(0);
pb.set_style(ProgressStyle::default_bar()
.template("{spinner:.green} [{elapsed_precise}] [{bar:40.cyan/blue}] {pos}/{len} ({eta}) {msg}")
.unwrap()
.progress_chars("#>-"));
let legacy = LegacyDirs::new(index_path, data_path);
let new_dirs = NewDirs::new(new_index_path, new_data_path);
if let Err(e) = do_migrate(legacy, new_dirs, |msg| {
if let Some(data) = msg.strip_prefix("PROGRESS:") {
let parts: Vec<&str> = data.split(':').collect();
if parts.len() == 2 {
let migrated = parts[0].parse::<u64>().unwrap_or(0);
let skipped = parts[1].parse::<u64>().unwrap_or(0);
pb.set_position(migrated + skipped); println!(
pb.set_message(format!( "\n{} {}",
"Migrated: {}, {} {}", style("").blue(),
style(migrated).green(), style("Batch size controls memory usage during migration:").dim()
style(skipped).red(), );
style("skipped").dim() println!(
)); " {} 1000 — ~500MB RAM (slower, low memory)",
} style("").dim()
} else if let Some(total) = msg.strip_prefix("TOTAL:") { );
pb.set_length(total.parse().unwrap_or(0)); println!(" {} 3000 — ~1GB RAM (recommended)", style("").dim());
} else if msg.starts_with("WARN:") { println!(
pb.println(format!("{} {}", style("").yellow(), &msg[5..])); " {} 5000 — ~2GB RAM (faster, high memory)",
} else if let Some(done_data) = msg.strip_prefix("DONE:") { style("").dim()
let parts: Vec<&str> = done_data.split(':').collect(); );
pb.finish_with_message(format!( println!(
"Migration finished. Total: {}, Skipped: {}", " {} Note: actual memory usage depends on your average email size.",
parts.get(0).unwrap_or(&"0"), style("").yellow()
parts.get(1).unwrap_or(&"0") );
)); println!(
" {} If your mailbox contains many large attachments, use a smaller batch size.\n",
style(" ").dim()
);
let batch_size: u32 = {
let input: String = Input::with_theme(&ColorfulTheme::default())
.with_prompt("Enter batch size (affects memory usage, see notes above)")
.default("3000".to_string())
.validate_with(|s: &String| match s.trim().parse::<usize>() {
Ok(n) if n > 0 => Ok(()),
_ => Err("Please enter a valid positive number"),
})
.interact_text()
.unwrap_or("3000".to_string());
input.trim().parse::<u32>().unwrap_or(3000)
};
println!(
"{} Using batch size: {}\n",
style("").green(),
style(batch_size).cyan().bold()
);
println!(
"{} Using batch size: {}\n",
style("").green(),
style(batch_size).cyan().bold()
);
let legacy = LegacyDirs::new(index_path.clone(), data_path.clone());
let total_segments = match count_eml_segments(&legacy) {
Ok(n) => n,
Err(e) => {
eprintln!(
"\n{} Failed to count EML segments:\n{:?}",
style("").red().bold(),
e
);
return;
} }
}) { };
eprintln!(
"\n{} Migration failed:\n{:?}", if total_segments == 0 {
style("").red().bold(), println!(
style(e).red() "{} {}",
style("").green(),
style("No EML segments found. Nothing to migrate.").bold()
); );
return; return;
} }
println!(
"{} EML segments to migrate: {}",
style("").yellow(),
style(total_segments).cyan()
);
let pb = ProgressBar::new(total_segments as u64);
pb.set_style(
ProgressStyle::default_bar()
.template(
"{spinner:.green} [{elapsed_precise}] [{bar:40.cyan/blue}] {pos}/{len} ({eta}) {msg}",
)
.unwrap()
.progress_chars("#>-"),
);
let mut grand_total_migrated: usize = 0;
let mut grand_total_skipped: usize = 0;
for seg_idx in 0..total_segments {
let seg_total: std::cell::Cell<usize> = std::cell::Cell::new(0);
pb.set_message(format!("Segment {}/{}", seg_idx + 1, total_segments));
let legacy = LegacyDirs::new(index_path.clone(), data_path.clone());
match do_migrate_segment(
batch_size,
legacy,
NewDirs::new(new_index_path.clone(), new_data_path.clone()),
seg_idx,
|msg| {
if let Some(data) = msg.strip_prefix("TOTAL:") {
seg_total.set(data.parse().unwrap_or(0));
} else if let Some(data) = msg.strip_prefix("PHASE1:") {
let parts: Vec<&str> = data.split('/').collect();
let scanned: usize = parts.get(0).and_then(|s| s.parse().ok()).unwrap_or(0);
let total: usize = parts
.get(1)
.and_then(|s| s.split_once(" skipped:").map(|(n, _)| n))
.and_then(|s| s.parse().ok())
.unwrap_or(0);
let skipped: usize = data
.split_once("skipped:")
.and_then(|(_, s)| s.parse().ok())
.unwrap_or(0);
let pct = if total > 0 {
(scanned * 100) / total
} else {
0
};
pb.set_message(format!(
"Segment {}/{} [scanning {}/{} skipped:{} {}%]",
seg_idx + 1,
total_segments,
scanned,
total,
skipped,
pct,
));
} else if let Some(data) = msg.strip_prefix("PROGRESS:") {
let parts: Vec<&str> = data.split(':').collect();
let migrated: usize = parts.get(0).and_then(|s| s.parse().ok()).unwrap_or(0);
let total = seg_total.get();
let pct = if total > 0 {
(migrated * 100) / total
} else {
0
};
pb.set_message(format!(
"Segment {}/{} [migrating {}/{} {}%]",
seg_idx + 1,
total_segments,
migrated,
total,
pct,
));
} else if let Some(warn) = msg.strip_prefix("WARN:") {
pb.println(format!("{} {}", style("").yellow(), warn));
} else if let Some(done_data) = msg.strip_prefix("DONE:") {
let parts: Vec<&str> = done_data.split(':').collect();
let migrated: usize = parts.get(0).and_then(|s| s.parse().ok()).unwrap_or(0);
let skipped: usize = parts.get(1).and_then(|s| s.parse().ok()).unwrap_or(0);
grand_total_migrated += migrated;
grand_total_skipped += skipped;
}
},
) {
Ok(()) => {}
Err(e) => {
pb.finish_with_message(format!("{}", style("Migration failed.").red()));
eprintln!("\n{} {:?}", style("").red().bold(), e);
return;
}
}
pb.set_position((seg_idx + 1) as u64);
}
pb.finish_with_message(format!(
"Migration finished. Total: {}, Skipped: {}",
grand_total_migrated, grand_total_skipped
));
println!( println!(
"{} {}", "{} {}",
style("").green(), style("").green(),
+2 -2
View File
@@ -1,4 +1,4 @@
use crate::BichonCtlConfig; use crate::BichonCliConfig;
use bichon_core::{base64_encode, envelope::meta::BichonMetadata, store::envelope::Envelope}; use bichon_core::{base64_encode, envelope::meta::BichonMetadata, store::envelope::Envelope};
use chrono::{TimeZone, Utc}; use chrono::{TimeZone, Utc};
use reqwest::Client; use reqwest::Client;
@@ -6,7 +6,7 @@ use tokio::io::AsyncWriteExt;
pub async fn download_and_export_with_json_header( pub async fn download_and_export_with_json_header(
client: &Client, client: &Client,
config: &BichonCtlConfig, config: &BichonCliConfig,
envelope: Envelope, envelope: Envelope,
file: &mut tokio::fs::File, file: &mut tokio::fs::File,
) -> bool { ) -> bool {
+2 -2
View File
@@ -5,11 +5,11 @@ use bichon_core::{
}; };
use reqwest::Client; use reqwest::Client;
use crate::BichonCtlConfig; use crate::BichonCliConfig;
pub async fn search_messages( pub async fn search_messages(
client: &Client, client: &Client,
config: &BichonCtlConfig, config: &BichonCliConfig,
page: u64, page: u64,
page_size: u64, page_size: u64,
) -> Option<DataPage<Envelope>> { ) -> Option<DataPage<Envelope>> {
+2 -2
View File
@@ -21,11 +21,11 @@ use reqwest::Client;
use bichon_core::import::BatchEmlRequest; use bichon_core::import::BatchEmlRequest;
use crate::BichonCtlConfig; use crate::BichonCliConfig;
pub async fn send_batch_request( pub async fn send_batch_request(
client: &Client, client: &Client,
config: &BichonCtlConfig, config: &BichonCliConfig,
account_id: u64, account_id: u64,
folder: &str, folder: &str,
emls: Vec<String>, emls: Vec<String>,
+2 -2
View File
@@ -1,11 +1,11 @@
use bichon_core::account::stats::AccountStats; use bichon_core::account::stats::AccountStats;
use reqwest::Client; use reqwest::Client;
use crate::BichonCtlConfig; use crate::BichonCliConfig;
pub async fn fetch_account_stats( pub async fn fetch_account_stats(
client: &Client, client: &Client,
config: &BichonCtlConfig, config: &BichonCliConfig,
account_id: u64, account_id: u64,
) -> Option<AccountStats> { ) -> Option<AccountStats> {
let url = format!("{}/api/v1/accounts/{}/stats", config.base_url, account_id); let url = format!("{}/api/v1/accounts/{}/stats", config.base_url, account_id);
+67 -42
View File
@@ -27,19 +27,17 @@ use bichon_core::{
users::{permissions::Permission, view::UserView}, users::{permissions::Permission, view::UserView},
}; };
use crate::BichonCtlConfig; use crate::BichonCliConfig;
pub async fn verify_user_and_get_account(
config: &BichonCtlConfig,
theme: &ColorfulTheme,
only_nosync: bool,
) -> MinimalAccount {
let client = Client::new();
let url = format!("{}/api/v1/current-user", config.base_url);
async fn fetch_json<T: serde::de::DeserializeOwned>(
client: &Client,
url: &str,
token: &str,
label: &str,
) -> T {
let response = match client let response = match client
.get(&url) .get(url)
.header("Authorization", format!("Bearer {}", config.api_token)) .header("Authorization", format!("Bearer {}", token))
.send() .send()
.await .await
{ {
@@ -59,19 +57,15 @@ pub async fn verify_user_and_get_account(
} }
}; };
if !response.status().is_success() { let status = response.status();
let status = response.status(); let body = response.text().await.unwrap_or_else(|_| String::new());
let error_body = response
.text()
.await
.unwrap_or_else(|_| "No error detail provided".to_string());
if !status.is_success() {
eprintln!( eprintln!(
"\n{} Server returned an error (Status: {})", "\n{} Server returned an error (Status: {})",
style("✘ API Error:").red().bold(), style("✘ API Error:").red().bold(),
style(status).yellow() style(status).yellow()
); );
if status == 401 { if status == 401 {
eprintln!( eprintln!(
"{} Your API Token seems to be invalid or expired.", "{} Your API Token seems to be invalid or expired.",
@@ -83,36 +77,66 @@ pub async fn verify_user_and_get_account(
style("Context:").dim() style("Context:").dim()
); );
} }
eprintln!("{} {}", style("Response:").dim(), body);
eprintln!("{} {}", style("Response:").dim(), error_body);
process::exit(1); process::exit(1);
} }
let user: UserView = response.json().await.expect("Failed to parse user data"); if body.is_empty() {
println!("Welcome, {}!", style(&user.username).cyan()); eprintln!(
"\n{} Server returned an empty response for [{}] (Status: {})",
let account_list_url = format!( style("✘ Empty Response:").red().bold(),
"{}/api/v1/minimal-account-list?only_nosync={only_nosync}", label,
config.base_url status
);
let acc_response = client
.get(&account_list_url)
.header("Authorization", format!("Bearer {}", config.api_token))
.send()
.await
.expect("Failed to fetch account list");
if !acc_response.status().is_success() {
panic!(
"Failed to retrieve accounts. Status: {}",
acc_response.status()
); );
eprintln!(
"{} This may be caused by a reverse proxy or middleware issue.",
style("Tip:").cyan()
);
process::exit(1);
} }
let accounts: Vec<MinimalAccount> = acc_response match serde_json::from_str::<T>(&body) {
.json() Ok(data) => data,
.await Err(e) => {
.expect("Failed to parse minimal account list"); eprintln!(
"\n{} Failed to parse response for [{}]: {}",
style("✘ Parse Error:").red().bold(),
label,
e
);
eprintln!("{} Raw body: {}", style("Debug:").dim(), body);
process::exit(1);
}
}
}
pub async fn verify_user_and_get_account(
config: &BichonCliConfig,
theme: &ColorfulTheme,
only_nosync: bool,
) -> MinimalAccount {
let client = Client::new();
let user: UserView = fetch_json(
&client,
&format!("{}/api/v1/current-user", config.base_url),
&config.api_token,
"current-user",
)
.await;
println!("Welcome, {}!", style(&user.username).cyan());
let accounts: Vec<MinimalAccount> = fetch_json(
&client,
&format!(
"{}/api/v1/minimal-account-list?only_nosync={only_nosync}",
config.base_url
),
&config.api_token,
"minimal-account-list",
)
.await;
if accounts.is_empty() { if accounts.is_empty() {
println!( println!(
@@ -129,6 +153,7 @@ pub async fn verify_user_and_get_account(
); );
process::exit(1); process::exit(1);
} }
let required_permission = Permission::DATA_IMPORT_BATCH; let required_permission = Permission::DATA_IMPORT_BATCH;
let mut selectable_accounts = Vec::new(); let mut selectable_accounts = Vec::new();
let mut options = Vec::new(); let mut options = Vec::new();
+3 -3
View File
@@ -29,10 +29,10 @@ use reqwest::Client;
use bichon_core::base64_encode_url_safe; use bichon_core::base64_encode_url_safe;
use crate::{BichonCtlConfig, api::sender::send_batch_request}; use crate::{BichonCliConfig, api::sender::send_batch_request};
pub async fn handle_eml_directory_import( pub async fn handle_eml_directory_import(
config: &BichonCtlConfig, config: &BichonCliConfig,
account_id: u64, account_id: u64,
theme: &ColorfulTheme, theme: &ColorfulTheme,
) { ) {
@@ -100,7 +100,7 @@ fn scan_dir(
} }
async fn process_and_upload( async fn process_and_upload(
config: &BichonCtlConfig, config: &BichonCliConfig,
account_id: u64, account_id: u64,
tasks: HashMap<String, Vec<PathBuf>>, tasks: HashMap<String, Vec<PathBuf>>,
) { ) {
+2 -2
View File
@@ -1,7 +1,7 @@
use crate::api::download::download_and_export_with_json_header; use crate::api::download::download_and_export_with_json_header;
use crate::api::search::search_messages; use crate::api::search::search_messages;
use crate::api::stats::fetch_account_stats; use crate::api::stats::fetch_account_stats;
use crate::BichonCtlConfig; use crate::BichonCliConfig;
use bichon_core::account::payload::MinimalAccount; use bichon_core::account::payload::MinimalAccount;
use console::style; use console::style;
use dialoguer::Confirm; use dialoguer::Confirm;
@@ -12,7 +12,7 @@ use std::path::{Path, PathBuf};
use sysinfo::Disks; use sysinfo::Disks;
pub async fn handle_account_export( pub async fn handle_account_export(
config: &BichonCtlConfig, config: &BichonCliConfig,
account: MinimalAccount, account: MinimalAccount,
theme: &ColorfulTheme, theme: &ColorfulTheme,
) { ) {
+5 -5
View File
@@ -39,7 +39,7 @@ pub mod thunderbird;
#[derive(Parser, Debug)] #[derive(Parser, Debug)]
#[command( #[command(
name = "bichonctl", name = "bichon-cli",
author = "rustmailer", author = "rustmailer",
version = bichon_version!(), version = bichon_version!(),
about = "A CLI tool to import email data into Bichon service" about = "A CLI tool to import email data into Bichon service"
@@ -57,7 +57,7 @@ pub struct BichonCli {
} }
#[derive(Serialize, Deserialize, Debug, Clone)] #[derive(Serialize, Deserialize, Debug, Clone)]
pub struct BichonCtlConfig { pub struct BichonCliConfig {
pub base_url: String, pub base_url: String,
pub api_token: String, pub api_token: String,
} }
@@ -67,11 +67,11 @@ async fn main() {
let cli = BichonCli::parse(); let cli = BichonCli::parse();
let theme = ColorfulTheme::default(); let theme = ColorfulTheme::default();
let config_path = &cli.config; let config_path = &cli.config;
let mut current_config: Option<BichonCtlConfig> = None; let mut current_config: Option<BichonCliConfig> = None;
if config_path.exists() { if config_path.exists() {
if let Ok(content) = fs::read_to_string(config_path) { if let Ok(content) = fs::read_to_string(config_path) {
if let Ok(config) = toml::from_str::<BichonCtlConfig>(&content) { if let Ok(config) = toml::from_str::<BichonCliConfig>(&content) {
println!("{}", style("✔ Existing configuration found:").green()); println!("{}", style("✔ Existing configuration found:").green());
println!(" Base URL: {}", style(&config.base_url).yellow()); println!(" Base URL: {}", style(&config.base_url).yellow());
println!(" API Token: {}", style(&config.api_token).yellow()); println!(" API Token: {}", style(&config.api_token).yellow());
@@ -105,7 +105,7 @@ async fn main() {
.interact_text() .interact_text()
.unwrap(); .unwrap();
let conf = BichonCtlConfig { let conf = BichonCliConfig {
base_url: url, base_url: url,
api_token: token, api_token: token,
}; };
+224 -5
View File
@@ -22,7 +22,7 @@ use std::path::PathBuf;
use crate::api::sender::send_batch_request; use crate::api::sender::send_batch_request;
use crate::mbox::gmail::determine_folder; use crate::mbox::gmail::determine_folder;
use crate::mbox::reader::MboxFile; use crate::mbox::reader::MboxFile;
use crate::BichonCtlConfig; use crate::BichonCliConfig;
use bichon_core::base64_encode_url_safe; use bichon_core::base64_encode_url_safe;
use bichon_core::envelope::meta::{parse_bichon_metadata, BichonMetadata}; use bichon_core::envelope::meta::{parse_bichon_metadata, BichonMetadata};
use console::style; use console::style;
@@ -32,11 +32,16 @@ use mail_parser::parsers::MessageStream;
use mail_parser::MessageParser; use mail_parser::MessageParser;
use reqwest::Client; use reqwest::Client;
/// Skip emails larger than this with a warning (100 MB).
const MAX_EMAIL_BYTES: usize = 100 * 1024 * 1024;
/// Flush a folder buffer when accumulated base64 bytes exceed this (200 MB).
const MAX_BUFFER_BYTES: usize = 200 * 1024 * 1024;
pub mod gmail; pub mod gmail;
pub mod reader; pub mod reader;
pub async fn handle_mbox_single_file_import( pub async fn handle_mbox_single_file_import(
config: &BichonCtlConfig, config: &BichonCliConfig,
account_id: u64, account_id: u64,
theme: &ColorfulTheme, theme: &ColorfulTheme,
) { ) {
@@ -120,7 +125,7 @@ pub async fn handle_mbox_single_file_import(
pub async fn run_import( pub async fn run_import(
account_id: u64, account_id: u64,
mbox_path: &PathBuf, mbox_path: &PathBuf,
config: &BichonCtlConfig, config: &BichonCliConfig,
target_folder: Option<String>, target_folder: Option<String>,
) { ) {
let client = Client::new(); let client = Client::new();
@@ -133,13 +138,29 @@ pub async fn run_import(
}; };
let mut folder_buffers: HashMap<String, Vec<String>> = HashMap::new(); let mut folder_buffers: HashMap<String, Vec<String>> = HashMap::new();
let mut total_buffered_bytes: usize = 0;
let batch_limit = 50; let batch_limit = 50;
let mut skipped_count: u64 = 0;
println!("Starting import process..."); println!("Starting import process...");
for (index, e) in mbox.iter().enumerate() { for (index, e) in mbox.iter().enumerate() {
let msg_num = index + 1; let msg_num = index + 1;
let body = e.data; let body = e.data;
if body.len() > MAX_EMAIL_BYTES {
let size_mb = body.len() as f64 / 1024.0 / 1024.0;
eprintln!(
"{} {}: email #{} is {:.1} MB (limit 100 MB). Skipping...",
style("Warning").yellow().bold(),
style(format!("oversized")).dim(),
msg_num,
size_mb,
);
skipped_count += 1;
continue;
}
let message = match MessageParser::new().parse(body) { let message = match MessageParser::new().parse(body) {
Some(msg) => msg, Some(msg) => msg,
None => { None => {
@@ -149,6 +170,7 @@ pub async fn run_import(
style(format!("at message #{}", msg_num)).dim(), style(format!("at message #{}", msg_num)).dim(),
"Failed to parse email structure. Skipping..." "Failed to parse email structure. Skipping..."
); );
skipped_count += 1;
continue; continue;
} }
}; };
@@ -178,14 +200,22 @@ pub async fn run_import(
get_default_folder() get_default_folder()
}; };
// Drop message before base64-encoding to free MIME parse memory.
drop(message);
let b64_eml = base64_encode_url_safe!(&body); let b64_eml = base64_encode_url_safe!(&body);
let encoded_len = b64_eml.len();
let buffer = folder_buffers let buffer = folder_buffers
.entry(folder_name.clone()) .entry(folder_name.clone())
.or_insert_with(|| Vec::new()); .or_insert_with(Vec::new);
buffer.push(b64_eml); buffer.push(b64_eml);
total_buffered_bytes += encoded_len;
if buffer.len() >= batch_limit { if buffer.len() >= batch_limit || total_buffered_bytes >= MAX_BUFFER_BYTES {
let emls_to_send = folder_buffers.remove(&folder_name).unwrap(); let emls_to_send = folder_buffers.remove(&folder_name).unwrap();
let freed: usize = emls_to_send.iter().map(|s| s.len()).sum();
total_buffered_bytes = total_buffered_bytes.saturating_sub(freed);
send_batch_request(&client, config, account_id, &folder_name, emls_to_send).await; send_batch_request(&client, config, account_id, &folder_name, emls_to_send).await;
} }
} }
@@ -196,5 +226,194 @@ pub async fn run_import(
} }
} }
if skipped_count > 0 {
println!(
"{}",
style(format!(
"Skipped {} email(s) (oversized or unparseable).",
skipped_count
))
.yellow()
.bold()
);
}
println!("{}", style("Import completed successfully!").green().bold()); println!("{}", style("Import completed successfully!").green().bold());
} }
#[cfg(test)]
mod tests {
use super::*;
use std::collections::HashMap;
/// Fake sender: records every flushed batch as (folder_name, email_count, total_bytes).
struct FakeSender {
batches: Vec<(String, usize, usize)>,
}
impl FakeSender {
fn new() -> Self {
Self { batches: vec![] }
}
fn send(&mut self, folder: &str, emls: Vec<String>) {
let count = emls.len();
let bytes: usize = emls.iter().map(|s| s.len()).sum();
self.batches.push((folder.to_string(), count, bytes));
// emls is dropped here, simulating real send
}
}
fn fake_encode(size: usize) -> String {
// base64 expands ~1.33x, so the encoded string is roughly this long.
// We just need a predictable byte size, so use a repeated character.
"x".repeat(size)
}
#[test]
fn flush_on_global_byte_threshold() {
let mut buffers: HashMap<String, Vec<String>> = HashMap::new();
let mut total_bytes: usize = 0;
let batch_limit = 50;
let mut sender = FakeSender::new();
// Simulate 3 emails, each 80 MB encoded, spread across 3 folders.
// After each email, global total goes up by 80 MB.
// After the 3rd email: 240 MB > 200 MB → flush the folder that got the 3rd email.
let emails = vec![
("Inbox", 80_000_000),
("Sent", 80_000_000),
("Archive", 80_000_000),
];
for (folder, eml_size) in emails {
let encoded = fake_encode(eml_size);
let len = encoded.len();
let buffer = buffers.entry(folder.to_string()).or_insert_with(Vec::new);
buffer.push(encoded);
total_bytes += len;
if buffer.len() >= batch_limit || total_bytes >= MAX_BUFFER_BYTES {
let sent = buffers.remove(folder).unwrap();
let freed: usize = sent.iter().map(|s| s.len()).sum();
total_bytes = total_bytes.saturating_sub(freed);
sender.send(folder, sent);
}
}
// The 3rd email should trigger a global flush of "Archive".
assert_eq!(sender.batches.len(), 1);
assert_eq!(sender.batches[0].0, "Archive");
assert_eq!(sender.batches[0].1, 1);
// "Inbox" and "Sent" are still buffered (160 MB total).
assert_eq!(buffers.len(), 2);
assert!(buffers.contains_key("Inbox"));
assert!(buffers.contains_key("Sent"));
assert_eq!(total_bytes, 160_000_000);
}
#[test]
fn flush_on_count_threshold() {
let mut buffers: HashMap<String, Vec<String>> = HashMap::new();
let mut total_bytes: usize = 0;
let batch_limit = 3;
let mut sender = FakeSender::new();
// 4 small emails all to Inbox, well under byte threshold.
for _ in 0..4 {
let encoded = fake_encode(100); // tiny
let len = encoded.len();
let buffer = buffers
.entry("Inbox".to_string())
.or_insert_with(Vec::new);
buffer.push(encoded);
total_bytes += len;
if buffer.len() >= batch_limit || total_bytes >= MAX_BUFFER_BYTES {
let sent = buffers.remove("Inbox").unwrap();
let freed: usize = sent.iter().map(|s| s.len()).sum();
total_bytes = total_bytes.saturating_sub(freed);
sender.send("Inbox", sent);
}
}
// Count=3 should trigger flush once; the 4th email stays buffered.
assert_eq!(sender.batches.len(), 1);
assert_eq!(sender.batches[0].1, 3); // 3 emails flushed
let remaining = buffers.get("Inbox").unwrap();
assert_eq!(remaining.len(), 1); // 1 still buffered
}
#[test]
fn global_bytes_exact_boundary() {
let mut buffers: HashMap<String, Vec<String>> = HashMap::new();
let mut total_bytes: usize = 0;
let mut sender = FakeSender::new();
// Push one email that puts us right at 200 MB.
let encoded = fake_encode(MAX_BUFFER_BYTES);
let len = encoded.len();
buffers
.entry("Inbox".to_string())
.or_insert_with(Vec::new)
.push(encoded);
total_bytes += len;
if total_bytes >= MAX_BUFFER_BYTES {
let sent = buffers.remove("Inbox").unwrap();
let freed: usize = sent.iter().map(|s| s.len()).sum();
total_bytes = total_bytes.saturating_sub(freed);
sender.send("Inbox", sent);
}
// Should have flushed on the boundary.
assert_eq!(sender.batches.len(), 1);
assert_eq!(total_bytes, 0);
}
#[test]
fn flush_one_folder_does_not_lose_others() {
let mut buffers: HashMap<String, Vec<String>> = HashMap::new();
let mut total_bytes: usize = 0;
let batch_limit = 50;
let mut sender = FakeSender::new();
// Build up A to 150 MB, B to 100 MB (total 250 MB > 200 MB).
// A should trigger flush; B should stay buffered.
let folder_a = "A".to_string();
let folder_b = "B".to_string();
// Folder A: 150 MB
let encoded = fake_encode(150_000_000);
let len = encoded.len();
buffers.entry(folder_a.clone()).or_insert_with(Vec::new).push(encoded);
total_bytes += len;
// Folder B: 100 MB → total 250 MB → trigger flush on B
let encoded = fake_encode(100_000_000);
let len = encoded.len();
buffers.entry(folder_b.clone()).or_insert_with(Vec::new).push(encoded);
total_bytes += len;
// Check trigger on B
let b_buffer = buffers.get(&folder_b).unwrap();
if b_buffer.len() >= batch_limit || total_bytes >= MAX_BUFFER_BYTES {
let sent = buffers.remove(&folder_b).unwrap();
let freed: usize = sent.iter().map(|s| s.len()).sum();
total_bytes = total_bytes.saturating_sub(freed);
sender.send(&folder_b, sent);
}
assert_eq!(sender.batches.len(), 1);
assert_eq!(sender.batches[0].0, "B"); // B flushed
assert!(buffers.contains_key("A")); // A still there
assert_eq!(total_bytes, 150_000_000);
}
#[test]
fn skip_oversized_email() {
assert!(100 <= MAX_EMAIL_BYTES);
// Use vec! so the 100 MB array lives on the heap, not the stack.
let huge = vec![0u8; MAX_EMAIL_BYTES + 1];
assert!(huge.len() > MAX_EMAIL_BYTES);
}
}
+5 -5
View File
@@ -25,7 +25,7 @@ use outlook_pst::ltp::prop_context::PropertyValue;
use crate::api::sender::send_batch_request; use crate::api::sender::send_batch_request;
use crate::pst::encoding::decode_subject; use crate::pst::encoding::decode_subject;
use crate::BichonCtlConfig; use crate::BichonCliConfig;
use bichon_core::base64_encode_url_safe; use bichon_core::base64_encode_url_safe;
use dialoguer::Confirm; use dialoguer::Confirm;
use outlook_pst::messaging::attachment::AttachmentProperties; use outlook_pst::messaging::attachment::AttachmentProperties;
@@ -60,7 +60,7 @@ pub struct EmailAttachment {
pub data: Option<Vec<u8>>, pub data: Option<Vec<u8>>,
} }
pub async fn handle_pst_import(config: &BichonCtlConfig, account_id: u64, theme: &ColorfulTheme) { pub async fn handle_pst_import(config: &BichonCliConfig, account_id: u64, theme: &ColorfulTheme) {
let path_str: String = Input::with_theme(theme) let path_str: String = Input::with_theme(theme)
.with_prompt("Enter the path to your SINGLE .pst file") .with_prompt("Enter the path to your SINGLE .pst file")
.validate_with(|input: &String| { .validate_with(|input: &String| {
@@ -115,7 +115,7 @@ pub async fn handle_pst_import(config: &BichonCtlConfig, account_id: u64, theme:
} }
} }
async fn parse_pst(pst_path: PathBuf, config: &BichonCtlConfig, account_id: u64) { async fn parse_pst(pst_path: PathBuf, config: &BichonCliConfig, account_id: u64) {
let client = Client::new(); let client = Client::new();
let pst_store = match outlook_pst::open_store(&pst_path) { let pst_store = match outlook_pst::open_store(&pst_path) {
@@ -161,7 +161,7 @@ fn process_folder_recursively<'a>(
client: &'a Client, client: &'a Client,
folder: &'a Rc<dyn Folder>, folder: &'a Rc<dyn Folder>,
parent_path: &'a str, parent_path: &'a str,
config: &'a BichonCtlConfig, config: &'a BichonCliConfig,
account_id: u64, account_id: u64,
) -> Pin<Box<dyn Future<Output = ()> + 'a>> { ) -> Pin<Box<dyn Future<Output = ()> + 'a>> {
Box::pin(async move { Box::pin(async move {
@@ -407,7 +407,7 @@ fn extract_recipients_list(message: &Rc<dyn Message>) -> (Vec<String>, Vec<Strin
async fn send_to_bichon( async fn send_to_bichon(
client: &Client, client: &Client,
config: &BichonCtlConfig, config: &BichonCliConfig,
account_id: u64, account_id: u64,
folder_path: &str, folder_path: &str,
emls: Vec<String>, emls: Vec<String>,
+2 -2
View File
@@ -18,12 +18,12 @@
use std::{collections::HashMap, path::PathBuf}; use std::{collections::HashMap, path::PathBuf};
use crate::{mbox::run_import, BichonCtlConfig}; use crate::{mbox::run_import, BichonCliConfig};
use console::style; use console::style;
use dialoguer::{theme::ColorfulTheme, Confirm, Input}; use dialoguer::{theme::ColorfulTheme, Confirm, Input};
pub async fn handle_thunderbird_import( pub async fn handle_thunderbird_import(
config: &BichonCtlConfig, config: &BichonCliConfig,
account_id: u64, account_id: u64,
theme: &ColorfulTheme, theme: &ColorfulTheme,
) { ) {
+25 -2
View File
@@ -32,6 +32,9 @@ use crate::{
raise_error, raise_error,
}; };
/// Skip individual emails larger than this after decoding (100 MB).
const MAX_SINGLE_EML_BYTES: usize = 100 * 1024 * 1024;
#[derive(Debug, Clone, Default, Eq, PartialEq, Serialize, Deserialize)] #[derive(Debug, Clone, Default, Eq, PartialEq, Serialize, Deserialize)]
#[cfg_attr(feature = "web-api", derive(poem_openapi::Object))] #[cfg_attr(feature = "web-api", derive(poem_openapi::Object))]
pub struct BatchEmlRequest { pub struct BatchEmlRequest {
@@ -66,7 +69,7 @@ pub struct BatchEmlResult {
pub struct ImportEmls; pub struct ImportEmls;
impl ImportEmls { impl ImportEmls {
pub async fn do_import(request: BatchEmlRequest) -> BichonResult<BatchEmlResult> { pub async fn do_import(mut request: BatchEmlRequest) -> BichonResult<BatchEmlResult> {
let account = AccountModel::check_account_exists(request.account_id)?; let account = AccountModel::check_account_exists(request.account_id)?;
if !account.enabled { if !account.enabled {
@@ -115,7 +118,8 @@ impl ImportEmls {
let mut failed_details: Vec<FailedEmlDetail> = Vec::new(); // Store failure details let mut failed_details: Vec<FailedEmlDetail> = Vec::new(); // Store failure details
let total = request.emls.len(); let total = request.emls.len();
for (index, eml_base64) in request.emls.into_iter().enumerate() { let mut index: usize = 0;
while let Some(eml_base64) = request.emls.pop() {
let decoded = match base64_decode_url_safe!(eml_base64.as_bytes()) { let decoded = match base64_decode_url_safe!(eml_base64.as_bytes()) {
Ok(bytes) => bytes, Ok(bytes) => bytes,
Err(e) => { Err(e) => {
@@ -126,9 +130,26 @@ impl ImportEmls {
index, index,
error_message: error_msg, error_message: error_msg,
}); });
index += 1;
continue; continue;
} }
}; };
// eml_base64 string dropped here — frees base64 memory before parsing
if decoded.len() > MAX_SINGLE_EML_BYTES {
let size_mb = decoded.len() as f64 / 1024.0 / 1024.0;
let error_msg = format!(
"Email at index {} is {:.1} MB (limit 50 MB). Skipping.",
index, size_mb,
);
tracing::warn!("{}", error_msg);
failed_details.push(FailedEmlDetail {
index,
error_message: error_msg,
});
index += 1;
continue;
}
match extract_envelope_from_eml(&decoded, account_id, mailbox_id).await { match extract_envelope_from_eml(&decoded, account_id, mailbox_id).await {
Ok(_) => { Ok(_) => {
@@ -144,9 +165,11 @@ impl ImportEmls {
index, index,
error_message: error_msg, error_message: error_msg,
}); });
index += 1;
continue; continue;
} }
}; };
index += 1;
} }
let failed_count = failed_details.len(); let failed_count = failed_details.len();
+164 -80
View File
@@ -1,4 +1,4 @@
use std::path::PathBuf; use std::{collections::HashMap, path::PathBuf};
use crate::{ use crate::{
error::{code::ErrorCode, BichonResult}, error::{code::ErrorCode, BichonResult},
@@ -10,7 +10,11 @@ use crate::{
settings::cli::SETTINGS, settings::cli::SETTINGS,
}; };
use tantivy::{ use tantivy::{
collector::TopDocs, query::AllQuery, schema::Value, DocAddress, Index, TantivyDocument, collector::TopDocs,
columnar::Column,
query::TermQuery,
schema::{IndexRecordOption, Value},
DocAddress, Index, TantivyDocument, Term,
}; };
pub mod legacy; pub mod legacy;
@@ -43,6 +47,18 @@ pub fn is_tantivy_index_dir(dir: &PathBuf) -> std::io::Result<bool> {
Ok(has_meta_json && match_count >= 3) Ok(has_meta_json && match_count >= 3)
} }
/// Return the number of segments in the legacy EML Tantivy index.
/// Each segment can be passed to `do_migrate_segment` for bounded-memory batch migration.
pub fn count_eml_segments(legacy: &LegacyDirs) -> BichonResult<usize> {
let eml_index = Index::open_in_dir(&legacy.eml_dir)
.map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?;
let reader = eml_index
.reader()
.map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?;
let searcher = reader.searcher();
Ok(searcher.segment_readers().len())
}
pub fn check_data_status() -> std::io::Result<bool> { pub fn check_data_status() -> std::io::Result<bool> {
let root_dir = PathBuf::from(&SETTINGS.bichon_root_dir); let root_dir = PathBuf::from(&SETTINGS.bichon_root_dir);
@@ -97,12 +113,22 @@ fn is_dir_not_empty(path: &PathBuf) -> std::io::Result<bool> {
Ok(entries.next().is_some()) Ok(entries.next().is_some())
} }
const PAGE_SIZE: usize = 100; /// Migrate all documents from a single EML segment to the new storage layout.
///
pub fn do_migrate<F>(legacy: LegacyDirs, new_dirs: NewDirs, mut on_progress: F) -> BichonResult<()> /// This is the core of the batch migration strategy: each Process B invocation
/// handles exactly one EML segment, so peak memory is bounded by that segment's
/// size regardless of the total archive size.
pub fn do_migrate_segment<F>(
batch_size: u32,
legacy: LegacyDirs,
new_dirs: NewDirs,
segment_index: usize,
mut on_progress: F,
) -> BichonResult<()>
where where
F: FnMut(&str), F: FnMut(&str),
{ {
// ── open legacy indices ────────────────────────────────────────────
let envelope_index = Index::open_in_dir(&legacy.envelope_dir) let envelope_index = Index::open_in_dir(&legacy.envelope_dir)
.map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?;
let eml_index = Index::open_in_dir(&legacy.eml_dir) let eml_index = Index::open_in_dir(&legacy.eml_dir)
@@ -118,114 +144,172 @@ where
let envelope_searcher = envelope_reader.searcher(); let envelope_searcher = envelope_reader.searcher();
let eml_searcher = eml_reader.searcher(); let eml_searcher = eml_reader.searcher();
let total_count = envelope_searcher.num_docs();
on_progress(&format!("TOTAL:{}", total_count));
let ef = SchemaTools::envelope_fields(); let ef = SchemaTools::envelope_fields();
let mf = SchemaTools::eml_fields(); let mf = SchemaTools::eml_fields();
let mut writer = NewIndexWriter::open(new_dirs)?; let eml_segments = eml_searcher.segment_readers();
let eml_segment = eml_segments.get(segment_index).ok_or_else(|| {
raise_error!(
format!(
"segment index {} out of range ({} segments)",
segment_index,
eml_segments.len()
),
ErrorCode::InternalError
)
})?;
let mut offset = 0usize; let num_docs = eml_segment.num_docs();
let mut total_migrated = 0usize; if num_docs == 0 {
let mut total_skipped = 0usize; on_progress("TOTAL:0");
on_progress("DONE:0:0");
return Ok(());
}
loop { on_progress(&format!("TOTAL:{}", num_docs));
let page: Vec<(_, DocAddress)> = envelope_searcher
.search( let max_doc = eml_segment.max_doc();
&AllQuery, let ff = eml_segment.fast_fields();
&TopDocs::with_limit(PAGE_SIZE) let f_id_col: Column<u64> = ff.u64("id").map_err(|e| {
.and_offset(offset) raise_error!(
.order_by_score(), format!("failed to open f_id fast field: {e:#?}"),
) ErrorCode::InternalError
)
})?;
// ── Phase 1: build eid → (uid, internal_date) from envelope, then drop it ──
let mut envelope_map: HashMap<u64, (u32, i64)> = HashMap::with_capacity(num_docs as usize);
let mut env_scanned = 0u32;
let mut env_skipped = 0u32;
for doc_id in 0..max_doc {
if eml_segment.is_deleted(doc_id) {
continue;
}
let eid = f_id_col.values.get_val(doc_id);
let term = Term::from_field_u64(ef.f_id, eid);
let query = TermQuery::new(term, IndexRecordOption::Basic);
let hits: Vec<(_, DocAddress)> = envelope_searcher
.search(&query, &TopDocs::with_limit(1).order_by_score())
.map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?;
if page.is_empty() { if let Some((_, addr)) = hits.first() {
break; let env_doc: TantivyDocument = envelope_searcher
} .doc(*addr)
let fetched = page.len();
for (_, doc_address) in page {
let doc: TantivyDocument = envelope_searcher
.doc(doc_address)
.map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?;
let uid = env_doc
let eid = match doc.get_first(ef.f_id).and_then(|v| v.as_u64()) {
Some(v) => v,
None => {
total_skipped += 1;
continue;
}
};
let account_id = match doc.get_first(ef.f_account_id).and_then(|v| v.as_u64()) {
Some(v) => v,
None => {
total_skipped += 1;
continue;
}
};
let mailbox_id = doc
.get_first(ef.f_mailbox_id)
.and_then(|v| v.as_u64())
.unwrap_or(0);
let uid = doc
.get_first(ef.f_uid) .get_first(ef.f_uid)
.and_then(|v| v.as_u64()) .and_then(|v| v.as_u64())
.unwrap_or(0) as u32; .unwrap_or(0) as u32;
let internal_date = doc let internal_date = env_doc
.get_first(ef.f_internal_date) .get_first(ef.f_internal_date)
.and_then(|v| v.as_i64()) .and_then(|v| v.as_i64())
.unwrap_or(0); .unwrap_or(0);
envelope_map.insert(eid, (uid, internal_date));
env_scanned += 1;
} else {
env_skipped += 1;
}
let eml_term = tantivy::Term::from_field_u64(mf.f_id, eid); if env_scanned % 10 == 0 {
let eml_query = on_progress(&format!(
tantivy::query::TermQuery::new(eml_term, tantivy::schema::IndexRecordOption::Basic); "PHASE1:{}/{} skipped:{}",
let eml_hits: Vec<(_, DocAddress)> = eml_searcher env_scanned, max_doc, env_skipped
.search(&eml_query, &TopDocs::with_limit(1).order_by_score()) ));
.map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; }
}
let eml_bytes = match eml_hits.first() { // Free the envelope index before the heavy EML processing.
Some((_, addr)) => { drop(envelope_searcher);
let eml_doc: TantivyDocument = eml_searcher drop(envelope_reader);
.doc(*addr) drop(envelope_index);
.map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?;
match eml_doc.get_first(mf.f_eml).and_then(|v| v.as_bytes()) { // ── Phase 2: process EML docs, streaming one at a time ─────────────
Some(b) => b.to_vec(), let mut writer = NewIndexWriter::open(new_dirs)?;
None => {
on_progress(&format!("WARN: Account {} ID {} eml field missing", account_id, eid)); let mut total_migrated = 0usize;
total_skipped += 1; let mut total_skipped = 0usize;
continue;
} // Recreate the StoreReader periodically to bound any internal caches.
} //const CHUNK_SIZE: u32 = 3000;
} let mut chunk_start = 0u32;
while chunk_start < max_doc {
let chunk_end = (chunk_start + batch_size).min(max_doc);
let store_reader = eml_segment
.get_store_reader(2)
.map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?;
for doc_id in chunk_start..chunk_end {
if eml_segment.is_deleted(doc_id) {
continue;
}
let eid = f_id_col.values.get_val(doc_id);
let (uid, internal_date) = match envelope_map.get(&eid) {
Some(v) => *v,
None => { None => {
on_progress(&format!("WARN:Account {} ID {} eml not found", account_id, eid)); on_progress(&format!("WARN: eid {} envelope not found", eid));
total_skipped += 1; total_skipped += 1;
continue; continue;
} }
}; };
if let Err(e) = writer.ingest(&eml_bytes, account_id, mailbox_id, uid, internal_date) { let eml_doc: TantivyDocument = store_reader
on_progress(&format!("ERROR:Account {} ID {} ingest failed: {}", account_id, eid, e)); .get(doc_id)
.map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?;
let account_id = match eml_doc.get_first(mf.f_account_id).and_then(|v| v.as_u64()) {
Some(v) => v,
None => {
on_progress(&format!("WARN: eid {} account_id missing", eid));
total_skipped += 1;
continue;
}
};
let mailbox_id = eml_doc
.get_first(mf.f_mailbox_id)
.and_then(|v| v.as_u64())
.unwrap_or(0);
// Borrow directly from eml_doc — no .to_vec() clone.
let eml_bytes = match eml_doc.get_first(mf.f_eml).and_then(|v| v.as_bytes()) {
Some(b) => b,
None => {
on_progress(&format!("WARN: eid {} eml bytes missing", eid));
total_skipped += 1;
continue;
}
};
if let Err(e) = writer.ingest(eml_bytes, account_id, mailbox_id, uid, internal_date) {
on_progress(&format!(
"ERROR: Account {} eid {} ingest failed: {}",
account_id, eid, e
));
total_skipped += 1; total_skipped += 1;
continue; continue;
} }
total_migrated += 1; total_migrated += 1;
if total_migrated % 100 == 0 || total_migrated == total_count as usize { if total_migrated % 10 == 0 || total_migrated as u32 == num_docs {
on_progress(&format!("PROGRESS:{}:{}", total_migrated, total_skipped)); on_progress(&format!("PROGRESS:{}:{}", total_migrated, num_docs));
} }
} }
offset += fetched; drop(store_reader);
if fetched < PAGE_SIZE {
break; // Flush Fjall buffers via ingestion API — bypasses memtable/WAL.
} writer.flush_fjall_buffers()?;
chunk_start = chunk_end;
} }
writer.commit()?; writer.finish_writers()?;
on_progress(&format!("DONE:{}:{}", total_migrated, total_skipped));
Ok(()) Ok(())
} }
+162 -39
View File
@@ -1,4 +1,4 @@
use std::path::PathBuf; use std::{path::PathBuf, time::Instant};
use bytes::Bytes; use bytes::Bytes;
use mail_parser::MimeHeaders; use mail_parser::MimeHeaders;
@@ -13,7 +13,7 @@ use fjall::{
CompressionType, Database, Keyspace, KeyspaceCreateOptions, KvSeparationOptions, CompressionType, Database, Keyspace, KeyspaceCreateOptions, KvSeparationOptions,
}; };
use mail_parser::MessageParser; use mail_parser::MessageParser;
use tantivy::{Index, IndexWriter, TantivyDocument}; use tantivy::{indexer::NoMergePolicy, Index, IndexWriter, TantivyDocument};
use uuid::Uuid; use uuid::Uuid;
use crate::{ use crate::{
@@ -121,14 +121,16 @@ pub fn detach_attachments_standalone(
} }
pub struct NewIndexWriter { pub struct NewIndexWriter {
pub envelope_writer: IndexWriter, pub envelope_writer: Option<IndexWriter>,
pub attachment_writer: IndexWriter, pub attachment_writer: Option<IndexWriter>,
pub email_ks: Keyspace, pub email_ks: Keyspace,
pub attachment_ks: Keyspace, pub attachment_ks: Keyspace,
pending: usize, pending: usize,
email_buf: Vec<(String, Vec<u8>)>,
attachment_buf: Vec<(String, Vec<u8>)>,
} }
const COMMIT_THRESHOLD: usize = 500; //const COMMIT_THRESHOLD: usize = 500;
impl NewIndexWriter { impl NewIndexWriter {
pub fn open(dirs: NewDirs) -> BichonResult<Self> { pub fn open(dirs: NewDirs) -> BichonResult<Self> {
@@ -153,9 +155,15 @@ impl NewIndexWriter {
.register("euro", EuroTokenizer::new()); .register("euro", EuroTokenizer::new());
let envelope_writer = envelope_index let envelope_writer = envelope_index
.writer_with_num_threads(2, 128 * 1024 * 1024) .writer_with_num_threads(3, 256 * 1024 * 1024)
.map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?;
// let mut merge_policy = LogMergePolicy::default();
// merge_policy.set_min_num_segments(25);
// merge_policy.set_min_layer_size(10_000);
// merge_policy.set_max_docs_before_merge(100_000);
envelope_writer.set_merge_policy(Box::new(NoMergePolicy));
// ── attachment index ───────────────────────────────────────────── // ── attachment index ─────────────────────────────────────────────
std::fs::create_dir_all(&dirs.attachment_dir) std::fs::create_dir_all(&dirs.attachment_dir)
.map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?;
@@ -176,21 +184,30 @@ impl NewIndexWriter {
.tokenizers() .tokenizers()
.register("euro", EuroTokenizer::new()); .register("euro", EuroTokenizer::new());
let attachment_writer = attachment_index let attachment_writer = attachment_index
.writer_with_num_threads(2, 64 * 1024 * 1024) .writer_with_num_threads(3, 256 * 1024 * 1024)
.map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?;
// let mut merge_policy = LogMergePolicy::default();
// merge_policy.set_min_num_segments(25);
// merge_policy.set_min_layer_size(10_000);
// merge_policy.set_max_docs_before_merge(100_000);
attachment_writer.set_merge_policy(Box::new(NoMergePolicy));
// ── blob store ─────────────────────────────────────────────────── // ── blob store ───────────────────────────────────────────────────
std::fs::create_dir_all(&dirs.storage_dir) std::fs::create_dir_all(&dirs.storage_dir)
.map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?;
let db = Database::builder(&dirs.storage_dir) let db = Database::builder(&dirs.storage_dir)
.cache_size(64 * 1024 * 1024) .cache_size(8 * 1024 * 1024)
.journal_compression(CompressionType::None)
.max_journaling_size(64 * 1024 * 1024)
.open() .open()
.map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?;
let email_ks = db let email_ks = db
.keyspace("email", || { .keyspace("email", || {
KeyspaceCreateOptions::default() KeyspaceCreateOptions::default()
.max_memtable_size(16 * 1024 * 1024) .max_memtable_size(4 * 1024 * 1024)
.data_block_size_policy(BlockSizePolicy::all(4 * 1024)) .data_block_size_policy(BlockSizePolicy::all(4 * 1024))
.data_block_compression_policy(CompressionPolicy::all(CompressionType::Lz4)) .data_block_compression_policy(CompressionPolicy::all(CompressionType::Lz4))
.with_kv_separation(Some( .with_kv_separation(Some(
@@ -205,7 +222,7 @@ impl NewIndexWriter {
let attachment_ks = db let attachment_ks = db
.keyspace("attachments", || { .keyspace("attachments", || {
KeyspaceCreateOptions::default() KeyspaceCreateOptions::default()
.max_memtable_size(16 * 1024 * 1024) .max_memtable_size(4 * 1024 * 1024)
.data_block_size_policy(BlockSizePolicy::all(4 * 1024)) .data_block_size_policy(BlockSizePolicy::all(4 * 1024))
.data_block_compression_policy(CompressionPolicy::all(CompressionType::Lz4)) .data_block_compression_policy(CompressionPolicy::all(CompressionType::Lz4))
.with_kv_separation(Some( .with_kv_separation(Some(
@@ -218,11 +235,13 @@ impl NewIndexWriter {
.map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?;
Ok(Self { Ok(Self {
envelope_writer, envelope_writer: Some(envelope_writer),
attachment_writer, attachment_writer: Some(attachment_writer),
email_ks, email_ks,
attachment_ks, attachment_ks,
pending: 0, pending: 0,
email_buf: Vec::new(),
attachment_buf: Vec::new(),
}) })
} }
@@ -299,23 +318,11 @@ impl NewIndexWriter {
// ── detach attachments → blob ────────────────────────────────────── // ── detach attachments → blob ──────────────────────────────────────
let (stripped_eml, attachment_output) = detach_attachments_standalone(eml_bytes, &message); let (stripped_eml, attachment_output) = detach_attachments_standalone(eml_bytes, &message);
if !self // Buffer for bulk ingestion — sorted + flushed later.
.email_ks self.email_buf
.contains_key(&email_content_hash) .push((email_content_hash.clone(), stripped_eml));
.unwrap_or(false)
{
self.email_ks
.insert(&email_content_hash, stripped_eml.as_slice())
.map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?;
}
// write attachment blobs
for (hash, data) in &attachment_output.blobs { for (hash, data) in &attachment_output.blobs {
if !self.attachment_ks.contains_key(hash).unwrap_or(false) { self.attachment_buf.push((hash.clone(), data.to_vec()));
self.attachment_ks
.insert(hash, data.as_ref())
.map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?;
}
} }
// ── build envelope doc ──────────────────────────────────────────── // ── build envelope doc ────────────────────────────────────────────
@@ -390,35 +397,151 @@ impl NewIndexWriter {
let envelope_doc = ea.to_document(&text, 0)?; let envelope_doc = ea.to_document(&text, 0)?;
self.envelope_writer self.envelope_writer
.as_mut()
.unwrap()
.add_document(envelope_doc) .add_document(envelope_doc)
.map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?;
for doc in attachment_docs { for doc in attachment_docs {
self.attachment_writer self.attachment_writer
.as_mut()
.unwrap()
.add_document(doc) .add_document(doc)
.map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?;
} }
self.pending += 1; self.pending += 1;
if self.pending >= COMMIT_THRESHOLD { // if self.pending >= COMMIT_THRESHOLD {
self.commit()?; // self.commit()?;
} // }
Ok(()) Ok(())
} }
pub fn commit(&mut self) -> BichonResult<()> { /// Commit pending Tantivy documents (mid-stream) — frees the in-memory
/// term dictionary / postings that accumulate in the IndexWriter.
fn commit_tantivy(&mut self) -> BichonResult<()> {
if self.pending == 0 { if self.pending == 0 {
return Ok(()); return Ok(());
} }
self.envelope_writer println!("Tantivy committing... this may take 2-3 minutes, please wait.");
.commit() let start = Instant::now();
.map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; if let Some(writer) = self.envelope_writer.as_mut() {
self.attachment_writer writer
.commit() .commit()
.map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?;
tracing::info!(count = self.pending, "committed batch"); }
if let Some(writer) = self.attachment_writer.as_mut() {
writer
.commit()
.map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?;
}
println!("tantivy commit elasped: {:#?}", start.elapsed());
tracing::info!(count = self.pending, "committed tantivy batch");
self.pending = 0; self.pending = 0;
Ok(()) Ok(())
} }
/// Final commit + segment merge for Tantivy writers (called once at end).
pub fn finish_writers(&mut self) -> BichonResult<()> {
self.commit_tantivy()?;
for (name, writer_opt) in [
("envelope", &mut self.envelope_writer),
("attachment", &mut self.attachment_writer),
] {
if let Some(writer) = writer_opt.as_mut() {
let reader = writer
.index()
.reader()
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?;
let seg_ids: Vec<_> = reader
.searcher()
.segment_readers()
.iter()
.map(|r| r.segment_id())
.collect();
println!("merging {} {} segments...", seg_ids.len(), name);
if seg_ids.len() > 1 {
let _ = writer.merge(&seg_ids);
}
}
if let Some(writer) = writer_opt.take() {
println!("waiting for {} merge to finish...", name);
let start = std::time::Instant::now();
let _ = writer.wait_merging_threads();
println!("{} merge done: {:#?}", name, start.elapsed());
}
}
Ok(())
}
/// Sort buffered (hash, data) pairs, dedup, and write via Fjall's
/// ingestion API — writes SSTables directly, bypassing memtable and WAL.
/// Also commits the Tantivy writers to bound their in-memory state.
pub fn flush_fjall_buffers(&mut self) -> BichonResult<()> {
self.commit_tantivy()?;
if !self.email_buf.is_empty() {
self.email_buf.sort_by(|a, b| a.0.cmp(&b.0));
self.email_buf.dedup_by(|a, b| a.0 == b.0);
let mut ingestion = self.email_ks.start_ingestion().map_err(|e| {
raise_error!(
format!("email ingestion start: {e:#?}"),
ErrorCode::InternalError
)
})?;
for (hash, data) in &self.email_buf {
ingestion
.write(hash.as_bytes(), data.as_slice())
.map_err(|e| {
raise_error!(
format!("email ingestion write: {e:#?}"),
ErrorCode::InternalError
)
})?;
}
ingestion.finish().map_err(|e| {
raise_error!(
format!("email ingestion finish: {e:#?}"),
ErrorCode::InternalError
)
})?;
self.email_buf.clear();
}
if !self.attachment_buf.is_empty() {
self.attachment_buf.sort_by(|a, b| a.0.cmp(&b.0));
self.attachment_buf.dedup_by(|a, b| a.0 == b.0);
let mut ingestion = self.attachment_ks.start_ingestion().map_err(|e| {
raise_error!(
format!("attachment ingestion start: {e:#?}"),
ErrorCode::InternalError
)
})?;
for (hash, data) in &self.attachment_buf {
ingestion
.write(hash.as_bytes(), data.as_slice())
.map_err(|e| {
raise_error!(
format!("attachment ingestion write: {e:#?}"),
ErrorCode::InternalError
)
})?;
}
ingestion.finish().map_err(|e| {
raise_error!(
format!("attachment ingestion finish: {e:#?}"),
ErrorCode::InternalError
)
})?;
self.attachment_buf.clear();
}
Ok(())
}
} }
@@ -78,6 +78,10 @@ pub struct IndexManager {
} }
impl IndexManager { impl IndexManager {
pub(crate) fn index_writer(&self) -> &Arc<Mutex<IndexWriter>> {
&self.index_writer
}
pub async fn shutdown(&self) { pub async fn shutdown(&self) {
let mut guard = self.handle.lock().await; let mut guard = self.handle.lock().await;
if let Some(handle) = guard.take() { if let Some(handle) = guard.take() {
+711
View File
@@ -0,0 +1,711 @@
use std::collections::HashMap;
use tantivy::schema::Term;
use tantivy::{IndexReader, IndexWriter};
use crate::common::periodic::{PeriodicTask, TaskHandle};
use crate::context::BichonTask;
use crate::error::code::ErrorCode;
use crate::error::BichonResult;
use crate::raise_error;
use crate::store::tantivy::attachment::ATTACHMENT_MANAGER;
use crate::store::tantivy::envelope::ENVELOPE_MANAGER;
use crate::store::tantivy::fields::{
F_ACCOUNT_ID, F_CONTENT_HASH, F_ID, F_INGEST_AT, F_MAILBOX_ID,
};
use crate::store::tantivy::schema::SchemaTools;
// ─── Types ────────────────────────────────────────────────────────────────────
/// A single document candidate for deduplication.
/// Holds just enough information to compare and delete duplicates.
struct DedupEntry {
/// Unix timestamp (seconds) when this document was ingested.
/// Used to determine which copy to keep: we always keep the latest,
/// so that a post-uidvalidity-reset uid is preferred over a stale one.
ingest_at: i64,
/// The email's f_id value, used to delete the duplicate email (via term
/// query on f_id) and to cascade-delete attachments whose f_envelope_id
/// matches this id.
email_id: String,
}
/// Dedup map for one account.
/// Key = (mailbox_id, content_hash) — stable identity across uidvalidity resets
/// Value = all documents sharing that key, to be reduced to exactly one.
type DedupMap = HashMap<(u64, String), Vec<DedupEntry>>;
// ─── Public entry point ───────────────────────────────────────────────────────
/// Background deduplication task.
///
/// Iterates over every account found in the index and removes duplicate emails
/// within each (mailbox_id, content_hash) group, keeping the most recently
/// ingested copy.
///
/// For each duplicate email removed, all attachments in the attachment index
/// whose f_envelope_id matches the removed email's f_id are also deleted.
///
/// Why keep the *latest* ingest_at?
/// After a uidvalidity reset the server reassigns UIDs. If we kept an old
/// copy (lower ingest_at) its uid would be stale, and uid-based incremental
/// sync would re-download emails that are already present.
///
/// Processing is done account-by-account so that peak memory is bounded by
/// the largest single account rather than the entire index.
pub async fn dedup_task(
email_reader: &IndexReader,
email_writer: &mut IndexWriter,
attachment_writer: &mut IndexWriter,
) -> BichonResult<()> {
let account_ids = collect_account_ids(email_reader)?;
let mut total_deleted = 0u64;
for account_id in account_ids {
total_deleted += dedup_account(email_reader, email_writer, attachment_writer, account_id)?;
}
tracing::info!("dedup: finished, total removed={}", total_deleted);
Ok(())
}
// ─── Periodic task ──────────────────────────────────────────────────────────
const DEDUP_INTERVAL: std::time::Duration = std::time::Duration::from_secs(12 * 60 * 60);
/// Periodically scans the email index for duplicate (mailbox_id, content_hash)
/// entries and removes redundant copies, keeping the most recently ingested one.
/// Attachments belonging to removed emails are cascade-deleted from the
/// attachment index.
pub struct DedupTask;
impl BichonTask for DedupTask {
fn start() -> TaskHandle {
let periodic_task = PeriodicTask::new("index-dedup");
let task = move |_: Option<u64>| {
Box::pin(async move {
// Acquire both writers before creating a reader. The fresh reader
// sees the last committed state, while the writers ensure we have
// exclusive access to perform deletions.
let mut email_writer = ENVELOPE_MANAGER.index_writer().lock().await;
let mut attach_writer = ATTACHMENT_MANAGER.index_writer().lock().await;
let email_reader = ENVELOPE_MANAGER.create_reader()?;
dedup_task(&email_reader, &mut email_writer, &mut attach_writer).await?;
// Commit any remaining changes from the dedup pass.
// dedup_account commits per-account, but we ensure a final commit
// so the attachment index is in sync.
crate::store::tantivy::fatal_commit(&mut attach_writer);
drop(attach_writer);
drop(email_writer);
Ok(())
})
};
periodic_task.start(task, None, DEDUP_INTERVAL, false, false)
}
}
// ─── Internals ────────────────────────────────────────────────────────────────
/// Collect the distinct set of account_ids present in the index.
///
/// Scans only the account_id FAST column — no stored field reads, no I/O
/// beyond the column file itself.
fn collect_account_ids(reader: &IndexReader) -> BichonResult<Vec<u64>> {
let searcher = reader.searcher();
let mut ids = std::collections::HashSet::new();
for segment_reader in searcher.segment_readers() {
let account_col = segment_reader
.fast_fields()
.u64(F_ACCOUNT_ID)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?;
let max_doc = segment_reader.max_doc();
for doc_id in 0..max_doc {
// Skip documents that have already been soft-deleted
if segment_reader.is_deleted(doc_id) {
continue;
}
ids.insert(account_col.values.get_val(doc_id));
}
}
Ok(ids.into_iter().collect())
}
/// Deduplicate all emails for a single account, and cascade-delete their attachments.
///
/// Strategy:
/// 1. Scan FAST columns for (mailbox_id, content_hash, ingest_at, f_id) — no heap reads.
/// 2. Group by (mailbox_id, content_hash).
/// 3. Within each group, sort descending by ingest_at and soft-delete all
/// but the first (most recent) entry.
/// 4. For each removed email, delete all attachments in the attachment index
/// whose f_envelope_id matches the removed email's f_id.
/// 5. Commit both writers once per account so memory is released before the
/// next account is processed.
///
/// Peak memory for this function ≈ account_email_count × ~160 bytes
/// (the extra ~80 bytes over previous version comes from storing email_id strings).
fn dedup_account(
email_reader: &IndexReader,
email_writer: &mut IndexWriter,
attachment_writer: &mut IndexWriter,
account_id: u64,
) -> BichonResult<u64> {
let searcher = email_reader.searcher();
let fields = SchemaTools::email_fields();
eprintln!(
"DEBUG dedup_account: entry account={account_id} f_id_field={:?} f_content_hash_field={:?}",
fields.f_id, fields.f_content_hash
);
let mut map: DedupMap = HashMap::new();
// ── Phase 1: build the dedup map via FAST column scans ──────────────────
for segment_reader in searcher.segment_readers() {
let account_col = segment_reader
.fast_fields()
.u64(F_ACCOUNT_ID)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?;
let mailbox_col = segment_reader
.fast_fields()
.u64(F_MAILBOX_ID)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?;
let ingest_col = segment_reader
.fast_fields()
.i64(F_INGEST_AT)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?;
// content_hash and f_id are text fields with FAST; stored as dictionary-encoded strings
let hash_col = segment_reader
.fast_fields()
.str(F_CONTENT_HASH)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
.ok_or_else(|| raise_error!(format!("FAST str column '{}' not found in segment; ensure the field is declared with FAST in the schema", F_CONTENT_HASH), ErrorCode::InternalError))?;
let id_col = segment_reader
.fast_fields()
.str(F_ID)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?
.ok_or_else(|| raise_error!(format!("FAST str column '{}' not found in segment; ensure the field is declared with FAST in the schema", F_ID), ErrorCode::InternalError))?;
let max_doc = segment_reader.max_doc();
for doc_id in 0..max_doc {
if segment_reader.is_deleted(doc_id) {
continue;
}
// Filter to the current account without touching stored fields
if account_col.values.get_val(doc_id) != account_id {
continue;
}
let mailbox_id = mailbox_col.values.get_val(doc_id);
let ingest_at = ingest_col.values.get_val(doc_id);
// Read content_hash from the dictionary-encoded string column
let hash_ord = hash_col.ords().values_for_doc(doc_id as u32).next().unwrap_or(0);
let mut hash_buf = String::new();
hash_col
.ord_to_str(hash_ord, &mut hash_buf)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?;
let content_hash = hash_buf;
// Read f_id from the dictionary-encoded string column
let id_ord = id_col.ords().values_for_doc(doc_id as u32).next().unwrap_or(0);
let mut id_buf = String::new();
id_col
.ord_to_str(id_ord, &mut id_buf)
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?;
let email_id = id_buf;
eprintln!(
"DEBUG dedup_account: account={account_id} doc_id={doc_id} mailbox={mailbox_id} hash={content_hash:?} id={email_id:?} ingest_at={ingest_at}"
);
map.entry((mailbox_id, content_hash))
.or_default()
.push(DedupEntry {
ingest_at,
email_id,
});
}
}
// ── Phase 2: delete duplicate emails and their attachments ───────────────
let attachment_fields = SchemaTools::attachment_fields();
let mut deleted = 0u64;
for (_key, mut entries) in map {
if entries.len() <= 1 {
// No duplicates in this group
continue;
}
// Sort descending: the most recently ingested document comes first.
// This ensures we keep the copy whose uid reflects the current
// uidvalidity, which is required for correct incremental sync.
entries.sort_by_key(|e| std::cmp::Reverse(e.ingest_at));
eprintln!("DEBUG Phase2: key={_key:?} kept={} deleting={}", entries[0].email_id, entries.len() - 1);
// Keep entries[0], soft-delete everything else via term query on f_id
for entry in &entries[1..] {
eprintln!(
"DEBUG Phase2: delete_term f_id={:?} text=\"{}\"",
fields.f_id, &entry.email_id
);
// Remove the duplicate email from the email index
let email_term = Term::from_field_text(fields.f_id, &entry.email_id);
email_writer.delete_term(email_term);
// Cascade: remove all attachments belonging to this email.
// f_envelope_id in the attachment index mirrors f_id in the email index.
let envelope_term =
Term::from_field_text(attachment_fields.f_envelope_id, &entry.email_id);
attachment_writer.delete_term(envelope_term);
deleted += 1;
}
}
// Commit both indexes once per account so the DedupMap memory for this
// account can be reclaimed before the next account is processed.
if deleted > 0 {
email_writer
.commit()
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?;
attachment_writer
.commit()
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?;
tracing::info!("dedup: account={} removed={}", account_id, deleted);
}
Ok(deleted)
}
// ─── Tests ─────────────────────────────────────────────────────────────────────
#[cfg(test)]
mod tests {
use super::*;
use crate::store::tantivy::fields::AttachmentFields;
use crate::store::tantivy::fields::EmailFields;
use crate::store::tantivy::schema::SchemaTools;
use crate::store::tantivy::tokenizers::EuroTokenizer;
use std::collections::HashSet;
use std::fmt::Write;
use std::fs;
use tantivy::Index;
use tantivy::TantivyDocument;
fn temp_dir(prefix: &str) -> std::path::PathBuf {
let dir = std::env::temp_dir()
.join("bichon-dedup-test")
.join(prefix)
.join(uuid::Uuid::new_v4().to_string());
fs::create_dir_all(&dir).unwrap();
dir
}
fn make_index(dir: &std::path::Path, schema: tantivy::schema::Schema) -> Index {
let index = Index::create_in_dir(dir, schema).unwrap();
index.tokenizers().register("euro", EuroTokenizer::new());
index
}
/// Collect non-deleted f_id values from the email index.
fn surviving_email_ids(reader: &IndexReader) -> HashSet<String> {
reader
.reload()
.expect("reader reload failed");
let searcher = reader.searcher();
let mut ids = HashSet::new();
let segments = searcher.segment_readers();
eprintln!("DEBUG surviving_email_ids: segment_count={}", segments.len());
for (seg_idx, seg) in segments.iter().enumerate() {
let id_col = seg
.fast_fields()
.str(F_ID)
.unwrap()
.expect("FAST str column 'id' missing");
let max_doc = seg.max_doc();
eprintln!("DEBUG surviving_email_ids: seg={seg_idx} max_doc={max_doc}");
for doc_id in 0..max_doc {
let is_del = seg.is_deleted(doc_id);
let ord = id_col.ords().values_for_doc(doc_id as u32).next().unwrap_or(0);
let mut buf = String::new();
id_col.ord_to_str(ord, &mut buf).unwrap();
eprintln!("DEBUG surviving_email_ids: seg={seg_idx} doc_id={doc_id} is_deleted={is_del} ord={ord} buf={buf:?}");
if !is_del {
ids.insert(buf);
}
}
}
ids
}
/// Collect non-deleted f_id values from the attachment index.
fn surviving_attachment_ids(reader: &IndexReader) -> HashSet<String> {
let searcher = reader.searcher();
let mut ids = HashSet::new();
for seg in searcher.segment_readers() {
let id_col = seg
.fast_fields()
.str(F_ID)
.unwrap()
.expect("FAST str column 'id' missing in attachment index");
let max_doc = seg.max_doc();
for doc_id in 0..max_doc {
if seg.is_deleted(doc_id) {
continue;
}
let ord = id_col.ords().values_for_doc(doc_id as u32).next().unwrap_or(0);
let mut buf = String::new();
id_col.ord_to_str(ord, &mut buf).unwrap();
ids.insert(buf);
}
}
ids
}
fn add_email(
f: &EmailFields,
w: &mut IndexWriter,
id: &str,
account: u64,
mailbox: u64,
hash: &str,
ingest_at: i64,
) {
let mut doc = TantivyDocument::new();
doc.add_text(f.f_id, id);
doc.add_u64(f.f_account_id, account);
doc.add_u64(f.f_mailbox_id, mailbox);
doc.add_text(f.f_content_hash, hash);
doc.add_i64(f.f_ingest_at, ingest_at);
w.add_document(doc).unwrap();
}
fn add_attachment(
f: &AttachmentFields,
w: &mut IndexWriter,
id: &str,
envelope_id: &str,
account: u64,
mailbox: u64,
) {
let mut doc = TantivyDocument::new();
doc.add_text(f.f_id, id);
doc.add_text(f.f_envelope_id, envelope_id);
doc.add_u64(f.f_account_id, account);
doc.add_u64(f.f_mailbox_id, mailbox);
w.add_document(doc).unwrap();
}
/// Prevent segment merges during dedup so delete operations are isolated
/// and test assertions target the exact expected document set.
fn apply_no_merge_policy(w: &mut IndexWriter) {
let mut mp = tantivy::indexer::LogMergePolicy::default();
mp.set_min_num_segments(500);
mp.set_max_docs_before_merge(1_000_000);
w.set_merge_policy(Box::new(mp));
}
struct Harness;
impl Harness {
async fn run<F>(
case: &str,
populate: F,
expected_emails: &[&str],
expected_attachments: &[&str],
) where
F: FnOnce(&EmailFields, &mut IndexWriter, &AttachmentFields, &mut IndexWriter),
{
let email_schema = SchemaTools::email_schema();
let attach_schema = SchemaTools::attachment_schema();
let email_f = SchemaTools::email_fields();
let attach_f = SchemaTools::attachment_fields();
let email_idx = make_index(&temp_dir(case), email_schema);
let attach_idx = make_index(&temp_dir(case), attach_schema);
let mut email_w = email_idx.writer_with_num_threads(1, 50_000_000).unwrap();
let mut attach_w = attach_idx.writer_with_num_threads(1, 50_000_000).unwrap();
apply_no_merge_policy(&mut email_w);
apply_no_merge_policy(&mut attach_w);
populate(&email_f, &mut email_w, &attach_f, &mut attach_w);
email_w.commit().unwrap();
attach_w.commit().unwrap();
drop(email_w);
drop(attach_w);
let mut email_w2 = email_idx.writer_with_num_threads(1, 50_000_000).unwrap();
let mut attach_w2 = attach_idx.writer_with_num_threads(1, 50_000_000).unwrap();
apply_no_merge_policy(&mut email_w2);
apply_no_merge_policy(&mut attach_w2);
let email_r = email_idx.reader().unwrap();
dedup_task(&email_r, &mut email_w2, &mut attach_w2)
.await
.unwrap();
let email_r = email_idx.reader().unwrap();
let survivors = surviving_email_ids(&email_r);
let expected: HashSet<String> =
expected_emails.iter().map(|s| s.to_string()).collect();
assert_eq!(survivors, expected, "[{case}] email survivors mismatch");
let attach_r = attach_idx.reader().unwrap();
let att_survivors = surviving_attachment_ids(&attach_r);
let att_expected: HashSet<String> =
expected_attachments.iter().map(|s| s.to_string()).collect();
assert_eq!(att_survivors, att_expected, "[{case}] attachment survivors mismatch");
}
}
#[tokio::test]
async fn dedup_removes_duplicates_and_cascades_to_attachments() {
Harness::run(
"basic",
|ef, ew, af, aw| {
add_email(ef, ew, "dup-old", 1, 200, "hash-dup", 1000);
add_email(ef, ew, "dup-new", 1, 200, "hash-dup", 3000);
add_email(ef, ew, "unique", 1, 200, "hash-uniq", 1000);
add_attachment(af, aw, "att-old", "dup-old", 1, 200);
add_attachment(af, aw, "att-new", "dup-new", 1, 200);
},
&["dup-new", "unique"],
&["att-new"],
)
.await;
}
#[tokio::test]
async fn dedup_keeps_latest_among_many_duplicates() {
Harness::run(
"many-dups",
|ef, ew, af, aw| {
for (i, ts) in [50, 100, 400, 200, 300].iter().enumerate() {
let id = format!("dup-{i}");
add_email(ef, ew, &id, 1, 1, "H", *ts);
add_attachment(af, aw, &format!("att-{i}"), &id, 1, 1);
}
},
&["dup-2"], // ingest_at=400, the latest
&["att-2"],
)
.await;
}
#[tokio::test]
async fn dedup_no_duplicates_is_noop() {
Harness::run(
"no-dups",
|ef, ew, af, aw| {
add_email(ef, ew, "a", 1, 1, "hash-a", 100);
add_email(ef, ew, "b", 1, 1, "hash-b", 200);
add_email(ef, ew, "c", 1, 1, "hash-c", 300);
add_attachment(af, aw, "att-a", "a", 1, 1);
add_attachment(af, aw, "att-b", "b", 1, 1);
add_attachment(af, aw, "att-c", "c", 1, 1);
},
&["a", "b", "c"],
&["att-a", "att-b", "att-c"],
)
.await;
}
#[tokio::test]
async fn dedup_isolates_accounts() {
// Same hash, same mailbox, DIFFERENT accounts → no dedup
Harness::run(
"cross-account",
|ef, ew, af, aw| {
add_email(ef, ew, "acc1-a", 1, 1, "hash-same", 100);
add_email(ef, ew, "acc1-b", 1, 1, "hash-same", 200);
add_email(ef, ew, "acc2-a", 2, 1, "hash-same", 100);
add_email(ef, ew, "acc2-b", 2, 1, "hash-same", 200);
add_attachment(af, aw, "att-1a", "acc1-a", 1, 1);
add_attachment(af, aw, "att-1b", "acc1-b", 1, 1);
add_attachment(af, aw, "att-2a", "acc2-a", 2, 1);
add_attachment(af, aw, "att-2b", "acc2-b", 2, 1);
},
// Each account keeps its latest: acc1 keeps acc1-b (200>100), acc2 keeps acc2-b
&["acc1-b", "acc2-b"],
&["att-1b", "att-2b"],
)
.await;
}
#[tokio::test]
async fn dedup_isolates_mailboxes() {
// Same hash, same account, DIFFERENT mailboxes → no dedup
Harness::run(
"cross-mailbox",
|ef, ew, af, aw| {
add_email(ef, ew, "mb1-a", 1, 1, "hash-same", 100);
add_email(ef, ew, "mb2-a", 1, 2, "hash-same", 100);
add_email(ef, ew, "mb1-b", 1, 1, "hash-same", 200);
add_email(ef, ew, "mb2-b", 1, 2, "hash-same", 200);
add_attachment(af, aw, "att-1a", "mb1-a", 1, 1);
add_attachment(af, aw, "att-1b", "mb1-b", 1, 1);
add_attachment(af, aw, "att-2a", "mb2-a", 1, 2);
add_attachment(af, aw, "att-2b", "mb2-b", 1, 2);
},
&["mb1-b", "mb2-b"],
&["att-1b", "att-2b"],
)
.await;
}
#[tokio::test]
async fn dedup_multiple_attachments_per_email() {
// Deleting an email cascades all its attachments, not just one
Harness::run(
"multi-att",
|ef, ew, af, aw| {
add_email(ef, ew, "old", 1, 1, "H", 100);
add_email(ef, ew, "new", 1, 1, "H", 200);
// The old email has 3 attachments — all should be removed
add_attachment(af, aw, "att1", "old", 1, 1);
add_attachment(af, aw, "att2", "old", 1, 1);
add_attachment(af, aw, "att3", "old", 1, 1);
// The kept email has 2 attachments — both should survive
add_attachment(af, aw, "att4", "new", 1, 1);
add_attachment(af, aw, "att5", "new", 1, 1);
},
&["new"],
&["att4", "att5"],
)
.await;
}
/// Inspects the production email index and reports duplicate counts.
///
/// A "duplicate" is defined as two or more emails sharing the same
/// (account_id, mailbox_id, content_hash) tuple.
///
/// This test is read-only — it does not modify the index.
#[test]
fn inspect_production_duplicates() {
let index_path = r"E:\db\data\bichon-indices\mail_metadata";
let report_path = std::path::PathBuf::from(r"E:\bichon\dedup_report.txt");
let mut report = String::new();
let _ = writeln!(report, "opening index at {index_path}...");
let index = match Index::open_in_dir(index_path) {
Ok(idx) => {
let _ = writeln!(report, "index opened successfully");
idx
}
Err(e) => {
let _ = writeln!(report, "Failed to open index at {index_path}: {e}");
let _ = std::fs::write(&report_path, &report);
return;
}
};
let reader = match index.reader() {
Ok(r) => r,
Err(e) => {
let _ = writeln!(report, "Failed to create reader: {e}");
let _ = std::fs::write(&report_path, &report);
return;
}
};
reader.reload().expect("reader reload failed");
let searcher = reader.searcher();
let mut total_docs = 0u64;
let mut groups: std::collections::HashMap<u64, std::collections::HashMap<(u64, String), u64>> =
std::collections::HashMap::new();
for segment_reader in searcher.segment_readers() {
let account_col = segment_reader
.fast_fields()
.u64(F_ACCOUNT_ID)
.unwrap();
let mailbox_col = segment_reader
.fast_fields()
.u64(F_MAILBOX_ID)
.unwrap();
let hash_col = match segment_reader
.fast_fields()
.str(F_CONTENT_HASH)
.unwrap()
{
Some(c) => c,
None => {
let _ = writeln!(report, "Segment has no FAST str column for content_hash, skipping");
continue;
}
};
let max_doc = segment_reader.max_doc();
for doc_id in 0..max_doc {
if segment_reader.is_deleted(doc_id) {
continue;
}
let account_id = account_col.values.get_val(doc_id);
let mailbox_id = mailbox_col.values.get_val(doc_id);
let hash_ord = hash_col.ords().values_for_doc(doc_id as u32).next().unwrap_or(0);
let mut hash_buf = String::new();
hash_col.ord_to_str(hash_ord, &mut hash_buf).unwrap();
let content_hash = hash_buf;
total_docs += 1;
groups
.entry(account_id)
.or_default()
.entry((mailbox_id, content_hash))
.and_modify(|c| *c += 1)
.or_insert(1);
}
}
// ── Summarize ──────────────────────────────────────────────────────────
let mut total_duplicate_groups = 0u64;
let mut total_duplicate_emails = 0u64;
for (account_id, account_groups) in &groups {
let mut account_dup_groups = 0u64;
let mut account_dup_emails = 0u64;
for ((_mailbox_id, _hash), count) in account_groups {
if *count > 1 {
account_dup_groups += 1;
account_dup_emails += count - 1;
}
}
if account_dup_groups > 0 {
let _ = writeln!(
report,
"account={account_id}: {account_dup_groups} duplicate groups, {account_dup_emails} redundant emails"
);
}
total_duplicate_groups += account_dup_groups;
total_duplicate_emails += account_dup_emails;
}
let _ = writeln!(
report,
"─── Summary ───\n\
total_docs = {total_docs}\n\
accounts = {}\n\
duplicate_groups = {total_duplicate_groups}\n\
redundant_emails = {total_duplicate_emails}\n\
unique_after_dedup = {}",
groups.len(),
total_docs - total_duplicate_emails,
);
std::fs::write(&report_path, &report).unwrap();
println!("report written to {}", report_path.display());
}
}
+10
View File
@@ -86,6 +86,16 @@ pub struct IndexManager {
} }
impl IndexManager { impl IndexManager {
pub(crate) fn index_writer(&self) -> &Arc<Mutex<IndexWriter>> {
&self.index_writer
}
pub(crate) fn create_reader(&self) -> BichonResult<IndexReader> {
self.index
.reader()
.map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))
}
pub async fn shutdown(&self) { pub async fn shutdown(&self) {
let mut guard = self.handle.lock().await; let mut guard = self.handle.lock().await;
if let Some(handle) = guard.take() { if let Some(handle) = guard.take() {
+1
View File
@@ -24,6 +24,7 @@ use crate::{
}; };
pub mod attachment; pub mod attachment;
pub mod dedup;
pub mod envelope; pub mod envelope;
pub mod fields; pub mod fields;
pub mod filter; pub mod filter;
+3
View File
@@ -19,6 +19,8 @@
use crate::common::periodic::TaskHandle; use crate::common::periodic::TaskHandle;
use crate::context::BichonTask; use crate::context::BichonTask;
use crate::oauth2::{refresh::OAuth2RefreshTask, task::OAuth2CleanTask}; use crate::oauth2::{refresh::OAuth2RefreshTask, task::OAuth2CleanTask};
use crate::store::tantivy::dedup::DedupTask;
pub struct PeriodicTasks { pub struct PeriodicTasks {
tasks: Vec<TaskHandle>, tasks: Vec<TaskHandle>,
} }
@@ -28,6 +30,7 @@ impl PeriodicTasks {
let mut tasks = Vec::new(); let mut tasks = Vec::new();
tasks.push(OAuth2CleanTask::start()); tasks.push(OAuth2CleanTask::start());
tasks.push(OAuth2RefreshTask::start()); tasks.push(OAuth2RefreshTask::start());
tasks.push(DedupTask::start());
Self { tasks } Self { tasks }
} }
+136
View File
@@ -0,0 +1,136 @@
// Temporary mock data for download-folders responsive testing
import { MailboxData, MailboxListResponse } from './api';
let _id = 1;
const m = (overrides: Partial<MailboxData>): MailboxData => ({
account_id: 1,
attributes: [],
delimiter: '/',
exists: Math.floor(Math.random() * 5000) + 100,
id: _id++,
name: '',
uid_next: null,
uid_validity: null,
unseen: null,
...overrides,
});
const SHORT = [
m({ name: 'INBOX' }),
m({ name: 'INBOX/Drafts' }),
m({ name: 'INBOX/Sent' }),
m({ name: 'INBOX/Trash' }),
m({ name: 'INBOX/Archive' }),
m({ name: 'INBOX/Spam' }),
m({ name: 'INBOX/Templates' }),
];
const PROJECTS = [
m({ name: 'INBOX/Projects' }),
m({ name: 'INBOX/Projects/Q4 2025 Financial Reports and Annual Budget Planning Review' }),
m({ name: 'INBOX/Projects/Q4 2025 Financial Reports and Annual Budget Planning Review/Drafts' }),
m({
name: 'INBOX/Projects/Q4 2025 Financial Reports and Annual Budget Planning Review/Drafts/Revision 3 - Updated Forecast Models and Department Sign-off Required',
attributes: [{ attr: 'HasChildren', extension: null }],
}),
m({
name: 'INBOX/Projects/Q4 2025 Financial Reports and Annual Budget Planning Review/Drafts/Revision 3 - Updated Forecast Models and Department Sign-off Required/Comments from CFO',
}),
m({
name: 'INBOX/Projects/Q4 2025 Financial Reports and Annual Budget Planning Review/Drafts/Revision 3 - Updated Forecast Models and Department Sign-off Required/Attachments',
}),
m({ name: 'INBOX/Projects/Q4 2025 Financial Reports and Annual Budget Planning Review/Final' }),
m({ name: 'INBOX/Projects/Q4 2025 Financial Reports and Annual Budget Planning Review/Final/Approved with Amendments' }),
m({ name: 'INBOX/Projects/Q1 2026 Strategic Initiative - Cloud Infrastructure Migration Assessment and Vendor Selection' }),
m({ name: 'INBOX/Projects/Q1 2026 Strategic Initiative - Cloud Infrastructure Migration Assessment and Vendor Selection/Kickoff' }),
m({ name: 'INBOX/Projects/Q1 2026 Strategic Initiative - Cloud Infrastructure Migration Assessment and Vendor Selection/Kickoff/Meeting Minutes and Action Items' }),
m({
name: 'INBOX/Projects/Q1 2026 Strategic Initiative - Cloud Infrastructure Migration Assessment and Vendor Selection/Vendor Proposals',
attributes: [{ attr: 'HasNoChildren', extension: null }],
}),
m({ name: 'INBOX/Projects/Q1 2026 Strategic Initiative - Cloud Infrastructure Migration Assessment and Vendor Selection/Vendor Proposals/AWS Proposal Package' }),
m({ name: 'INBOX/Projects/Q1 2026 Strategic Initiative - Cloud Infrastructure Migration Assessment and Vendor Selection/Vendor Proposals/Azure Proposal Package' }),
m({ name: 'INBOX/Projects/Q1 2026 Strategic Initiative - Cloud Infrastructure Migration Assessment and Vendor Selection/Vendor Proposals/GCP Proposal Package' }),
m({ name: 'INBOX/Projects/Q1 2026 Strategic Initiative - Cloud Infrastructure Migration Assessment and Vendor Selection/Internal Review and Scoring Committee' }),
m({ name: 'INBOX/Projects/HR Portal Redesign - Employee Self-Service Platform Modernization' }),
m({ name: 'INBOX/Projects/HR Portal Redesign - Employee Self-Service Platform Modernization/Wireframes and Mockups' }),
m({ name: 'INBOX/Projects/HR Portal Redesign - Employee Self-Service Platform Modernization/Wireframes and Mockups/Iteration 1' }),
m({ name: 'INBOX/Projects/HR Portal Redesign - Employee Self-Service Platform Modernization/Wireframes and Mockups/Iteration 2' }),
m({ name: 'INBOX/Projects/HR Portal Redesign - Employee Self-Service Platform Modernization/Usability Testing Results and Feedback Compilation' }),
];
const CLIENTS = [
m({ name: 'INBOX/Clients' }),
m({
name: 'INBOX/Clients/Acme Corporation - Enterprise Software Licensing and Support Agreement Renewal 2026',
attributes: [{ attr: 'HasChildren', extension: null }],
}),
m({ name: 'INBOX/Clients/Acme Corporation - Enterprise Software Licensing and Support Agreement Renewal 2026/Contract Documents' }),
m({ name: 'INBOX/Clients/Acme Corporation - Enterprise Software Licensing and Support Agreement Renewal 2026/Contract Documents/Redlined Versions' }),
m({
name: 'INBOX/Clients/Acme Corporation - Enterprise Software Licensing and Support Agreement Renewal 2026/Contract Documents/Redlined Versions/Legal Review Round 1',
}),
m({
name: 'INBOX/Clients/Acme Corporation - Enterprise Software Licensing and Support Agreement Renewal 2026/Contract Documents/Redlined Versions/Legal Review Round 2 - Final',
}),
m({
name: 'INBOX/Clients/Acme Corporation - Enterprise Software Licensing and Support Agreement Renewal 2026/Invoices and Payment Records',
}),
m({ name: 'INBOX/Clients/Acme Corporation - Enterprise Software Licensing and Support Agreement Renewal 2026/Support Tickets and Correspondence' }),
m({
name: 'INBOX/Clients/Globex Industries - Multi-Year Digital Transformation Consulting Engagement',
attributes: [{ attr: 'HasChildren', extension: null }],
}),
m({ name: 'INBOX/Clients/Globex Industries - Multi-Year Digital Transformation Consulting Engagement/Phase 1 Discovery and Assessment' }),
m({ name: 'INBOX/Clients/Globex Industries - Multi-Year Digital Transformation Consulting Engagement/Phase 1 Discovery and Assessment/Stakeholder Interviews' }),
m({ name: 'INBOX/Clients/Globex Industries - Multi-Year Digital Transformation Consulting Engagement/Phase 1 Discovery and Assessment/Current State Architecture Documentation' }),
m({ name: 'INBOX/Clients/Globex Industries - Multi-Year Digital Transformation Consulting Engagement/Phase 2 Implementation Roadmap' }),
m({ name: 'INBOX/Clients/Globex Industries - Multi-Year Digital Transformation Consulting Engagement/Phase 2 Implementation Roadmap/Sprint Planning and Resource Allocation' }),
m({ name: 'INBOX/Clients/Globex Industries - Multi-Year Digital Transformation Consulting Engagement/Phase 2 Implementation Roadmap/Risk Assessment and Mitigation Strategies' }),
m({ name: 'INBOX/Clients/Initech Solutions - Cybersecurity Audit and Compliance Remediation Program 2026' }),
m({ name: 'INBOX/Clients/Initech Solutions - Cybersecurity Audit and Compliance Remediation Program 2026/Penetration Testing Reports' }),
m({ name: 'INBOX/Clients/Initech Solutions - Cybersecurity Audit and Compliance Remediation Program 2026/Penetration Testing Reports/External Network Assessment' }),
m({ name: 'INBOX/Clients/Initech Solutions - Cybersecurity Audit and Compliance Remediation Program 2026/Penetration Testing Reports/Internal Network Assessment' }),
m({ name: 'INBOX/Clients/Initech Solutions - Cybersecurity Audit and Compliance Remediation Program 2026/Penetration Testing Reports/Web Application Security Scan Results' }),
m({ name: 'INBOX/Clients/Initech Solutions - Cybersecurity Audit and Compliance Remediation Program 2026/Compliance Gap Analysis and Remediation Tracking' }),
m({ name: 'INBOX/Clients/Massive Dynamic - Research Collaboration on Advanced Machine Learning Applications in Healthcare Informatics' }),
m({ name: 'INBOX/Clients/Massive Dynamic - Research Collaboration on Advanced Machine Learning Applications in Healthcare Informatics/Data Sharing Agreements and Ethics Board Approvals' }),
m({ name: 'INBOX/Clients/Massive Dynamic - Research Collaboration on Advanced Machine Learning Applications in Healthcare Informatics/Literature Review and Prior Art Analysis' }),
m({ name: 'INBOX/Clients/Massive Dynamic - Research Collaboration on Advanced Machine Learning Applications in Healthcare Informatics/Model Training Datasets and Validation Results' }),
];
const NOTIFICATIONS = [
m({ name: 'INBOX/Notifications' }),
m({ name: 'INBOX/Notifications/GitHub Enterprise - Pull Request Reviews and CI/CD Pipeline Status Updates' }),
m({ name: 'INBOX/Notifications/Jira Service Management - Incident Response Alerts and Escalation Notifications' }),
m({ name: 'INBOX/Notifications/Confluence - Documentation Updates and Page Modification Summaries' }),
m({ name: 'INBOX/Notifications/Slack Workspace - Channel Highlights and Direct Message Digest Compilation' }),
m({ name: 'INBOX/Notifications/Datadog - Application Performance Monitoring Alerts and Anomaly Detection Reports' }),
m({ name: 'INBOX/Notifications/PagerDuty - On-Call Rotation Schedule and Incident Acknowledgment Confirmations' }),
m({ name: 'INBOX/Notifications/Microsoft 365 - Calendar Invitations and Meeting Room Booking Confirmations' }),
];
const NEWSLETTERS = [
m({ name: 'INBOX/Newsletters' }),
m({ name: 'INBOX/Newsletters/Rust Weekly - Community Updates, Crate Highlights, and RFC Progress Tracking Digest' }),
m({
name: 'INBOX/Newsletters/Systems Programming Insider - Deep Dive Articles on Memory Management and Concurrency Patterns',
attributes: [{ attr: 'HasNoChildren', extension: null }],
}),
m({ name: 'INBOX/Newsletters/Cloud Native Computing Foundation - Kubernetes Ecosystem Updates and Project Maturity Reports' }),
m({ name: 'INBOX/Newsletters/Software Architecture Monthly - Case Studies in Distributed Systems Design and Microservices Patterns' }),
m({ name: 'INBOX/Newsletters/DevOps Weekly Digest - Tool Reviews, Pipeline Optimization Techniques, and Platform Engineering Insights' }),
m({ name: 'INBOX/Newsletters/Information Security Briefing - CVE Disclosures, Threat Intelligence Reports, and Zero-Day Advisories' }),
m({ name: 'INBOX/Newsletters/Tech Leadership Forum - Engineering Management Best Practices and Organizational Scaling Strategies' }),
];
export const MOCK_MAILBOX_LIST: MailboxListResponse = {
status: 'ready',
mailboxes: [
...SHORT,
...PROJECTS,
...CLIENTS,
...NOTIFICATIONS,
...NEWSLETTERS,
],
};
+104
View File
@@ -0,0 +1,104 @@
// Temporary mock data for dashboard responsive testing
import { DashboardStats } from './api';
const NOW = Date.now();
const DAY = 86400000;
export const MOCK_DASHBOARD_STATS: DashboardStats = {
account_count: 12,
email_count: 145892,
attachment_count: 34201,
total_size_bytes: 128_849_018_880, // ~120 GB logical
storage_usage_bytes: 85_899_345_920, // ~80 GB blob
index_usage_bytes: 12_884_901_888, // ~12 GB index
recent_activity: Array.from({ length: 30 }, (_, i) => ({
timestamp_ms: NOW - (29 - i) * DAY,
count: Math.floor(Math.random() * 2000) + 200,
})),
top_senders: [
{ key: 'alexander.hamilton@verylongemaildomain-truncation-test.com', count: 4523 },
{ key: 'noreply@github-enterprise-notifications.system.example.org', count: 3891 },
{ key: 'jane.doe+project-alpha-beta-gamma@company-with-long-name.io', count: 3102 },
{ key: 'newsletter-subscriptions@really-long-marketing-domain.co.uk', count: 2845 },
{ key: 'support-tickets+priority-high@helpdesk.corporate.example.com', count: 2100 },
{ key: 'bot-pipeline-ci-cd-failures@devops.internal.long-subdomain.net', count: 1789 },
{ key: 'short@s.dev', count: 1500 },
{ key: 'alerts-monitoring-productions-east-us@observability-platform.com', count: 1256 },
{ key: 'weekly-digest-no-reply@newsletter.huge-media-conglomerate.org', count: 980 },
{ key: 'invitations-events-calendar-reminders@social-network-app.io', count: 760 },
],
top_accounts: [
{ key: 'primary.work.mailbox@enterprise-long-domain-name.com', count: 78500 },
{ key: 'personal.archive+all@very-lengthy-personal-domain.me', count: 42300 },
{ key: 'secondary.backup@another-extremely-long-domain.co', count: 15100 },
{ key: 'team-leads@department-of-engineering.corp.example.org', count: 8992 },
{ key: 'short@x.co', count: 1000 },
],
with_attachment_count: 18500,
without_attachment_count: 15701,
top_largest_emails: [
{
id: 'msg-001',
subject: 'RE: [EXTERNAL] Q4 Financial Reports & Budget Planning Documents for Review - Please Provide Feedback by EOD Friday with Department Head Sign-off Required',
size_bytes: 52_428_800,
},
{
id: 'msg-002',
subject: 'Fwd: Urgent: Client Presentation Draft - Version 7 Final (With Legal Team Amendments and Compliance Review Attached)',
size_bytes: 48_234_496,
},
{
id: 'msg-003',
subject: 'Meeting Minutes: Cross-Functional Architecture Review Session - Microservices Migration Strategy and Timeline Discussion (Part 3 of 5)',
size_bytes: 41_943_040,
},
{
id: 'msg-004',
subject: 'Invoice #INV-2026-04582 - Professional Services Engagement: Cloud Infrastructure Assessment and Remediation Planning Phase II Deliverables',
size_bytes: 38_797_312,
},
{
id: 'msg-005',
subject: '[ACTION REQUIRED] Security Incident Response: Post-Mortem Analysis and Remediation Steps for CVE-2026-12345 - Department-Wide Mandatory Review',
size_bytes: 35_651_584,
},
{
id: 'msg-006',
subject: 'Monthly Newsletter: Engineering Blog Digest - Articles on Distributed Systems, Rust Async Runtime Internals, and Performance Optimization Techniques',
size_bytes: 31_457_280,
},
{
id: 'msg-007',
subject: 'Contract Review: Master Service Agreement Amendment #7 with Third-Party Vendor Integration Services for Payment Processing Platform',
size_bytes: 28_311_552,
},
{
id: 'msg-008',
subject: 'Travel Itinerary & Expense Report: International Conference on Systems Programming - Accommodation, Flight, and Per Diem Documentation Package',
size_bytes: 25_165_824,
},
{
id: 'msg-009',
subject: 'Re: [INTERNAL] Employee Onboarding Documentation Package - Benefits Enrollment, Tax Forms, Direct Deposit Setup, and IT Access Request Forms Bundle',
size_bytes: 22_020_096,
},
{
id: 'msg-010',
subject: 'Data Export Request: Complete Transaction History 2024-2026 with Audit Trail and Compliance Certification for External Regulatory Review Board',
size_bytes: 18_874_368,
},
],
top_largest_attachments: [
{ id: 'att-001', name: 'Q4_2025_Financial_Statements_Audited_with_Supporting_Schedules_and_Notes_v3_FINAL.xlsx', size_bytes: 45_254_100 },
{ id: 'att-002', name: 'project_deliverables_package_phase_2_with_test_results_coverage_report_and_deployment_guide.zip', size_bytes: 38_900_500 },
{ id: 'att-003', name: '2026-01-15_production_database_backup_full_with_transaction_logs_and_stored_procedures.sql.gz', size_bytes: 35_200_000 },
{ id: 'att-004', name: 'client_presentation_deck_v7_final_approved_with_speaker_notes_and_embedded_video_demo.pptx', size_bytes: 31_000_000 },
{ id: 'att-005', name: 'system_architecture_diagrams_microservices_v2_with_sequence_flows_and_deployment_topology.pdf', size_bytes: 28_500_000 },
{ id: 'att-006', name: 'annual_company_event_photograph_high_resolution_group_photo_panorama_2026.jpg', size_bytes: 25_000_000 },
{ id: 'att-007', name: 'complete_source_code_archive_feature_branch_refactor_auth_module_2026_01_15.tar.gz', size_bytes: 22_800_000 },
{ id: 'att-008', name: 'product_demo_screencast_walkthrough_new_features_2026_release_candidate.mp4', size_bytes: 19_500_000 },
{ id: 'att-009', name: 'legal_contract_review_package_with_redlined_amendments_and_counsel_opinion_letters.pdf', size_bytes: 16_200_000 },
{ id: 'att-010', name: 'employee_training_module_compliance_and_security_awareness_2026_v2_interactive.iso', size_bytes: 12_800_000 },
],
system_version: '1.0.1',
};
@@ -31,13 +31,14 @@ import { Loader2, CheckSquare, Square } from 'lucide-react'
import { useCallback, useEffect, useMemo, useState } from 'react' import { useCallback, useEffect, useMemo, useState } from 'react'
import { toast } from '@/hooks/use-toast' import { toast } from '@/hooks/use-toast'
import { list_mailboxes, MailboxData } from '@/api/mailbox/api' import { list_mailboxes, MailboxData } from '@/api/mailbox/api'
//import { MOCK_MAILBOX_LIST } from '@/api/mailbox/mock-mailboxes'
import { buildTree, ExtendedTreeItemProps } from '@/lib/build-tree' import { buildTree, ExtendedTreeItemProps } from '@/lib/build-tree'
import { Skeleton } from '@/components/ui/skeleton' import { Skeleton } from '@/components/ui/skeleton'
import { AccountModel, update_account } from '@/api/account/api' import { AccountModel, update_account } from '@/api/account/api'
import { ToastAction } from '@/components/ui/toast' import { ToastAction } from '@/components/ui/toast'
import axios, { AxiosError } from 'axios' import axios, { AxiosError } from 'axios'
import { ScrollArea } from '@/components/ui/scroll-area'
import { useTranslation } from 'react-i18next' import { useTranslation } from 'react-i18next'
import { ScrollArea } from '@/components/ui/scroll-area'
import { RichTreeView } from '@mui/x-tree-view/RichTreeView'; import { RichTreeView } from '@mui/x-tree-view/RichTreeView';
import { useTheme } from '@/context/theme-context' import { useTheme } from '@/context/theme-context'
import React from 'react' import React from 'react'
@@ -108,7 +109,7 @@ function CustomLabel({
</div> </div>
{exists !== undefined && ( {exists !== undefined && (
<span <span
className="text-sm opacity-60 min-w-[40px] text-right text-inherit" className="text-sm opacity-60 min-w-[40px] text-right text-inherit mr-4"
> >
{exists} {exists}
</span> </span>
@@ -181,7 +182,7 @@ export function DownloadFoldersDialog({ currentRow, open, onOpenChange }: Props)
setItemsWithChildren(itemsWithChildren); setItemsWithChildren(itemsWithChildren);
setExpandedItems(itemsWithChildren); setExpandedItems(itemsWithChildren);
const download_folders = data const download_folders = data
.filter(mailbox => currentRow.download_folders.includes(mailbox.name)) .filter(mailbox => currentRow.download_folders?.includes(mailbox.name))
.map(mailbox => mailbox.id.toString()); .map(mailbox => mailbox.id.toString());
setSelectedItems(download_folders); setSelectedItems(download_folders);
}; };
@@ -404,7 +405,7 @@ export function DownloadFoldersDialog({ currentRow, open, onOpenChange }: Props)
return ( return (
<Dialog open={open} onOpenChange={onOpenChange}> <Dialog open={open} onOpenChange={onOpenChange}>
<DialogContent className="sm:max-w-3xl max-h-[90vh] flex flex-col"> <DialogContent className="max-w-[95vw] sm:max-w-3xl max-h-[90vh] flex flex-col overflow-hidden">
<DialogHeader className="flex-shrink-0"> <DialogHeader className="flex-shrink-0">
<DialogTitle>{t('accounts.selectMailboxes')}</DialogTitle> <DialogTitle>{t('accounts.selectMailboxes')}</DialogTitle>
<DialogDescription> <DialogDescription>
@@ -412,8 +413,8 @@ export function DownloadFoldersDialog({ currentRow, open, onOpenChange }: Props)
</DialogDescription> </DialogDescription>
</DialogHeader> </DialogHeader>
<div className="space-y-4"> <div className="flex-1 min-h-0 grid gap-4" style={{ gridTemplateRows: 'auto 1fr' }}>
<div className="flex flex-col pt-2 gap-2"> <div className="flex flex-col gap-2">
<div className="flex gap-2 flex-wrap"> <div className="flex gap-2 flex-wrap">
<Button <Button
variant="outline" variant="outline"
@@ -479,7 +480,7 @@ export function DownloadFoldersDialog({ currentRow, open, onOpenChange }: Props)
</div> </div>
</div> </div>
<ScrollArea className="h-[32rem] flex-1 min-h-0 w-full pr-4 -mr-4 py-1"> <ScrollArea className="min-h-0 w-full py-1">
{isLoading && ( {isLoading && (
<div className="p-8 space-y-8"> <div className="p-8 space-y-8">
<div className="flex flex-col items-center gap-3 text-muted-foreground"> <div className="flex flex-col items-center gap-3 text-muted-foreground">
+63 -53
View File
@@ -17,9 +17,11 @@ import { Mail, Users, Inbox, Zap, Paperclip } from 'lucide-react';
import { formatBytes, formatNumber } from '@/lib/utils'; import { formatBytes, formatNumber } from '@/lib/utils';
import { useQuery } from '@tanstack/react-query'; import { useQuery } from '@tanstack/react-query';
import { get_dashboard_stats, INITIAL_DASHBOARD_STATS, TimeBucket } from '@/api/system/api'; import { get_dashboard_stats, INITIAL_DASHBOARD_STATS, TimeBucket } from '@/api/system/api';
//import { MOCK_DASHBOARD_STATS } from '@/api/system/mock-dashboard';
import { Main } from '@/components/layout/main'; import { Main } from '@/components/layout/main';
import { FixedHeader } from '@/components/layout/fixed-header'; import { FixedHeader } from '@/components/layout/fixed-header';
import { useTranslation } from 'react-i18next'; import { useTranslation } from 'react-i18next';
import LongText from '@/components/long-text';
import { getToken } from '@/stores/authStore'; import { getToken } from '@/stores/authStore';
import { useNavigate } from '@tanstack/react-router'; import { useNavigate } from '@tanstack/react-router';
import useMinimalAccountList from '@/hooks/use-minimal-account-list'; import useMinimalAccountList from '@/hooks/use-minimal-account-list';
@@ -346,18 +348,18 @@ export default function MailArchiveDashboard() {
</CardContent> </CardContent>
</Card> </Card>
</div> </div>
<div className="grid gap-6 grid-cols-1 md:grid-cols-2 lg:grid-cols-4"> <div className="grid gap-6 grid-cols-1 md:grid-cols-2 lg:grid-cols-3 xl:grid-cols-4">
<Card> <Card className="overflow-hidden">
<CardHeader className="!px-4 !pt-4 !pb-1"> <CardHeader className="!px-4 !pt-4 !pb-1">
<CardTitle className="text-xs font-bold uppercase tracking-wider">{t('dashboard.top10Senders')}</CardTitle> <CardTitle className="text-xs font-bold uppercase tracking-wider">{t('dashboard.top10Senders')}</CardTitle>
</CardHeader> </CardHeader>
<CardContent className="p-0"> <CardContent className="p-0">
{hasTopSenders ? ( {hasTopSenders ? (
<Table> <Table className="table-fixed">
<TableHeader> <TableHeader>
<TableRow> <TableRow>
<TableHead className="text-xs">{t('dashboard.sender')}</TableHead> <TableHead className="text-xs">{t('dashboard.sender')}</TableHead>
<TableHead className="text-right text-xs">{t('dashboard.count')}</TableHead> <TableHead className="w-[96px] text-right text-xs">{t('dashboard.count')}</TableHead>
</TableRow> </TableRow>
</TableHeader> </TableHeader>
<TableBody> <TableBody>
@@ -368,16 +370,18 @@ export default function MailArchiveDashboard() {
<div className="absolute left-0 top-0 bottom-0 w-[2px] bg-primary opacity-0 group-hover:opacity-100 transition-opacity" /> <div className="absolute left-0 top-0 bottom-0 w-[2px] bg-primary opacity-0 group-hover:opacity-100 transition-opacity" />
<div className="text-xs flex flex-wrap gap-x-1 min-w-0 flex-1"> <div className="text-xs flex flex-wrap gap-x-1 min-w-0 flex-1">
<span className="flex items-center"> <span className="flex items-center">
<button <LongText className="max-w-[180px] md:max-w-[160px] lg:max-w-[200px] xl:max-w-[220px]">
type="button" <button
onClick={(e) => { type="button"
e.stopPropagation(); onClick={(e) => {
handleQuickSearch({ from: s.key }) e.stopPropagation();
}} handleQuickSearch({ from: s.key })
className="hover:text-primary hover:underline transition-colors truncate max-w-[258px]" }}
> className="hover:text-primary hover:underline transition-colors"
{s.key} >
</button> {s.key}
</button>
</LongText>
</span> </span>
</div> </div>
</div> </div>
@@ -392,17 +396,17 @@ export default function MailArchiveDashboard() {
)} )}
</CardContent> </CardContent>
</Card> </Card>
<Card> <Card className="overflow-hidden">
<CardHeader className="!px-4 !pt-4 !pb-1"> <CardHeader className="!px-4 !pt-4 !pb-1">
<CardTitle className="text-xs font-bold uppercase tracking-wider">{t('dashboard.top10LargestEmails')}</CardTitle> <CardTitle className="text-xs font-bold uppercase tracking-wider">{t('dashboard.top10LargestEmails')}</CardTitle>
</CardHeader> </CardHeader>
<CardContent className="p-0"> <CardContent className="p-0">
{hasTopEmails ? ( {hasTopEmails ? (
<Table> <Table className="table-fixed">
<TableHeader> <TableHeader>
<TableRow> <TableRow>
<TableHead className="text-xs">{t('dashboard.subject')}</TableHead> <TableHead className="text-xs">{t('dashboard.subject')}</TableHead>
<TableHead className="text-right text-xs">{t('dashboard.size')}</TableHead> <TableHead className="w-[96px] text-right text-xs">{t('dashboard.size')}</TableHead>
</TableRow> </TableRow>
</TableHeader> </TableHeader>
<TableBody> <TableBody>
@@ -413,16 +417,18 @@ export default function MailArchiveDashboard() {
<div className="absolute left-0 top-0 bottom-0 w-[2px] bg-primary opacity-0 group-hover:opacity-100 transition-opacity" /> <div className="absolute left-0 top-0 bottom-0 w-[2px] bg-primary opacity-0 group-hover:opacity-100 transition-opacity" />
<div className="text-xs flex flex-wrap gap-x-1 min-w-0 flex-1"> <div className="text-xs flex flex-wrap gap-x-1 min-w-0 flex-1">
<span className="flex items-center"> <span className="flex items-center">
<button <LongText className="max-w-[180px] md:max-w-[160px] lg:max-w-[200px] xl:max-w-[220px]">
type="button" <button
onClick={(e) => { type="button"
e.stopPropagation(); onClick={(e) => {
handleQuickSearch({ id: m.id }) e.stopPropagation();
}} handleQuickSearch({ id: m.id })
className="hover:text-primary hover:underline transition-colors truncate max-w-[258px]" }}
> className="hover:text-primary hover:underline transition-colors"
{m.subject || t('dashboard.noSubject')} >
</button> {m.subject || t('dashboard.noSubject')}
</button>
</LongText>
</span> </span>
</div> </div>
</div> </div>
@@ -437,7 +443,7 @@ export default function MailArchiveDashboard() {
)} )}
</CardContent> </CardContent>
</Card> </Card>
<Card> <Card className="overflow-hidden">
<CardHeader className="!px-4 !pt-4 !pb-1"> <CardHeader className="!px-4 !pt-4 !pb-1">
<CardTitle className="text-xs font-bold uppercase tracking-wider"> <CardTitle className="text-xs font-bold uppercase tracking-wider">
{t('dashboard.top10LargestAttachments')} {t('dashboard.top10LargestAttachments')}
@@ -445,11 +451,11 @@ export default function MailArchiveDashboard() {
</CardHeader> </CardHeader>
<CardContent className="p-0"> <CardContent className="p-0">
{stats?.top_largest_attachments?.length ? ( {stats?.top_largest_attachments?.length ? (
<Table> <Table className="table-fixed">
<TableHeader> <TableHeader>
<TableRow> <TableRow>
<TableHead className="text-xs">{t('attachment.name')}</TableHead> <TableHead className="text-xs">{t('attachment.name')}</TableHead>
<TableHead className="text-right text-xs">{t('dashboard.size')}</TableHead> <TableHead className="w-[96px] text-right text-xs">{t('dashboard.size')}</TableHead>
</TableRow> </TableRow>
</TableHeader> </TableHeader>
<TableBody> <TableBody>
@@ -463,16 +469,18 @@ export default function MailArchiveDashboard() {
<div className="absolute left-0 top-0 bottom-0 w-[2px] bg-primary opacity-0 group-hover:opacity-100 transition-opacity" /> <div className="absolute left-0 top-0 bottom-0 w-[2px] bg-primary opacity-0 group-hover:opacity-100 transition-opacity" />
<div className="text-xs flex flex-wrap gap-x-1 min-w-0 flex-1"> <div className="text-xs flex flex-wrap gap-x-1 min-w-0 flex-1">
<span className="flex items-center"> <span className="flex items-center">
<button <LongText className="max-w-[160px] md:max-w-[140px] lg:max-w-[180px] xl:max-w-[200px]">
type="button" <button
onClick={(e) => { type="button"
e.stopPropagation(); onClick={(e) => {
handleQuickAttachmentSearch({ id: a.id }) e.stopPropagation();
}} handleQuickAttachmentSearch({ id: a.id })
className="hover:text-primary hover:underline transition-colors truncate max-w-[238px]" }}
> className="hover:text-primary hover:underline transition-colors"
{a.name || 'Unnamed'} >
</button> {a.name || 'Unnamed'}
</button>
</LongText>
</span> </span>
</div> </div>
</div> </div>
@@ -489,17 +497,17 @@ export default function MailArchiveDashboard() {
)} )}
</CardContent> </CardContent>
</Card> </Card>
<Card> <Card className="overflow-hidden">
<CardHeader className="!px-4 !pt-4 !pb-1"> <CardHeader className="!px-4 !pt-4 !pb-1">
<CardTitle className="text-xs font-bold uppercase tracking-wider">{t('dashboard.top10Accounts')}</CardTitle> <CardTitle className="text-xs font-bold uppercase tracking-wider">{t('dashboard.top10Accounts')}</CardTitle>
</CardHeader> </CardHeader>
<CardContent className="p-0"> <CardContent className="p-0">
{hasTopAccounts ? ( {hasTopAccounts ? (
<Table> <Table className="table-fixed">
<TableHeader> <TableHeader>
<TableRow> <TableRow>
<TableHead className="text-xs">{t('dashboard.account')}</TableHead> <TableHead className="text-xs">{t('dashboard.account')}</TableHead>
<TableHead className="text-right text-xs">{t('dashboard.emails')}</TableHead> <TableHead className="w-[96px] text-right text-xs">{t('dashboard.emails')}</TableHead>
</TableRow> </TableRow>
</TableHeader> </TableHeader>
<TableBody> <TableBody>
@@ -510,16 +518,18 @@ export default function MailArchiveDashboard() {
<div className="absolute left-0 top-0 bottom-0 w-[2px] bg-primary opacity-0 group-hover:opacity-100 transition-opacity" /> <div className="absolute left-0 top-0 bottom-0 w-[2px] bg-primary opacity-0 group-hover:opacity-100 transition-opacity" />
<div className="text-xs flex flex-wrap gap-x-1 min-w-0 flex-1"> <div className="text-xs flex flex-wrap gap-x-1 min-w-0 flex-1">
<span className="flex items-center"> <span className="flex items-center">
<button <LongText className="max-w-[180px] md:max-w-[160px] lg:max-w-[200px] xl:max-w-[220px]">
type="button" <button
onClick={(e) => { type="button"
e.stopPropagation(); onClick={(e) => {
handleQuickSearch({ account_ids: [getAccountIdByEmail(acc.key) || 0] }) e.stopPropagation();
}} handleQuickSearch({ account_ids: [getAccountIdByEmail(acc.key) || 0] })
className="hover:text-primary hover:underline transition-colors truncate max-w-[258px]" }}
> className="hover:text-primary hover:underline transition-colors"
{acc.key} >
</button> {acc.key}
</button>
</LongText>
</span> </span>
</div> </div>
</div> </div>