From 951901ac0bf949188492eb1c821cfa1506fcf292 Mon Sep 17 00:00:00 2001 From: rustmailer Date: Tue, 14 Jul 2026 05:49:54 +0800 Subject: [PATCH] update --- Cargo.lock | 10 +- Cargo.toml | 2 +- crates/admin/src/main.rs | 10 +- crates/admin/src/migrate.rs | 667 ------------------------------ crates/admin/src/migrate_store.rs | 560 ------------------------- crates/admin/src/migrate_v037.rs | 40 +- crates/admin/src/migrate_v1.rs | 44 +- 7 files changed, 76 insertions(+), 1257 deletions(-) delete mode 100644 crates/admin/src/migrate.rs delete mode 100644 crates/admin/src/migrate_store.rs diff --git a/Cargo.lock b/Cargo.lock index ceb1c61..54c3756 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -299,7 +299,7 @@ checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" [[package]] name = "bichon-admin" -version = "1.6.2" +version = "2.0.0" dependencies = [ "bichon-blob", "bichon-core", @@ -346,7 +346,7 @@ dependencies = [ [[package]] name = "bichon-cli" -version = "1.6.2" +version = "2.0.0" dependencies = [ "base64 0.22.1", "bichon-core", @@ -368,7 +368,7 @@ dependencies = [ [[package]] name = "bichon-core" -version = "1.6.2" +version = "2.0.0" dependencies = [ "async-imap", "base64 0.22.1", @@ -444,7 +444,7 @@ dependencies = [ [[package]] name = "bichon-server" -version = "1.6.2" +version = "2.0.0" dependencies = [ "bichon-core", "bichon-smtp", @@ -469,7 +469,7 @@ dependencies = [ [[package]] name = "bichon-smtp" -version = "1.6.2" +version = "2.0.0" dependencies = [ "base64 0.22.1", "bichon-core", diff --git a/Cargo.toml b/Cargo.toml index ab0916b..dd41153 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -13,7 +13,7 @@ members = [ resolver = "2" [workspace.package] -version = "1.6.2" +version = "2.0.0" edition = "2021" [workspace.dependencies] diff --git a/crates/admin/src/main.rs b/crates/admin/src/main.rs index 0fa42cd..526e12f 100644 --- a/crates/admin/src/main.rs +++ b/crates/admin/src/main.rs @@ -19,12 +19,10 @@ use console::style; use dialoguer::{theme::ColorfulTheme, Select}; -use crate::{migrate::handle_migration, migrate_v037::handle_migration_v037, migrate_v1::handle_migrate_v1, reset::handle_reset_password}; +use crate::{migrate_v037::handle_migration_v037, migrate_v1::handle_migrate_v1, reset::handle_reset_password}; pub mod legacy; pub mod meta; -pub mod migrate; -pub mod migrate_store; pub mod migrate_store_v2; pub mod migrate_v037; pub mod migrate_v1; @@ -45,7 +43,6 @@ async fn run_interactive() { let main_options = vec![ "Reset Admin Password", - "Migrate Legacy v0.3.7 Storage to v1.x (Fjall)", "Migrate Legacy v0.3.7 Storage to v2.x (bichon-blob)", "Migrate v1.x Storage to v2.x (Fjall → bichon-blob)", "Exit", @@ -60,9 +57,8 @@ async fn run_interactive() { match selection { 0 => handle_reset_password(&theme), - 1 => handle_migration(&theme), - 2 => handle_migration_v037(&theme), - 3 => handle_migrate_v1(&theme), + 1 => handle_migration_v037(&theme), + 2 => handle_migrate_v1(&theme), _ => { println!("{}", style("Exiting...").dim()); } diff --git a/crates/admin/src/migrate.rs b/crates/admin/src/migrate.rs deleted file mode 100644 index 2db47c4..0000000 --- a/crates/admin/src/migrate.rs +++ /dev/null @@ -1,667 +0,0 @@ -use std::{collections::HashMap, path::PathBuf}; - -use bichon_core::{ - error::{code::ErrorCode, BichonResult}, - migrate::is_tantivy_index_dir, - raise_error, -}; -use console::style; -use dialoguer::{theme::ColorfulTheme, Confirm, Input}; -use indicatif::{ProgressBar, ProgressStyle}; -use tantivy::{ - collector::TopDocs, - columnar::Column, - query::TermQuery, - schema::{IndexRecordOption, Value}, - DocAddress, Index, TantivyDocument, Term, -}; - -use crate::legacy::schema::SchemaTools; -use crate::migrate_store::{LegacyDirs, NewDirs, NewIndexWriter}; - -pub fn handle_migration(theme: &ColorfulTheme) { - println!( - "\n{}", - style("MIGRATION: Bichon v0.3.7 Storage Architecture → v1.x") - .bold() - .yellow() - ); - - println!( - "{}", - style( - "This tool migrates data from the legacy v0.3.7 Tantivy-based storage \ - architecture to the new v1.x \ - separated index and blob-backed storage format." - ) - .dim() - ); - - println!( - "{}", - style( - "Legacy v0.3.7 architecture:\n\ - • envelope metadata stored in Tantivy\n\ - • message data stored in Tantivy\n\n\ - New v1.x architecture:\n\ - • mail indexes stored in Tantivy\n\ - • attachment indexes stored in Tantivy\n\ - • raw message data stored in blob engine\n\ - • attachment blobs stored in blob engine" - ) - .dim() - ); - - println!( - "\n{} {}", - style("IMPORTANT:").yellow().bold(), - style( - "The paths below must exactly match what your old bichon server was configured with." - ) - .yellow() - ); - - // --- bichon-root-dir --- - let root_dir_str: String = Input::with_theme(theme) - .with_prompt("Enter --bichon-root-dir (same value used by the old server)") - .validate_with(|input: &String| -> Result<(), &str> { - let path = PathBuf::from(input); - if !path.is_absolute() { - return Err("Path must be absolute."); - } - if !path.exists() { - return Err("Directory does not exist."); - } - Ok(()) - }) - .interact_text() - .unwrap(); - - let root_path = PathBuf::from(&root_dir_str); - - // --- bichon-index-dir --- - let default_index = root_path.join("envelope"); - let default_new_index = root_path.join("bichon-indices"); - let index_dir_str: String = Input::with_theme(theme) - .with_prompt(format!( - "Enter --bichon-index-dir (leave blank to use default: {})", - style(default_index.display()).cyan() - )) - .allow_empty(true) - .validate_with(|input: &String| -> Result<(), &str> { - if input.is_empty() { - return Ok(()); - } - let path = PathBuf::from(input); - if !path.is_absolute() { - return Err("Path must be absolute."); - } - - if !path.exists() { - return Err("Directory does not exist."); - } - Ok(()) - }) - .interact_text() - .unwrap(); - - let index_path = if index_dir_str.is_empty() { - default_index - } else { - PathBuf::from(&index_dir_str) - }; - - let new_index_path = if index_dir_str.is_empty() { - default_new_index - } else { - PathBuf::from(&index_dir_str).join("bichon-indices") - }; - - // --- bichon-data-dir --- - let default_data = root_path.join("eml"); - let default_new_data = root_path.join("bichon-storage"); - let data_dir_str: String = Input::with_theme(theme) - .with_prompt(format!( - "Enter --bichon-data-dir (leave blank to use default: {})", - style(default_data.display()).cyan() - )) - .allow_empty(true) - .validate_with(|input: &String| -> Result<(), &str> { - if input.is_empty() { - return Ok(()); - } - let path = PathBuf::from(input); - if !path.is_absolute() { - return Err("Path must be absolute."); - } - if !path.exists() { - return Err("Directory does not exist."); - } - Ok(()) - }) - .interact_text() - .unwrap(); - - let data_path = if data_dir_str.is_empty() { - default_data - } else { - PathBuf::from(&data_dir_str) - }; - - let new_data_path = if data_dir_str.is_empty() { - default_new_data - } else { - PathBuf::from(&data_dir_str).join("bichon-storage") - }; - - println!("\n{}", style("Paths to be migrated:").bold()); - println!("----------------------------------------"); - println!( - "{:<20} : {}", - "bichon-root-dir", - style(root_path.display()).cyan() - ); - println!( - "{:<20} : {}", - "bichon-index-dir", - style(index_path.display()).cyan() - ); - println!( - "{:<20} : {}", - "bichon-data-dir", - style(data_path.display()).cyan() - ); - println!("----------------------------------------"); - - println!( - "\n{} Checking legacy v0.3.7 storage layout...", - style("⌛").yellow() - ); - - match is_legacy_data_layout_with_paths(&index_path, &data_path) { - Ok(true) => { - println!( - "{} {}", - style("✔").green(), - style("Legacy v0.3.7 Tantivy-based storage detected. Migration to v1.x is required.") - .yellow() - ); - } - Ok(false) => { - println!( - "{} {}", - style("✔").green(), - style("No legacy v0.3.7 storage layout was detected at the specified paths.").green() - ); - - println!( - "{}", - style( - "The selected directories may already be using the v1.x storage architecture." - ) - .dim() - ); - - return; - } - Err(e) => { - eprintln!( - "{} Failed to verify legacy storage layout: {:?}", - style("ERROR:").red().bold(), - e - ); - - std::process::exit(1); - } - } - - println!( - "\n{} {}", - style("⚠").yellow(), - style( - "This migration is non-destructive. Existing v0.x storage files will remain unchanged." - ) - .yellow() - ); - - if !Confirm::with_theme(theme) - .with_prompt("Ready to migrate?") - .default(true) - .interact() - .unwrap() - { - println!("{}", style("Migration cancelled.").dim()); - return; - } - - // Step 1: Migrate metadata (meta.db + mailbox.db → memdb) - match crate::meta::migrate_metadata(&root_path) { - Ok(()) => {} - Err(e) => { - eprintln!( - "\n{} Metadata migration failed:\n{}", - style("✘").red().bold(), - style(e).red() - ); - eprintln!( - "{}", - style("Aborting migration. No changes have been made to Tantivy data.").yellow() - ); - return; - } - } - - println!( - "\n{} {}", - style("⌛").yellow(), - style("Step 2: Migrating email index and blob data...").cyan() - ); - - println!( - "\n{} {}", - style("ℹ").blue(), - style("Batch size controls memory usage during migration:").dim() - ); - println!( - " {} 1000 — ~500MB RAM (slower, low memory)", - style("•").dim() - ); - println!(" {} 3000 — ~1GB RAM (recommended)", style("•").dim()); - println!( - " {} 5000 — ~2GB RAM (faster, high memory)", - style("•").dim() - ); - println!( - " {} Note: actual memory usage depends on your average email size.", - style("•").yellow() - ); - 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::() { - Ok(n) if n > 0 => Ok(()), - _ => Err("Please enter a valid positive number"), - }) - .interact_text() - .unwrap_or("3000".to_string()); - input.trim().parse::().unwrap_or(3000) - }; - - 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; - } - }; - - if total_segments == 0 { - println!( - "{} {}", - style("✔").green(), - style("No EML segments found. Nothing to migrate.").bold() - ); - 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 writer = match NewIndexWriter::open(NewDirs::new( - new_index_path.clone(), - new_data_path.clone(), - )) { - Ok(w) => w, - Err(e) => { - pb.finish_with_message(format!("{}", style("Migration failed.").red())); - eprintln!("\n{} {:?}", style("✘").red().bold(), e); - return; - } - }; - - 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 = 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, - &mut writer, - 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.set_message(style("Finalizing indexes...").dim().to_string()); - if let Err(e) = writer.finish_writers() { - pb.finish_with_message(format!("{}", style("Migration failed.").red())); - eprintln!("\n{} {:?}", style("✘").red().bold(), e); - return; - } - - pb.finish_with_message(format!( - "Migration finished. Total: {}, Skipped: {}", - grand_total_migrated, grand_total_skipped - )); - - println!( - "{} {}", - style("✔").green(), - style("Migration completed successfully!").bold() - ); -} - -pub fn is_legacy_data_layout_with_paths( - envelope_dir: &PathBuf, - eml_dir: &PathBuf, -) -> std::io::Result { - let envelope_result = is_tantivy_index_dir(envelope_dir)?; - let eml_result = is_tantivy_index_dir(eml_dir)?; - - Ok(envelope_result || eml_result) -} - -/// 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 { - 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()) -} - -/// Migrate all documents from a single EML segment to the new storage layout. -/// -/// 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( - batch_size: u32, - legacy: LegacyDirs, - writer: &mut NewIndexWriter, - segment_index: usize, - mut on_progress: F, -) -> BichonResult<()> -where - F: FnMut(&str), -{ - // ── open legacy indices ──────────────────────────────────────────── - let envelope_index = Index::open_in_dir(&legacy.envelope_dir) - .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; - let eml_index = Index::open_in_dir(&legacy.eml_dir) - .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; - - let envelope_reader = envelope_index - .reader() - .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; - let eml_reader = eml_index - .reader() - .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; - - let envelope_searcher = envelope_reader.searcher(); - let eml_searcher = eml_reader.searcher(); - - let ef = SchemaTools::envelope_fields(); - let mf = SchemaTools::eml_fields(); - - 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 num_docs = eml_segment.num_docs(); - if num_docs == 0 { - on_progress("TOTAL:0"); - on_progress("DONE:0:0"); - return Ok(()); - } - - on_progress(&format!("TOTAL:{}", num_docs)); - - let max_doc = eml_segment.max_doc(); - let ff = eml_segment.fast_fields(); - let f_id_col: Column = ff.u64("id").map_err(|e| { - raise_error!( - 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 = 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))?; - - if let Some((_, addr)) = hits.first() { - let env_doc: TantivyDocument = envelope_searcher - .doc(*addr) - .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; - let uid = env_doc - .get_first(ef.f_uid) - .and_then(|v| v.as_u64()) - .unwrap_or(0) as u32; - let internal_date = env_doc - .get_first(ef.f_internal_date) - .and_then(|v| v.as_i64()) - .unwrap_or(0); - envelope_map.insert(eid, (uid, internal_date)); - env_scanned += 1; - } else { - env_skipped += 1; - } - - if env_scanned % 10 == 0 { - on_progress(&format!( - "PHASE1:{}/{} skipped:{}", - env_scanned, max_doc, env_skipped - )); - } - } - - // Free the envelope index before the heavy EML processing. - drop(envelope_searcher); - drop(envelope_reader); - drop(envelope_index); - - // ── Phase 2: process EML docs, streaming one at a time ───────────── - let mut total_migrated = 0usize; - let mut total_skipped = 0usize; - - 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 => { - on_progress(&format!("WARN: eid {} envelope not found", eid)); - total_skipped += 1; - continue; - } - }; - - let eml_doc: TantivyDocument = store_reader - .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; - continue; - } - - total_migrated += 1; - - if total_migrated % 10 == 0 || total_migrated as u32 == num_docs { - on_progress(&format!("PROGRESS:{}:{}", total_migrated, num_docs)); - } - } - - drop(store_reader); - - // Flush blob buffers via ingestion API — bypasses memtable/WAL. - writer.flush_fjall_buffers()?; - - chunk_start = chunk_end; - } - - on_progress(&format!("DONE:{}:{}", total_migrated, total_skipped)); - Ok(()) -} diff --git a/crates/admin/src/migrate_store.rs b/crates/admin/src/migrate_store.rs deleted file mode 100644 index 792e674..0000000 --- a/crates/admin/src/migrate_store.rs +++ /dev/null @@ -1,560 +0,0 @@ -use std::{path::PathBuf, time::Instant}; - -use bytes::Bytes; -use mail_parser::MimeHeaders; - -use bichon_core::{ - envelope::extractor::extract_references, message::content::AttachmentInfo, - store::tantivy::tokenizers::EuroTokenizer, utils::compute_content_hash, -}; - -use fjall::{ - config::{BlockSizePolicy, CompressionPolicy}, - CompressionType, Database, Keyspace, KeyspaceCreateOptions, KvSeparationOptions, -}; -use mail_parser::MessageParser; -use tantivy::{indexer::NoMergePolicy, Index, IndexWriter, TantivyDocument}; -use uuid::Uuid; - -use bichon_core::{ - common::AddrVec, - envelope::extractor::{compute_thread_id, generate_message_id}, - error::{code::ErrorCode, BichonResult}, - raise_error, - store::envelope::Envelope, - store::tantivy::{ - model::{AttachmentModel, EnvelopeWithAttachments}, - schema::SchemaTools, - }, - utc_now, -}; - -pub struct LegacyDirs { - pub envelope_dir: PathBuf, - pub eml_dir: PathBuf, -} - -pub struct NewDirs { - pub envelope_dir: PathBuf, - pub attachment_dir: PathBuf, - pub storage_dir: PathBuf, -} - -impl LegacyDirs { - pub fn new(index: PathBuf, data: PathBuf) -> Self { - Self { - envelope_dir: index, - eml_dir: data, - } - } -} - -impl NewDirs { - pub fn new(index: PathBuf, data: PathBuf) -> Self { - Self { - envelope_dir: index.join("mail_metadata"), - attachment_dir: index.join("attachment_metadata"), - storage_dir: data, - } - } -} - -pub struct DetachOutput { - pub infos: Vec, - pub blobs: Vec<(String, Bytes)>, -} - -pub fn detach_attachments_standalone( - original_body: &[u8], - message: &mail_parser::Message<'_>, -) -> (Vec, DetachOutput) { - let mut stripped_eml = original_body.to_vec(); - let mut infos = Vec::new(); - let mut blobs = Vec::new(); - - let mut ranges: Vec<_> = message - .attachments() - .map(|att| { - ( - att.raw_body_offset() as usize, - att.raw_end_offset() as usize, - att, - ) - }) - .collect(); - ranges.sort_by(|a, b| b.0.cmp(&a.0)); - - for (raw_start, raw_end, att) in ranges { - let content_hash = compute_content_hash(att.contents()); - let body_len = original_body.len(); - let raw_start = raw_start.min(body_len); - let raw_end = raw_end.min(body_len); - let range_valid = raw_start < raw_end; - - if range_valid { - blobs.push(( - content_hash.clone(), - Bytes::copy_from_slice(&original_body[raw_start..raw_end]), - )); - } - - if range_valid { - let placeholder = format!("<>", &content_hash); - stripped_eml.splice(raw_start..raw_end, placeholder.as_bytes().iter().cloned()); - } - - infos.push(AttachmentInfo { - filename: att.attachment_name().map(|n| n.to_string()), - size: att.contents().len(), - inline: att - .content_disposition() - .map(|d| d.is_inline()) - .unwrap_or_else(|| att.content_id().is_some()), - file_type: att - .content_type() - .map(|ct| { - format!( - "{}/{}", - ct.c_type.as_ref(), - ct.c_subtype.as_deref().unwrap_or("") - ) - }) - .unwrap_or_else(|| "application/octet-stream".to_string()), - content_id: att.content_id().map(|id| id.to_string()), - content_hash, - is_message: att.is_message(), - extracted_text: None, - extracted_page_count: None, - extracted_is_ocr: false, - }); - } - - (stripped_eml, DetachOutput { infos, blobs }) -} - -pub struct NewIndexWriter { - pub envelope_writer: Option, - pub attachment_writer: Option, - pub email_ks: Keyspace, - pub attachment_ks: Keyspace, - pending: usize, - email_buf: Vec<(String, Vec)>, - attachment_buf: Vec<(String, Vec)>, -} - -//const COMMIT_THRESHOLD: usize = 500; - -impl NewIndexWriter { - pub fn open(dirs: NewDirs) -> BichonResult { - // ── envelope index ────────────────────────────────────────────── - std::fs::create_dir_all(&dirs.envelope_dir) - .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; - let envelope_index = if dirs - .envelope_dir - .read_dir() - .map(|mut d| d.next().is_none()) - .unwrap_or(true) - { - Index::create_in_dir(&dirs.envelope_dir, SchemaTools::email_schema()) - .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))? - } else { - Index::open_in_dir(&dirs.envelope_dir) - .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))? - }; - - envelope_index - .tokenizers() - .register("euro", EuroTokenizer::new()); - - let envelope_writer = envelope_index - .writer_with_num_threads(3, 256 * 1024 * 1024) - .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 ───────────────────────────────────────────── - std::fs::create_dir_all(&dirs.attachment_dir) - .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; - let attachment_index = if dirs - .attachment_dir - .read_dir() - .map(|mut d| d.next().is_none()) - .unwrap_or(true) - { - Index::create_in_dir(&dirs.attachment_dir, SchemaTools::attachment_schema()) - .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))? - } else { - Index::open_in_dir(&dirs.attachment_dir) - .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))? - }; - - attachment_index - .tokenizers() - .register("euro", EuroTokenizer::new()); - let attachment_writer = attachment_index - .writer_with_num_threads(3, 256 * 1024 * 1024) - .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 ─────────────────────────────────────────────────── - std::fs::create_dir_all(&dirs.storage_dir) - .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; - let db = Database::builder(&dirs.storage_dir) - .cache_size(8 * 1024 * 1024) - .journal_compression(CompressionType::None) - .max_journaling_size(64 * 1024 * 1024) - .open() - .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; - - let email_ks = db - .keyspace("email", || { - KeyspaceCreateOptions::default() - .max_memtable_size(4 * 1024 * 1024) - .data_block_size_policy(BlockSizePolicy::all(4 * 1024)) - .data_block_compression_policy(CompressionPolicy::all(CompressionType::Lz4)) - .with_kv_separation(Some( - KvSeparationOptions::default() - .separation_threshold(1024) - .compression(CompressionType::Lz4) - .file_target_size(512 * 1024 * 1024), - )) - }) - .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; - - let attachment_ks = db - .keyspace("attachments", || { - KeyspaceCreateOptions::default() - .max_memtable_size(4 * 1024 * 1024) - .data_block_size_policy(BlockSizePolicy::all(4 * 1024)) - .data_block_compression_policy(CompressionPolicy::all(CompressionType::Lz4)) - .with_kv_separation(Some( - KvSeparationOptions::default() - .separation_threshold(1024) - .compression(CompressionType::Lz4) - .file_target_size(512 * 1024 * 1024), - )) - }) - .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; - - Ok(Self { - envelope_writer: Some(envelope_writer), - attachment_writer: Some(attachment_writer), - email_ks, - attachment_ks, - pending: 0, - email_buf: Vec::new(), - attachment_buf: Vec::new(), - }) - } - - pub fn ingest( - &mut self, - eml_bytes: &[u8], - account_id: u64, - mailbox_id: u64, - uid: u32, - internal_date: i64, - ) -> BichonResult<()> { - let email_content_hash = compute_content_hash(eml_bytes); - - let message = MessageParser::new() - .parse(eml_bytes) - .ok_or_else(|| raise_error!("failed to parse eml".into(), ErrorCode::InternalError))?; - - if message.parts.is_empty() { - return Err(raise_error!( - "Malformed or completely empty EML (no parts found)".into(), - ErrorCode::InternalError - )); - } - // ── text / preview ──────────────────────────────────────────────── - let text = message - .body_text(0) - .map(|c| c.into_owned()) - .or_else(|| { - message - .body_html(0) - .map(|html| bichon_core::utils::html::extract_text(html.into_owned())) - }) - .unwrap_or_default(); - let text = text.split_whitespace().collect::>().join(" "); - let preview = if text.chars().count() > 100 { - text.chars().take(100).collect::() + "..." - } else { - text.clone() - }; - - // ── headers ─────────────────────────────────────────────────────── - let message_id = message - .message_id() - .map(String::from) - .unwrap_or_else(generate_message_id); - - let in_reply_to = message.in_reply_to().as_text().map(String::from); - let references = extract_references(&message); - let thread_id = compute_thread_id(in_reply_to, references, &message_id); - - let subject = message.subject().map(String::from).unwrap_or_default(); - let date = message.date().map(|d| d.to_timestamp() * 1000).unwrap_or(0); - let internal_date = if internal_date == 0 { - date - } else { - internal_date - }; - - let parse_addrs = |addrs: Option<&mail_parser::Address<'_>>| { - addrs - .map(|addr| { - AddrVec::from(addr) - .0 - .into_iter() - .filter_map(|a| a.address) - .collect::>() - }) - .unwrap_or_default() - }; - - let from = message - .from() - .and_then(|addr| AddrVec::from(addr).0.into_iter().next()) - .and_then(|a| a.address) - .unwrap_or_else(|| "unknown".to_string()); - let to = parse_addrs(message.to()); - let cc = parse_addrs(message.cc()); - let bcc = parse_addrs(message.bcc()); - - // ── detach attachments → blob ────────────────────────────────────── - let (stripped_eml, attachment_output) = detach_attachments_standalone(eml_bytes, &message); - - // Buffer for bulk ingestion — sorted + flushed later. - self.email_buf - .push((email_content_hash.clone(), stripped_eml)); - for (hash, data) in &attachment_output.blobs { - self.attachment_buf.push((hash.clone(), data.to_vec())); - } - - // ── build envelope doc ──────────────────────────────────────────── - let envelope_id = Uuid::new_v4().to_string(); - let now = utc_now!(); - - let attachment_docs: Vec = attachment_output - .infos - .iter() - .filter(|a| !a.inline || a.content_id.is_none()) - .map(|a| { - AttachmentModel { - id: Uuid::new_v4().to_string(), - envelope_id: envelope_id.clone(), - account_id, - account_email: None, - mailbox_id, - mailbox_name: None, - subject: subject.clone(), - content_hash: a.content_hash.clone(), - from: from.clone(), - date, - ingest_at: now, - size: a.size as u64, - ext: a.get_extension(), - category: a.get_category().to_string(), - content_type: a.file_type.clone(), - shard_id: 0, - text: None, - has_text: false, - is_ocr: false, - page_count: None, - is_indexed: false, - is_message: a.is_message, - name: a.filename.clone(), - tags: None, - auto_tags: None, - } - .into_document() - }) - .collect(); - - let envelope = Envelope { - id: envelope_id, - message_id, - account_id, - mailbox_id, - uid, - subject, - preview, - from, - to, - cc, - bcc, - date, - internal_date, - ingest_at: now, - size: eml_bytes.len() as u32, - thread_id, - attachment_count: message.attachment_count(), - regular_attachment_count: attachment_docs.len(), - tags: None, - account_email: None, - account_name: None, - mailbox_name: None, - content_hash: email_content_hash, - }; - - let ea = EnvelopeWithAttachments { - envelope, - attachments: Some(attachment_output.infos), - }; - let envelope_doc = ea.to_document(&text, 0)?; - - self.envelope_writer - .as_mut() - .unwrap() - .add_document(envelope_doc) - .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; - - for doc in attachment_docs { - self.attachment_writer - .as_mut() - .unwrap() - .add_document(doc) - .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; - } - - self.pending += 1; - // if self.pending >= COMMIT_THRESHOLD { - // self.commit()?; - // } - - Ok(()) - } - - /// 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 { - return Ok(()); - } - println!("Tantivy committing... this may take 2-3 minutes, please wait."); - let start = Instant::now(); - if let Some(writer) = self.envelope_writer.as_mut() { - writer - .commit() - .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; - } - if let Some(writer) = self.attachment_writer.as_mut() { - writer - .commit() - .map_err(|e| raise_error!(format!("{e:#?}"), ErrorCode::InternalError))?; - } - println!("tantivy commit elapsed: {:#?}", start.elapsed()); - tracing::info!(count = self.pending, "committed tantivy batch"); - self.pending = 0; - 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 seg_ids = writer - .index() - .searchable_segment_ids() - .map_err(|e| raise_error!(format!("{:#?}", e), ErrorCode::InternalError))?; - 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(()) - } -} diff --git a/crates/admin/src/migrate_v037.rs b/crates/admin/src/migrate_v037.rs index cada749..7effc60 100644 --- a/crates/admin/src/migrate_v037.rs +++ b/crates/admin/src/migrate_v037.rs @@ -2,7 +2,7 @@ use std::{collections::HashMap, path::PathBuf}; use bichon_core::{ error::{code::ErrorCode, BichonResult}, - migrate::write_storage_version, + migrate::{is_tantivy_index_dir, write_storage_version}, raise_error, }; use console::style; @@ -17,10 +17,42 @@ use tantivy::{ }; use crate::legacy::schema::SchemaTools; -use crate::migrate::is_legacy_data_layout_with_paths; -use crate::migrate_store::LegacyDirs; use crate::migrate_store_v2::{NewDirs, NewIndexWriterV2}; +pub struct LegacyDirs { + pub envelope_dir: PathBuf, + pub eml_dir: PathBuf, +} + +impl LegacyDirs { + pub fn new(index: PathBuf, data: PathBuf) -> Self { + Self { + envelope_dir: index, + eml_dir: data, + } + } +} + +pub fn is_legacy_data_layout_with_paths( + envelope_dir: &PathBuf, + eml_dir: &PathBuf, +) -> std::io::Result { + let envelope_result = is_tantivy_index_dir(envelope_dir)?; + let eml_result = is_tantivy_index_dir(eml_dir)?; + Ok(envelope_result || eml_result) +} + +/// Return the number of segments in the legacy EML Tantivy index. +pub fn count_eml_segments(legacy: &LegacyDirs) -> BichonResult { + 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 handle_migration_v037(theme: &ColorfulTheme) { println!( "\n{}", @@ -301,7 +333,7 @@ pub fn handle_migration_v037(theme: &ColorfulTheme) { ); let legacy = LegacyDirs::new(index_path.clone(), data_path.clone()); - let total_segments = match crate::migrate::count_eml_segments(&legacy) { + let total_segments = match count_eml_segments(&legacy) { Ok(n) => n, Err(e) => { eprintln!( diff --git a/crates/admin/src/migrate_v1.rs b/crates/admin/src/migrate_v1.rs index 021565f..acc474e 100644 --- a/crates/admin/src/migrate_v1.rs +++ b/crates/admin/src/migrate_v1.rs @@ -96,22 +96,40 @@ pub fn handle_migrate_v1(theme: &ColorfulTheme) { ); let root_dir: String = Input::with_theme(theme) - .with_prompt("Bichon root directory") - .with_initial_text("/var/lib/bichon") - .interact() + .with_prompt("Enter --bichon-root-dir (same value used by the old server)") + .validate_with(|input: &String| -> Result<(), &str> { + let path = PathBuf::from(input); + if !path.is_absolute() { + return Err("Path must be absolute."); + } + if !path.exists() { + return Err("Directory does not exist."); + } + Ok(()) + }) + .interact_text() .unwrap(); let root_dir = PathBuf::from(root_dir.trim()); - let data_dir: String = Input::with_theme(theme) - .with_prompt("Bichon data directory (leave empty to use root directory)") - .with_initial_text("") - .allow_empty(true) - .interact() - .unwrap(); - let data_base = if data_dir.trim().is_empty() { - root_dir.clone() - } else { - PathBuf::from(data_dir.trim()) + let data_base = { + let input: String = Input::with_theme(theme) + .with_prompt("Enter --bichon-data-dir (leave blank to use root directory)") + .allow_empty(true) + .interact_text() + .unwrap(); + if input.trim().is_empty() { + root_dir.clone() + } else { + let path = PathBuf::from(input.trim()); + if !path.exists() { + eprintln!( + "{}", + style(format!("Data directory does not exist: {}", path.display())).red() + ); + return; + } + path + } }; let fjall_path = data_base.join("bichon-storage");