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

update

Browse Source
This commit is contained in:
rustmailer
2026-07-14 05:49:54 +08:00
parent b7e757dbf7
commit 951901ac0b
7 changed files with 76 additions and 1257 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+5 -5
View File
@@ -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",
Cargo.toml
+1 -1
View File
@@ -13,7 +13,7 @@ members = [
resolver = "2"
[workspace.package]
version = "1.6.2"
version = "2.0.0"
edition = "2021"
[workspace.dependencies]
crates/admin/src/main.rs
+3 -7
View File
@@ -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());
}
crates/admin/src/migrate.rs
-667
View File
@@ -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::<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()
);
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<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,
&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<bool> {
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<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())
}
/// 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<F>(
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<u64> = 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<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))?;
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(())
}
crates/admin/src/migrate_store.rs
-560
View File
@@ -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<AttachmentInfo>,
pub blobs: Vec<(String, Bytes)>,
}
pub fn detach_attachments_standalone(
original_body: &[u8],
message: &mail_parser::Message<'_>,
) -> (Vec<u8>, 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!("<<BICHON_DETACH_HASH:{}>>", &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<IndexWriter>,
pub attachment_writer: Option<IndexWriter>,
pub email_ks: Keyspace,
pub attachment_ks: Keyspace,
pending: usize,
email_buf: Vec<(String, Vec<u8>)>,
attachment_buf: Vec<(String, Vec<u8>)>,
}
//const COMMIT_THRESHOLD: usize = 500;
impl NewIndexWriter {
pub fn open(dirs: NewDirs) -> BichonResult<Self> {
// ── 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::<Vec<_>>().join(" ");
let preview = if text.chars().count() > 100 {
text.chars().take(100).collect::<String>() + "..."
} 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::<Vec<_>>()
})
.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<TantivyDocument> = 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(())
}
}
crates/admin/src/migrate_v037.rs
+36 -4
View File
@@ -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<bool> {
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<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 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!(
crates/admin/src/migrate_v1.rs
+31 -13
View File
@@ -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");